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
|
from algo import Algo
from ema_algo import Ema_Algo
from api import fetch_chart_data_yahoo, fetch_chart_data_backtest
import datetime
import json
import random
import os
"""
Function that takes in data and returns a buy, sell, or hold singal per interval
"""
def backtest_algo(algo : Algo, timestamps, prices, init_offset=5, starting_money=10000):
# take a fraction of the OG data (or a set #)
assert len(timestamps) == len(prices)
assert init_offset < len(timestamps) # make sure enough data to start with
current_timestamps = timestamps[:init_offset]
current_prices = prices[:init_offset]
is_bought = 0.0 # not holding anything
current_liquid = 10000
shares_owned = 0
buy_data = []
sell_data = []
for i in range(init_offset, len(timestamps)):
# update prices array and run algo
current_timestamps.append(timestamps[i])
current_prices.append(prices[i])
current_signal = algo.detemine_signal(current_timestamps, current_prices)
cur_p = current_prices[-1]
cur_t = current_timestamps[-1]
if current_signal == 1.0: # signal is to buy
if is_bought == 0.0: # if we haven't bought, purchase
shares_owned = current_liquid / cur_p
current_liquid = 0
is_bought = 1.0
buy_data.append(i)
#print("buy", shares_owned, current_liquid, datetime.datetime.fromtimestamp(timestamps[i]))
elif current_signal == 0.0: # signal sell all
if is_bought == 1.0: # if we have bought, sell!
current_liquid = shares_owned * cur_p
shares_owned = 0
is_bought = 0.0
sell_data.append(i)
#print('sell', shares_owned, current_liquid, datetime.datetime.fromtimestamp(timestamps[i]))
# calculate total assets
assets = prices[-1] * shares_owned + current_liquid
percent_gain = 100 * (assets - starting_money) / starting_money
print(assets, percent_gain)
# create a json to store the reuslts
results = {
"timestamps" : timestamps,
"prices" : prices,
"buy_indices" : buy_data,
"sell_indices" : sell_data,
"percent_gain" : percent_gain,
"starting_assets" : starting_money,
"final_assets" : assets,
"algo_name" : algo.name,
"algo_graph_data" : algo.graph_data
}
return results
# store all algo name, buy, sell, price data, timestamps, into a json so it can be viewed as a trial
# caluclate some metrics (NOW: only how much money gained, how many trades, trades per day... etc)
ticker_bank = ['BTC-USD', 'ADA-USD', 'ETH-USD', 'ETC-USD', 'DOGE-USD']
year_bank = [2020, 2021, 2022, 2023, 2024, 2025]
# time_bank = [0]
# max_seconds = 24 * 60 * 60
# timedelta(seconds=rand_second) + datetime(created with 0 time)
"""
Runs N trials with random parameters
"""
def run_batch(batch_name, algo, num_trials=100):
i = 1
while i <= num_trials:
# pick a random set of parameters
rand_ticker = ticker_bank[random.randint(0, len(ticker_bank) - 1)]
rand_year = year_bank[random.randint(0, len(year_bank) - 1)]
rand_day = random.randint(1, 31)
rand_month = random.randint(1, 12)
# ensure the date is valid
rand_date = None
try:
rand_date = datetime.datetime(rand_year, rand_month, rand_day)
except ValueError:
# date creation failed
continue
# ensure date is not in future
curr_date = datetime.datetime.now()
if rand_date > curr_date:
continue
# pull chart data for these params and run the algo
data = fetch_chart_data_backtest(rand_ticker, '1m', rand_date)
results = backtest_algo(algo, data['timestamps'], data['prices'], 13)
# TODO: make this generalized
url_params = {
"ticker" : rand_ticker,
"period" : '8d',
"interval" : '1m',
}
# store the results in into a file
trial_data = {
"chart_data" : data,
"url_params" : url_params,
"backtest_results" : results
}
# make a new directory for the batch
path = f'batches_{algo.name}/{batch_name}'
if i == 1:
print(path)
try:
os.makedirs(path)
except PermissionError:
print(f"Permission denied: Unable to create {path}.")
return
except Exception as e:
print(f"An error occurred: {e}")
return
percent_gain = results['percent_gain']
date_format = datetime.datetime.isoformat(rand_date)
file_name = f'{path}/{percent_gain}_{rand_ticker}_{date_format}.json'
fd = open(file_name, 'w')
fd.write(json.dumps(trial_data))
fd.close()
# increment trial num
i += 1
# run_batch('test', Ema_Algo(), 5)
def test():
print("MAIN simulate.py")
ema_algo = Ema_Algo()
ticker = 'XRP/USD'
period = '5d'
interval = '1m'
# get data
# data = fetch_chart_data(ticker, period, interval)
# period_end_date = datetime.datetime.now(tz=datetime.timezone.utc) - datetime.timedelta(days=7)
# print(period_end_date)
data = fetch_chart_data_yahoo('XRP-USD', '1h', None, datetime.timedelta(weeks=52))
print(data.keys())
url_params = {
"ticker" : ticker,
"period" : period,
"interval" : interval,
}
results = backtest_algo(ema_algo, data['timestamps'], data['prices'], 13)
# write the data into a json to be viewed in a chart
trial_data = {
"chart_data" : data,
"url_params" : url_params,
"backtest_results" : results
}
fd = open('bt-recent.json', 'w')
fd.write(json.dumps(trial_data))
fd.close()
test()
|
