-
Notifications
You must be signed in to change notification settings - Fork 0
/
SignalMaker.py
162 lines (123 loc) · 6.79 KB
/
SignalMaker.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
#_*_ coding: utf-8 _*_
import pandas as pd
import Utils
class SignalMaker: # 클래스
def __init__(self):
print("Generate SignalMaker.")
def __del__(self):
print("Destroy SignalMaker.")
def get_golden_cross_buy_signal(self, series, series_num, short_term, long_term
, short_term_momentum_threshold, long_term_momentum_threshold, volume_momentum_threshold=None
, direction=1):
# short term & long term 확인
short_len = min(short_term, series_num)
long_len = min(long_term, series_num)
if short_len == long_len:
return False
rolling_short = series.rolling(window=short_len)
rolling_short_mean = rolling_short.mean().fillna(0)
rolling_long = series.rolling(window=long_len)
rolling_long_mean = rolling_long.mean().fillna(0)
price = series['close'][-1]
short_avg_price = rolling_short_mean['close'][-1]
long_avg_price = rolling_long_mean['close'][-1]
volume = series['volume'][-1]
short_avg_volume = rolling_short_mean['volume'][-1]
long_avg_volume = rolling_long_mean['volume'][-1]
# 신규 포지션을 위해 골든 크로스 확인하는 경우: 1 or 이익 실현을 위해 골든 크로스 확인하는 경우: -1
# 두 경우를 반대로 적용하면 더 빠르게 반응할 수 있다.
short_multiple = short_term_momentum_threshold if direction == 1 else 1/short_term_momentum_threshold
long_multiple = long_term_momentum_threshold if direction == 1 else 1/long_term_momentum_threshold
if price > short_avg_price*short_multiple and short_avg_price > long_avg_price*long_multiple:
if volume_momentum_threshold is None or short_avg_volume > long_avg_volume*volume_momentum_threshold:
return 'BUY'
else:
return False
else:
return False
def get_dead_cross_sell_signal(self, series, series_num, short_term, long_term):
# short term & long term 확인
short_len = min(short_term, series_num)
long_len = min(long_term, series_num)
if short_len == long_len:
return False
price = series.tail(1)['close'].values[0]
short_avg_price = series.tail(short_len)['close'].mean()
long_avg_price = series.tail(long_len)['close'].mean()
volume = series.tail(1)['volume'].values[0]
short_avg_volume = series.tail(short_len)['volume'].mean()
long_avg_volume = series.tail(long_len)['volume'].mean()
if price < short_avg_price and short_avg_price < long_avg_price and short_avg_volume > long_avg_volume:
return 'SELL'
else:
return False
def get_momentum_z_buy_signal(self, series, series_num, short_term, long_term, base=0.0):
# short term & long term 확인
short_len = min(short_term, series_num)
long_len = min(long_term, series_num)
if short_len == long_len:
return False
rolling_short = series.rolling(window=short_len)
rolling_short_mean = rolling_short.mean().fillna(0)
rolling_short_std = rolling_short.std().fillna(0)
rolling_short_z = (series-rolling_short_mean)/rolling_short_std
rolling_short_z_rolling = rolling_short_z.rolling(window=short_len)
rolling_short_z_rolling_mean = rolling_short_z_rolling.mean().fillna(0)
rolling_long = series.rolling(window=long_len)
rolling_long_mean = rolling_long.mean().fillna(0)
rolling_long_std = rolling_long.std().fillna(0)
rolling_long_z = (series-rolling_long_mean)/rolling_long_std
rolling_long_z_rolling = rolling_long_z.rolling(window=long_len)
rolling_long_z_rolling_mean = rolling_long_z_rolling.mean().fillna(0)
if 0:
short_price_momentum = rolling_short_z['close'][-1]
long_price_momentum = rolling_long_z['close'][-1]
else:
short_price_momentum = rolling_short_z_rolling_mean['close'][-1]
long_price_momentum = rolling_long_z_rolling_mean['close'][-1]
# 단기 모멘텀이 최소 수준 이상이면서 단기 모멘텅이 장기 모멘텀보다 큰 경우
if short_price_momentum > base and short_price_momentum > long_price_momentum:
recent_period = 5
if short_price_momentum > rolling_short_z_rolling_mean['close'][-recent_period:].mean() and long_price_momentum > rolling_long_z_rolling_mean['close'][-recent_period:].mean():
# 꼭지에 걸리지 않기 위해 최근 단기 급등하지 않은 경우
rate_threshold = 0.095
if max(series['high'][-recent_period:]) / min(series['low'][-recent_period:]) - 1.0 < rate_threshold:
return 'BUY'
else:
return False
else:
return False
else:
return False
def get_momentum_z_sell_signal(self, series, series_num, short_term, long_term, base=0.0):
# short term & long term 확인
short_len = min(short_term, series_num)
long_len = min(long_term, series_num)
if short_len == long_len:
return False
rolling_short = series.rolling(window=short_len)
rolling_short_mean = rolling_short.mean().fillna(0)
rolling_short_std = rolling_short.std().fillna(0)
rolling_short_z = (series-rolling_short_mean)/rolling_short_std
rolling_short_z_rolling = rolling_short_z.rolling(window=short_len)
rolling_short_z_rolling_mean = rolling_short_z_rolling.mean().fillna(0)
rolling_long = series.rolling(window=long_len)
rolling_long_mean = rolling_long.mean().fillna(0)
rolling_long_std = rolling_long.std().fillna(0)
rolling_long_z = (series-rolling_long_mean)/rolling_long_std
rolling_long_z_rolling = rolling_long_z.rolling(window=long_len)
rolling_long_z_rolling_mean = rolling_long_z_rolling.mean().fillna(0)
if 0:
short_price_momentum = rolling_short_z['close'][-1]
long_price_momentum = rolling_long_z['close'][-1]
short_price_momentum_prev = rolling_short_z['close'][-2]
long_price_momentum_prev = rolling_long_z['close'][-2]
else:
short_price_momentum = rolling_short_z_rolling_mean['close'][-1]
long_price_momentum = rolling_long_z_rolling_mean['close'][-1]
short_price_momentum_prev = rolling_short_z_rolling_mean['close'][-2]
long_price_momentum_prev = rolling_long_z_rolling_mean['close'][-2]
if short_price_momentum < base and short_price_momentum < long_price_momentum and short_price_momentum < short_price_momentum_prev:
return 'SELL'
else:
return False