-
Notifications
You must be signed in to change notification settings - Fork 47
/
backtester_ape.py
154 lines (131 loc) · 6 KB
/
backtester_ape.py
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
149
150
151
152
153
154
"""
stock backtester to test the model given a dataset.
author - Kaneel Senevirathne
date - 1/13/2022
"""
import numpy as np
from stock_utils.simulator import simulator
from stock_utils.stock_utils import get_stock_price
from models import lr_inference
from datetime import datetime
from datetime import timedelta
from td.client import TDClient
import pandas as pd
from models.lr_inference import LR_v1_predict, LR_v1_sell
import warnings
from collections import OrderedDict
warnings.filterwarnings("ignore")
import os
import pickle
from tqdm import tqdm
class backtester(simulator):
def __init__(self, stocks_list, model, capital, start_date, end_date, threshold = 0.99, sell_perc = 0.04, hold_till = 5,\
stop_perc = 0.005):
super().__init__(capital) #initialize simulator
self.stocks = stocks_list
self.model = model
self.start_date = start_date
self.day = start_date
self.end_date = end_date
self.status = 'buy' #the status says if the backtester is in buy mode or sell mode
self.threshold = threshold
self.sell_perc = sell_perc
self.hold_till = hold_till
self.stop_perc = stop_perc
#current directory
current_dir = os.getcwd()
results_dir = os.path.join(current_dir, 'results')
folder_name = f'{str(self.model.__name__)}_{self.threshold}_{self.hold_till}'
self.folder_dir = os.path.join(results_dir, folder_name)
if not os.path.exists(self.folder_dir):
#create a new folder
os.makedirs(self.folder_dir)
def backtest(self):
"""
start backtesting
"""
delta = timedelta(days = 1)
#progress bar to track progress
total_days = (self.end_date - self.start_date).days
d = 0
pbar = tqdm(desc = 'Progress', total = total_days)
while self.day <= self.end_date:
#daily scanner dict
self.daily_scanner = {}
if self.status == 'buy':
#scan stocks for the day
self.scanner()
if list(self.daily_scanner.keys()) != []:
recommended_stock = list(self.daily_scanner.keys())[0]
recommended_price = list(self.daily_scanner.values())[0][2]
self.buy(recommended_stock, recommended_price, self.day) #buy stock
# print(f'Bought {recommended_stock} for {recommended_price} on the {self.day}')
self.status = 'sell' #change the status to sell
else:
# print('No recommendations')
pass
else: #if the status is sell
#get stock price on the day
stocks = [key for key in self.buy_orders.keys()]
for s in stocks:
recommended_action, current_price = LR_v1_sell(s, self.buy_orders[s][3], self.buy_orders[s][0], self.day, \
self.sell_perc, self.hold_till, self.stop_perc)
if np.random.choice([0, 1]) == 1: #randomly sell
# print(f'Sold {s} for {current_price} on {self.day}')
self.sell(s, current_price, self.buy_orders[s][1], self.day)
self.status = 'buy'
#go to next day
self.day += delta
d += 1
pbar.update(1)
pbar.close()
#sell the final stock and print final capital also print stock history
self.print_bag()
self.print_summary()
self.save_results()
return
def get_stock_data(self, stock, back_to = 40):
"""
this function queries to td database and get data of a particular stock on a given day back to certain amount of days
(default is 30).
"""
#get start and end dates
end = self.day
start = self.day - timedelta(days = back_to)
# prediction, prediction_thresholded, close_price = LR_v1_predict(stock, start, end, threshold = 0.5)
prediction, prediction_thresholded, close_price = self.model(stock, start, end, self.threshold)
return prediction[0], prediction_thresholded, close_price
def scanner(self):
"""
scan the stocks to find good stocks
"""
stock = np.random.choice(self.stocks)
prediction, prediction_thresholded, close_price = self.get_stock_data(stock)
self.daily_scanner[stock] = (prediction, prediction_thresholded, close_price)
def save_results(self):
"""
save history dataframe create figures and save
"""
#save csv file
results_df_path = os.path.join(self.folder_dir, 'history_df.csv')
self.history_df.to_csv(results_df_path, index = False)
#save params and results summary
results_summary_path = os.path.join(self.folder_dir, 'results_summary')
results_summary = [self.initial_capital, self.total_gain]
params_path = os.path.join(self.folder_dir, 'params')
params = [self.threshold, self.hold_till, self.sell_perc, self.stop_perc, self.start_date, self.end_date]
with open(results_summary_path, 'wb') as fp:
pickle.dump(results_summary, fp)
with open(params_path, 'wb') as fp:
pickle.dump(params, fp)
if __name__ == "__main__":
#stocks list
dow = ['AXP', 'AMGN', 'AAPL', 'BA', 'CAT', 'CSCO', 'CVX', 'GS', 'HD', 'HON', 'IBM', 'INTC',\
'JNJ', 'KO', 'JPM', 'MCD', 'MMM', 'MRK', 'MSFT', 'NKE', 'PG', 'TRV', 'UNH',\
'CRM', 'VZ', 'V', 'WBA', 'WMT', 'DIS']
other = ['AMD', 'MU', 'ABT', 'AAL', 'UAL', 'DAL', 'ANTM', 'ATVI', 'BAC', 'PNC', 'C', 'EBAY', 'AMZN', 'GOOG', 'FB', 'SNAP', 'TWTR'\
'FDX', 'MCD', 'PEP', ]
stocks = list(np.unique(dow + other))
back = backtester(dow, LR_v1_predict, 3000, datetime(2021, 1, 1), datetime(2021, 12, 31), threshold = 1, sell_perc = 0.03, hold_till = 1,\
stop_perc = 0.03)
back.backtest()