-
Notifications
You must be signed in to change notification settings - Fork 0
/
range_trading.py
201 lines (152 loc) · 8.55 KB
/
range_trading.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
from distutils.command.build_py import build_py
import pandas as pd
import numpy as np
# local imports
from backtester import tester
TRAINING_PERIOD = 20 # How far the rolling average takes into calculation
STD_DV_CONST = 3.5 # Number of Standard Deviations from the mean the Bollinger Bands sit
STD_MULTIPLIER = 2 #Multiples of Standard Devs Const used for Boll Bands
INTERVAL = 14 # Interval for bollinger bands and ADX calculations
# bounds for RSI
RSI_BOUNDS = {
"upper": 70,
"lower": 30,
}
'''
logic() function:
Context: Called for every row in the input data.
Input: account - the account object
lookback - the lookback dataframe, containing all data up until this point in time
Output: none, but the account object will be modified on each call
'''
# lookback is going to be a pandas dataframe e.g. 'Open':[100, 87, 69, 11]
# 'Volume':[23, 45, 76, 93]
def logic(account, lookback): # Logic function to be used for each time interval in backtest
last_index = len(lookback)-1
if(last_index <= TRAINING_PERIOD): # If the lookback is long enough to calculate the Bollinger Bands
return # A coding style that is the same as tabbed if statement.
ADX = lookback['ADX'+str(INTERVAL)][last_index]
RSI = lookback['RSI'][last_index]
close_price = lookback['close'][last_index]
DI_POSITIVE = lookback['+DI'+str(INTERVAL)][last_index]
DI_NEGATIVE = lookback['-DI'+str(INTERVAL)][last_index]
#if ADX >= 20 and ADX < 25:
# coeff = STD_MULTIPLIER*STD_DV_CONST
if ADX>= 25:
coeff = 1*STD_DV_CONST
elif ADX > 10 and ADX < 25:
coeff = 1*STD_DV_CONST
else:
coeff = 0.8*STD_DV_CONST
BOLU = lookback['MA-TP'][last_index] + coeff*lookback['std'][last_index] # Calculate Upper Bollinger Band
BOLD = lookback['MA-TP'][last_index] - coeff*lookback['std'][last_index] # Calculate Lower Bollinger Band
# when ADX is >= 25, only close positions
if ADX >= 25:
# If it trends below BOLD, and is headed downwards, close long positions
if(close_price < BOLD):
if DI_POSITIVE < DI_NEGATIVE:
for position in account.positions:
if position.type_ == 'long':
account.close_position(position, 1, close_price)
else:
for position in account.positions:
if position.type_ == 'short':
account.close_position(position, 1, close_price)
if (close_price > BOLU):
if DI_POSITIVE > DI_NEGATIVE:
for position in account.positions:
if position.type_ == 'short':
account.close_position(position, 1, close_price)
else:
for position in account.positions:
if position.type_ == 'long':
account.close_position(position, 1, close_price)
return
if(close_price < BOLD): # If current price is below lower Bollinger Band, enter a long position
# if RSI < 30 (undervalued) and if price is under BOLD, then buy
if RSI > RSI_BOUNDS["lower"]:
return
for position in account.positions: # Close all current positions
account.close_position(position, 1, close_price)
if(account.buying_power > 0):
account.enter_position('long', account.buying_power, close_price) # Enter a long position (enter full position)
if(close_price > BOLU): # If today's price is above the upper Bollinger Band, enter a short position
# if RSI > 70 (overvalued) and price moves above Upper bol band then short
if RSI < RSI_BOUNDS["upper"]:
return
for position in account.positions: # Close all current positions
account.close_position(position, 1, close_price)
if(account.buying_power > 0):
account.enter_position('short', account.buying_power, close_price) # Enter a short position
def calc_rsi(data, periods=14):
close_delta = data['close'].diff()
# Make two series: one for lower closes and one for higher closes
up = close_delta.clip(lower=0)
down = -1 * close_delta.clip(upper=0)
ma_up = up.ewm(com=periods - 1, adjust=True, min_periods=periods).mean()
ma_down = down.ewm(com=periods - 1, adjust=True,
min_periods=periods).mean()
rsi = ma_up / ma_down
rsi = 100 - (100/(1 + rsi))
return rsi
'''
preprocess_data() function:
Context: Called once at the beginning of the backtest. TOTALLY OPTIONAL.
Each of these can be calculated at each time interval, however this is likely slower.
Input: list_of_stocks - a list of stock data csvs to be processed
Output: list_of_stocks_processed - a list of processed stock data csvs
'''
def preprocess_data(list_of_stocks):
list_of_stocks_processed = [] # create empty list to append to
for stock in list_of_stocks:
df = pd.read_csv("data/" + stock + ".csv", parse_dates=[0])
df['TP'] = (df['close'] + df['low'] + df['high'])/3 # Calculate Typical Price
df['std'] = df['TP'].rolling(TRAINING_PERIOD).std() # Calculate Standard Deviation
df['MA-TP'] = df['TP'].rolling(TRAINING_PERIOD).mean() # Calculate Moving Average of Typical Price
df['sum']=df['close'].cumsum()
df['-DM'] = df['low'].shift(1) - df['low'] # Directional Movement : previous low minus current low
df['+DM'] = df['high'] - df['high'].shift(1) # Directional Movement : current high minus previous high
df['+DM'] = np.where((df['+DM'] > df['-DM']) & (df['+DM']>0), df['+DM'], 0.0)
df['-DM'] = np.where((df['-DM'] > df['+DM']) & (df['-DM']>0), df['-DM'], 0.0)
df['TR_TMP1'] = df['high'] - df['low']
df['TR_TMP2'] = np.abs(df['high'] - df['close'].shift(1))
df['TR_TMP3'] = np.abs(df['low'] - df['close'].shift(1))
df['TR'] = df[['TR_TMP1', 'TR_TMP2', 'TR_TMP3']].max(axis=1)
df['TR'+str(INTERVAL)] = df['TR'].rolling(INTERVAL).sum()
df['+DMI'+str(INTERVAL)] = df['+DM'].rolling(INTERVAL).sum()
df['-DMI'+str(INTERVAL)] = df['-DM'].rolling(INTERVAL).sum()
df['+DI'+str(INTERVAL)] = df['+DMI'+str(INTERVAL)] / df['TR'+str(INTERVAL)]*100
df['-DI'+str(INTERVAL)] = df['-DMI'+str(INTERVAL)] / df['TR'+str(INTERVAL)]*100
df['DI'+str(INTERVAL)+'-'] = abs(df['+DI'+str(INTERVAL)] - df['-DI'+str(INTERVAL)])
df['DI'+str(INTERVAL)] = df['+DI'+str(INTERVAL)] + df['-DI'+str(INTERVAL)]
df['DX'] = (df['DI'+str(INTERVAL)+'-'] / df['DI'+str(INTERVAL)])*100
df['ADX'+str(INTERVAL)] = df['DX'].rolling(INTERVAL).mean()
df['ADX'+str(INTERVAL)] = df['ADX'+str(INTERVAL)].fillna(df['ADX'+str(INTERVAL)].mean())
del df['TR_TMP1'], df['TR_TMP2'], df['TR_TMP3'], df['TR'], df['TR'+str(INTERVAL)]
del df['+DMI'+str(INTERVAL)], df['DI'+str(INTERVAL)+'-']
del df['DI'+str(INTERVAL)], df['-DMI'+str(INTERVAL)] # interval is 14 (ADX14)
#del df['+DI'+str(INTERVAL)], df['-DI'+str(INTERVAL)]
del df['DX']
# RSI
df["RSI"] = calc_rsi(df, INTERVAL)
df.to_csv("data/" + stock + "_Processed.csv", index=False) # Save to CSV
list_of_stocks_processed.append(stock + "_Processed")
return list_of_stocks_processed
if __name__ == "__main__":
REMAKE_DATA = True
# remake *_Processed.csv for all stocks in "stocks_to_process"
if REMAKE_DATA:
stocks_to_process = ["WMT_2020-10-05_2022-08-26_15min", "NDAQ_2020-10-05_2022-08-26_15min"] + ["TSLA_2020-03-09_2022-01-28_15min", "AAPL_2020-03-24_2022-02-12_15min"]
list_of_stocks = preprocess_data(
stocks_to_process) # Preprocess the data
# resuse cvs from "list_of_stocks"
else:
# List of stock data csv's to be tested, located in "data/" folder
#"TSLA_2020-03-09_2022-01-28_15min_Processed"
list_of_stocks = ["WMT_2020-10-05_2022-08-26_15min_Processed", "NDAQ_2020-10-05_2022-08-26_15min_Processed"]
results = tester.test_array(list_of_stocks, logic, chart=True) # Run backtest on list of stocks using the logic function
# passing logic function as parameter.
print("training period " + str(TRAINING_PERIOD))
print("standard deviations " + str(STD_DV_CONST))
df = pd.DataFrame(list(results), columns=["Buy and Hold","Strategy","Longs","Sells","Shorts","Covers","Stdev_Strategy","Stdev_Hold","Stock"]) # Create dataframe of results
df.to_csv("results/Test_Results.csv", index=False) # Save results to csv