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
|
import json
import datetime
import os
def summarize_results(batch_name):
"""
Summarizes the results of the backtesting by calculating average percent gain/loss,
and finding the best stock order.
"""
# get the result data json files under batches
batch_folder = f'batches/{batch_name}'
result_files = [f for f in os.listdir(batch_folder) if f.endswith('.json')]
results_summary = []
# make a dict for the algo table
algos_to_results = {}
print(len(result_files), "results found in", batch_folder)
for result in result_files:
file_path = os.path.join(batch_folder, result)
with open(file_path, 'r') as fd:
result_data = json.load(fd)
# extract the relevant data
result = result_data['backtest_results']
if not result:
continue
algo_name = result['algo_name']
algo_params = result['algo_params']
algo_key = f"{algo_name}_{algo_params}"
if algo_key not in algos_to_results:
algos_to_results[algo_key] = {
'percent_gains': [],
'stock_orders': []
}
algos_to_results[algo_key]['percent_gains'].append(result['percent_gain'])
stock_ticker = result_data['url_params']['ticker']
algos_to_results[algo_key]['stock_orders'].append(stock_ticker)
# summarize the results for each algo
for algo_name, result in algos_to_results.items():
algo_params = algo_name.split('_')[1] # Extract params from the key
if not result['percent_gains'] and not result['percent_losses']:
continue # Skip if no gains or losses
# calculate average percent gain/loss
avg_percent_gain = sum(result['percent_gains']) / len(result['percent_gains']) if result['percent_gains'] else 0
# calculate which stock ticker produced the best result
stock_orders = result['stock_orders']
if stock_orders:
best_stock_order = max(set(stock_orders), key=stock_orders.count)
else:
best_stock_order = "N/A"
# Append the summarized data
results_summary.append([
algo_name,
algo_params,
avg_percent_gain,
best_stock_order
])
# Sort the results by average percent gain in descending order
results_summary.sort(key=lambda x: x[2], reverse=True)
print(algos_to_results)
return results_summary
def test():
"""
Test function to summarize results from a specific batch.
"""
batch_name = 'test-1-ema'
results = summarize_results(batch_name)
if results:
print("Results Summary:")
for row in results:
print(row)
else:
print("No results found.")
test()
# pull stock data from json files
# timestamps_file = open('timestamps.json', 'r')
# timestamps_file_data = timestamps_file.read()
# timestamps = json.loads(timestamps_file_data)
# timestamps = [datetime.datetime.fromtimestamp(t) for t in timestamps]
# prices_file = open('close_prices.json', 'r')
# prices = json.loads(prices_file.read())
# print('timestamps:\t', timestamps, '\nprices:\t', prices)
# make the line data for the 5 day exponential moving average (EMA)
def interpolate_intersection(intersection_indices, timestamps, prices1, prices2):
left_index = intersection_indices[0]
right_index = intersection_indices[1]
if right_index == -1:
return timestamps[left_index]
y_1 = prices1[left_index]
y_2 = prices1[right_index] # first line
v_1 = prices2[left_index]
v_2 = prices2[right_index] # second line
x_1 = 0 # take this as zero the simplify the algebra
x_diff = timestamps[right_index] - timestamps[left_index] # same for both lines
# find intersection between those lines
x_diff = x_diff.total_seconds()
m_1 = (y_2 - y_1) / x_diff # slope of line 1
m_2 = (v_2 - v_1) / x_diff
x_interpolated = (v_1 - y_1) / (m_1 - m_2)
y_interpolated = m_1 * (x_interpolated) + y_1
# add back the time we subtracted to make x_1=0
x_interpolated = datetime.timedelta(seconds = x_interpolated) + timestamps[left_index]
return (x_interpolated, y_interpolated)
"""
Returns the indices of where two arrays' values intersects
"""
def find_intersections(prices1, prices2, offset=0):
if len(prices1) != len(prices2):
print("ERROR IN find_intersections: len of arrs not the same")
return []
prev_p1 = prices1[offset]
prev_p2 = prices2[offset]
intersection_indices = set()
for i in range(1 + offset, len(prices1)):
next_p1 = prices1[i]
next_p2 = prices2[i]
# if the sign (negative to positive) changes, then there was an intersection between these pts
sub_prev = prev_p1 - prev_p2
sub_next = next_p1 - next_p2
if (sub_prev > 0 and sub_next < 0) or (sub_prev < 0 and sub_next > 0):
intersection_indices.add((i-1, i))
if sub_next == 0:
intersection_indices.add((i, -1))
prev_p1 = next_p1
prev_p2 = next_p2
return intersection_indices
|
