Repository navigation
Expand file tree
/
Copy pathget_backtest_data
More file actions
executable file
·148 lines (115 loc) · 4.71 KB
/
Copy pathget_backtest_data
File metadata and controls
executable file
·148 lines (115 loc) · 4.71 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
#!/usr/bin/env python
# -*- coding: utf-8 -*-
from __future__ import print_function, absolute_import, division
import datetime
import json
from contextlib import contextmanager
import click
import six
from quantopian_tools.algorithms import new_algorithm, save_algorithm, delete_algorithm
from quantopian_tools.backtests import run_backtest
from quantopian_tools.securities import lookup_sid
WIDTH = 80
algorithm_template = """
def initialize(context):
context.security = sid({sid})
def before_trading_start(context, data):
info = data.history(context.security, ['price', 'open', 'high', 'low', 'close', 'volume'], 1, '1d')
record(volume=info['volume'][0],
open=info['open'][0],
close=info['close'][0],
high=info['high'][0],
low=info['low'][0])
"""
class TradingDayParamType(click.ParamType):
name = 'date'
def convert(self, value, param, ctx):
if isinstance(value, six.string_types):
value = datetime.datetime.strptime(value, '%m/%d/%Y').date()
return value
def __repr__(self):
return 'STRING'
class DateEncoder(json.JSONEncoder):
def default(self, o):
if isinstance(o, datetime.date):
return o.isoformat()
return json.JSONEncoder.default(self, o)
@contextmanager
def action_status(msg):
click.echo(msg.ljust(WIDTH, '.'), nl=False)
try:
yield
except:
click.secho(' ERROR', fg='red')
raise
click.secho(' SUCCESS', fg='green')
def weekdays_between(start, end):
return sum([1 for daydelta in range(1, (end - start).days + 1)
if (start + datetime.timedelta(daydelta)).weekday() < 5])
def get_security(sid):
with action_status("Looking up security id {}".format(sid)):
security = lookup_sid(sid)
if not security:
raise ValueError("Unable to find a security with sid={}".format(sid))
return security
def create_temp_algorithm(security, start_date, end_date):
title = 'TEMP: {} Backtest Data ({}-{})'.format(security.symbol, start_date, end_date)
with action_status("Creating temporary algorithm"):
algorithm_id = new_algorithm(title)
return {
'id': algorithm_id,
'title': title
}
def delete_temp_algorithm(algrithm):
with action_status("Deleting temp algorithm {}".format(algrithm['id'])):
delete_algorithm(algrithm)
def upload_code_to_algorithm(security, algorithm):
with action_status("Uploading code to algorithm"):
algorithm['code'] = algorithm_template.format(sid=security.sid)
if not save_algorithm(algorithm):
raise RuntimeError("Failed to save algorithm code")
return algorithm
def run_backtest_and_fetch_data(algorithm, start_date, end_date):
data = []
label = "Running backtest from {} to {}".format(start_date, end_date)
with click.progressbar(run_backtest(algorithm,
start_date=start_date,
end_date=end_date,
capital_base=1000000,
data_frequency='minute'),
fill_char='.',
empty_char=' ',
bar_template='%(label)s%(bar)s %(info)s',
label=label,
width=WIDTH-len(label),
info_sep=' ',
show_eta=False,
length=weekdays_between(start_date, end_date)) as results:
for result in results:
if 'daily_performance' not in result:
continue
custom_data = dict(result.daily_performance.custom_data)
custom_data['price'] = custom_data['close']
custom_data['date'] = result['daily_performance']['date']
data.append(custom_data)
click.echo("Retrieved {} data points".format(len(data)))
return data
def save_data(data, output):
with action_status("Saving data"):
json.dump(data, output, sort_keys=True, indent=4, cls=DateEncoder)
@click.command()
@click.argument('sid', type=click.INT)
@click.argument('start_date', type=TradingDayParamType())
@click.argument('end_date', type=TradingDayParamType())
@click.argument('output', type=click.File('w+'))
def main(sid, start_date, end_date, output):
security = get_security(sid)
algorithm = create_temp_algorithm(security, start_date, end_date)
try:
algorithm = upload_code_to_algorithm(security, algorithm)
data = run_backtest_and_fetch_data(algorithm, start_date, end_date)
save_data(data, output)
finally:
delete_temp_algorithm(algorithm)
if __name__ == '__main__':
main()