-
Notifications
You must be signed in to change notification settings - Fork 0
/
helper.py
221 lines (198 loc) · 9.85 KB
/
helper.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
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
import pandas as pd
from datetime import timedelta
from os import stat
import numpy as np
import quantstats as qs
class Trade:
"""
This class handles all trade objects.
"""
def __init__(self, tradeop):
self.id = tradeop[0]
self.value = tradeop[1]
self.exp = tradeop[2]
self.strike = tradeop[3]
def update_trade(self, value):
"""
This function is used to update the value of a trade.
This should be used to close a trade.
"""
self.value = value
def get_value(self):
return self.value
class AccountBal:
"""
This is the Account Balance class which will handle accounts in the simulation.
"""
def __init__(self, value):
self.value = value
self.trade_val = []
self.account_bal = []
self.wins = []
def add_trade(self, trade_val):
"""
This function is used to add the value of trades to the list within the account.
"""
self.trade_val.append(trade_val)
if trade_val <= 0:
change = self.trade_val[-2] + trade_val # if it is 0 or negative means closing trade
self.value = self.value + change
self.account_bal.append(self.value)
else:
pass
def update_balance(self, change):
"""
This function helps to update the account balance and keep track of the changing balance from day to day
"""
self.value = self.value + change
self.account_bal.append(self.value)
def update_wins(self, change):
"""
This function updates the number of wins in an account
"""
self.wins.append(change)
class TradeOperations:
"""
static storage for various trade operations that
work on a higher level
"""
@staticmethod
def find_trade(dataset):
"""
This function is used to find a new trade to enter. It looks for a strike price and sets a DTE range.
These two parameters are sufficient to source for the trade to enter. It returns a tuple which
is then used within the Trade class to create a new Trade object.
"""
strike_price = np.round(dataset["Underlying"].iloc[0])
dte_max = timedelta(9)
dte_min = timedelta(6)
trade_dets = dataset[(dataset["DTE"] < dte_max) &
(dataset["DTE"] > dte_min) & (dataset["Strike"] == strike_price)][["oid", "Bid", "Expiration", "Strike"]].values[0]
trade_oid = trade_dets[0]
trade_val = trade_dets[1]
trade_exp = trade_dets[2]
trade_strike = trade_dets[3]
return trade_oid, trade_val, trade_exp, trade_strike
@staticmethod
def find_same_strike(previous_trade, dataset):
"""
This function is used to find a new trade to enter, when we are holding the strike.
It matches the strike price to the previous trade and sets a DTE range.
These two parameters are sufficient to source for the trade to enter. It returns a tuple which
is then used within the Trade class to create a new Trade object.
"""
strike_price = previous_trade # set the strike price to the previous trade to "hold the strike"
dte_max = timedelta(9)
dte_min = timedelta(6)
trade_dets = dataset[(dataset["DTE"] < dte_max) &
(dataset["DTE"] > dte_min) & (dataset["Strike"] == strike_price)][["oid", "Bid", "Expiration", "Strike"]].values[0]
trade_oid = trade_dets[0]
trade_val = trade_dets[1]
trade_exp = trade_dets[2]
trade_strike = trade_dets[3]
return trade_oid, trade_val, trade_exp, trade_strike
@staticmethod
def close_trade(account, curr_trade, close_val, next_trade):
"""
This function is used to close a trade and open a new one. It replicates the effect of
"rolling"
"""
account.add_trade(curr_trade.get_value())
account.update_balance(curr_trade.get_value() - close_val)
curr_trade.update_trade(close_val)
account.add_trade(-curr_trade.get_value())
account.add_trade(next_trade.get_value())
class BacktestOperations:
"""
Static holder to store operations for backtesting
"""
@staticmethod
def run_backtest(curr_date, dataset, acc_initial_value): # start backtest, with own dataset
days = []
curr_account = AccountBal(acc_initial_value)
tradeOn = False
while curr_date < dataset["DataDate"].max(): # loop through the days
curr_date += timedelta(1) # increase days
days.append(curr_date) # store days
daily_set = dataset[dataset["DataDate"] == curr_date] # create sub-df with just today's data
if tradeOn == False: # if there is no trade on, we want to search for a trade
try:
new_trade = Trade(TradeOperations.find_trade(daily_set)) # create new trade class
curr_account.add_trade(new_trade.get_value())
curr_account.account_bal.append(curr_account.value)
tradeOn = True
except:
print(str(curr_date) + " is a Weekend, unable to open")
curr_account.account_bal.append(curr_account.value)
elif tradeOn == True:
try:
if new_trade.exp-timedelta(0) == curr_date:
new_trade.update_trade(daily_set[daily_set["oid"] == new_trade.id]["Ask"].values[0])
curr_account.add_trade(-new_trade.get_value())
print(str(curr_date) + " New Trade Added " + str(new_trade.id))
new_trade = Trade(TradeOperations.find_trade(daily_set))
curr_account.add_trade(new_trade.get_value())
else:
print(str(curr_date) + " No Trades Running " + str(new_trade.id))
curr_account.account_bal.append(curr_account.value)
pass
except:
print(str(curr_date) + " ERROR: Closing Trade Manually ")
new_trade.update_trade(float(new_trade.get_value())/2)
curr_account.add_trade(-new_trade.get_value())
tradeOn = False
return curr_account.trade_val, curr_account.account_bal, days
@staticmethod
def run_backtest_HTS(curr_date, dataset, acc_initial_value): # start backtest, with own dataset
days = []
curr_account = AccountBal(acc_initial_value)
tradeOn = False
while curr_date < dataset["DataDate"].max(): # loop through the days
curr_date += timedelta(1) # increase days
days.append(curr_date) # store days
daily_set = dataset[dataset["DataDate"] == curr_date] # create sub-df with just today's data
if tradeOn == False: # if there is no trade on, we want to search for a trade
try:
new_trade = Trade(TradeOperations.find_trade(daily_set)) # create new trade class
curr_account.add_trade(new_trade.get_value())
curr_account.account_bal.append(curr_account.value)
tradeOn = True
except:
print(str(curr_date) + " is a Weekend, unable to open")
curr_account.account_bal.append(curr_account.value)
elif tradeOn == True:
try:
if new_trade.exp-timedelta(0) == curr_date: # for the day itself...
closing_trade_val = new_trade.get_value() # here, we store the values for the trade that is being closed and its strike
closing_trade_strike = new_trade.strike
opening_trade_val = daily_set[daily_set["oid"] == new_trade.id]["Ask"].values[0] # find the trade's closing price
new_trade.update_trade(opening_trade_val) # either way, we need to add in the closing price of the trade
curr_account.add_trade(-new_trade.get_value())
if closing_trade_val < opening_trade_val: # if we are losing, hold the strike
new_trade = Trade(TradeOperations.find_same_strike(closing_trade_strike, daily_set))
print(str(curr_date) + " HOLD THE STRIKE " + str(new_trade.id))
curr_account.add_trade(new_trade.get_value())
elif closing_trade_val >= opening_trade_val: # if we won, just open a new trade as per usual (i.e. ATM)
new_trade = Trade(TradeOperations.find_trade(daily_set))
print(str(curr_date) + " New Trade Added " + str(new_trade.id))
curr_account.add_trade(new_trade.get_value())
else:
print(str(curr_date) + " No Trades Running " + str(new_trade.id))
curr_account.account_bal.append(curr_account.value)
pass
except:
print(str(curr_date) + " ERROR: Closing Trade Manually ")
new_trade.update_trade(float(new_trade.get_value())/2)
curr_account.add_trade(-new_trade.get_value())
tradeOn = False
return curr_account.trade_val, curr_account.account_bal, days
@staticmethod
def benchmark_results(bal, days, benchmark="SPY"):
"""
Function to allow one to analyse the results of a
backtest. Requires list of balances and dates from
backtest results.
Default benchmark is SPY but this can be changed.
"""
returns_series = pd.DataFrame({"Returns": pd.Series(bal[0:len(days)]).pct_change(), "Date":days}).set_index("Date").squeeze()
return qs.reports.basic(returns_series, benchmark)