-
Notifications
You must be signed in to change notification settings - Fork 0
/
trade.py
105 lines (86 loc) · 2.75 KB
/
trade.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
from Param import Param
import numpy as np
import math
import copy
import dataHelper as dh
class Signal:
LONG = 1
SHORT = 0
HOLD = -1
current_pos = HOLD
def trade(predictions, file_number):
# Array for accumulating signals until we have enough information to act
x = np.full([predictions.shape[0], Param.n_out], np.inf)
actual = dh.get_testing_output(Param.filenames[file_number])
actual = dh.get_column(Param.col_to_trade_on, actual)
# will hold the action that we took on each tick
d = np.full([predictions.shape[0], 1], np.inf)
profit = 0
gains = 0
losses = 0
prev_profit = 0
for i in range(predictions.shape[0] - Param.n_out - 1):
# have we made any money this time step?
prev_profit = copy.deepcopy(profit)
profit = check_position(profit, actual[i, 0] - actual[i-1, 0])
if profit - prev_profit > 0:
gains += profit - prev_profit
else:
losses += prev_profit - profit
# END if
curr = np.reshape(actual[i, 0], (-1,1))
future = predictions[i+1, :]
window = np.concatenate((curr, future))
for j in range(1, Param.n_out + 1):
if window[j] - window[j-1] > 0:
is_full = add_signal(x, i+j, Signal.LONG)
elif window[j] - window[j-1] < 0:
is_full = add_signal(x, i+j, Signal.SHORT)
else:
pass
# END if
if is_full:
d[i+j-1] = make_trade(x, i+j-1, actual[i+j-1, 0], predictions[i+j, 0])
Signal.current_pos = d[i+j-1]
# END if
# END for
# END for
print("Gains/Losses for stock " + str(file_number)+ ": " + str(gains/losses))
return profit
def add_signal(x, t, signal):
is_full = False
for i in range(Param.n_out):
if np.isinf(x[t, i]):
x[t, i] = signal
if i == Param.n_out - 1:
is_full = True
# END if
break
# END if
# END for
return is_full
# END add_signal
def make_trade(x, t, recent_actual, predicted):
total_signal = np.sum(x[t,:])
if np.isinf(total_signal):
return np.inf
elif total_signal >= 2:
signal = Signal.LONG
else:
signal = Signal.SHORT
# END if
predicted = predicted[0]
recent_actual = recent_actual[0]
if np.abs((predicted - recent_actual)/recent_actual) > 0.001:
return signal
else:
return Signal.HOLD
# END if
# END make_trade
def check_position(profit, delta):
mult = 1
if Signal.current_pos == Signal.SHORT:
mult = -1
if Signal.current_pos == Signal.SHORT or Signal.current_pos == Signal.LONG:
profit += delta*mult
return profit