-
Notifications
You must be signed in to change notification settings - Fork 1
/
mean_reversion_example.py
70 lines (56 loc) · 3.7 KB
/
mean_reversion_example.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
import pandas as pd
import time
import multiprocessing as mp
# local imports
from backtester import engine, tester
from backtester import API_Interface as api
training_period = 20 # How far the rolling average takes into calculation
standard_deviations = 3.5 # Number of Standard Deviations from the mean the Bollinger Bands sit
'''
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
'''
def logic(account, lookback): # Logic function to be used for each time interval in backtest
today = len(lookback)-1
if(today > training_period): # If the lookback is long enough to calculate the Bollinger Bands
if(lookback['close'][today] < lookback['BOLD'][today]): # If current price is below lower Bollinger Band, enter a long position
for position in account.positions: # Close all current positions
account.close_position(position, 1, lookback['close'][today])
if(account.buying_power > 0):
account.enter_position('long', account.buying_power, lookback['close'][today]) # Enter a long position
if(lookback['close'][today] > lookback['BOLU'][today]): # If today's price is above the upper Bollinger Band, enter a short position
for position in account.positions: # Close all current positions
account.close_position(position, 1, lookback['close'][today])
if(account.buying_power > 0):
account.enter_position('short', account.buying_power, lookback['close'][today]) # Enter a short position
'''
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 = []
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['BOLU'] = df['MA-TP'] + standard_deviations*df['std'] # Calculate Upper Bollinger Band
df['BOLD'] = df['MA-TP'] - standard_deviations*df['std'] # Calculate Lower Bollinger Band
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__":
# list_of_stocks = ["TSLA_2020-03-01_2022-01-20_1min"]
list_of_stocks = ["TSLA_2020-03-09_2022-01-28_15min", "AAPL_2020-03-24_2022-02-12_15min"] # List of stock data csv's to be tested, located in "data/" folder
list_of_stocks_proccessed = preprocess_data(list_of_stocks) # Preprocess the data
results = tester.test_array(list_of_stocks_proccessed, logic, chart=True) # Run backtest on list of stocks using the logic function
print("training period " + str(training_period))
print("standard deviations " + str(standard_deviations))
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