forked from waditu/czsc
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathtest_objects.py
280 lines (244 loc) · 10.4 KB
/
test_objects.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
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
# coding: utf-8
import numpy as np
from collections import OrderedDict
from czsc.utils import x_round
from czsc.objects import Signal, Factor, Event, Freq, Operate
from czsc.objects import cal_break_even_point
def test_raw_bar():
from test.test_analyze import read_daily
from czsc.utils.ta import SMA
bars = read_daily()
ma = SMA(np.array([x.close for x in bars]), 5)
key = "SMA5"
# 技术指标的全部更新
for i in range(1, len(bars) + 1):
c = dict(bars[-i].cache) if bars[-i].cache else dict()
c.update({key: ma[-i]})
bars[-i].cache = c
assert np.array([x.cache[key] for x in bars]).sum() == ma.sum()
# 技术指标的部分更新
for i in range(1, 101):
c = dict(bars[-i].cache) if bars[-i].cache else dict()
c.update({key: ma[-i] + 2})
bars[-i].cache = c
assert np.array([x.cache[key] for x in bars]).sum() == ma.sum() + 200
def test_zs():
"""测试中枢对象"""
from test.test_analyze import read_daily
from czsc.objects import ZS
from czsc.analyze import CZSC
bars = read_daily()
c = CZSC(bars)
zs = ZS(c.bi_list[-5:])
assert zs.zd < zs.zg
assert zs.is_valid
zs = ZS(c.bi_list[-8:-3])
assert not zs.is_valid
def test_cal_break_even_point():
assert cal_break_even_point([]) == 1
assert cal_break_even_point([1]) == 1
assert cal_break_even_point([-1, -2, 4, 5, 5]) == 0.6
assert cal_break_even_point([-1, -2]) == 1
assert cal_break_even_point([1, 2]) == 0.5
assert cal_break_even_point([-1, 1, -2]) == 1
assert cal_break_even_point([0, 1, -1, 2, 3, -6, -7]) == 1
assert x_round(cal_break_even_point([0, 1, -1, 2, 3, -6, 7, 8])) == 0.875
assert x_round(cal_break_even_point([-6, -1, 0, 1, 2, 3, 7, 8])) == 0.875
assert x_round(cal_break_even_point([2, 3, 4, 2, 1, 4, 0, 1, -1, 2, 3, -6, 7, 8])) == 0.5714
def test_signal():
s = Signal(k1="1分钟", k3="倒1形态", v1="类一买", v2="七笔", v3="基础型", score=3)
assert str(s) == "Signal('1分钟_任意_倒1形态_类一买_七笔_基础型_3')"
assert s.key == "1分钟_倒1形态"
s1 = Signal(signal='1分钟_任意_倒1形态_类一买_七笔_基础型_3')
assert s == s1
assert s.is_match({"1分钟_倒1形态": "类一买_七笔_基础型_3"})
assert not s.is_match({"1分钟_倒1形态": "类一买_七笔_特例一_3"})
assert not s.is_match({"1分钟_倒1形态": "类一买_九笔_基础型_3"})
s = Signal(k1="1分钟", k2="倒1形态", k3="类一买", score=3)
assert str(s) == "Signal('1分钟_倒1形态_类一买_任意_任意_任意_3')"
assert s.key == "1分钟_倒1形态_类一买"
try:
s = Signal(k1="1分钟", k2="倒1形态", k3="类一买", score=101)
except ValueError as e:
assert str(e) == 'score 必须在0~100之间'
def test_factor():
freq = Freq.F15
s = OrderedDict()
default_signals = [
Signal(k1=str(freq.value), k2="倒0笔", k3="方向", v1="向上", v2='其他', v3='其他'),
Signal(k1=str(freq.value), k2="倒0笔", k3="长度", v1="大于5", v2='其他', v3='其他'),
Signal(k1=str(freq.value), k2="倒0笔", k3="三K形态", v1="顶分型", v2='其他', v3='其他'),
Signal(k1=str(freq.value), k2="倒1笔", k3="表里关系", v1="其他", v2='其他', v3='其他'),
Signal(k1=str(freq.value), k2="倒1笔", k3="RSQ状态", v1="小于0.2", v2='其他', v3='其他'),
]
for signal in default_signals:
s[signal.key] = signal.value
factor = Factor(
name="单测",
signals_all=[
Signal(k1=str(freq.value), k2="倒0笔", k3="方向", v1="向上", v2='其他', v3='其他'),
Signal(k1=str(freq.value), k2="倒0笔", k3="长度", v1="大于5", v2='其他', v3='其他')
]
)
assert factor.is_match(s)
factor_raw = factor.dump()
new_factor = Factor.load(factor_raw)
assert new_factor.is_match(s)
factor = Factor(
name="单测",
signals_all=[
Signal(k1=str(freq.value), k2="倒0笔", k3="方向", v1="向上", v2='其他', v3='其他'),
Signal(k1=str(freq.value), k2="倒0笔", k3="长度", v1="大于5", v2='其他', v3='其他')
],
signals_any=[
Signal(k1=str(freq.value), k2="倒1笔", k3="RSQ状态", v1="小于0.2", v2='其他', v3='其他')
]
)
assert factor.is_match(s)
factor = Factor(
name="单测",
signals_all=[
Signal(k1=str(freq.value), k2="倒0笔", k3="方向", v1="向上", v2='其他', v3='其他'),
Signal(k1=str(freq.value), k2="倒0笔", k3="长度", v1="大于5", v2='其他', v3='其他')
],
signals_any=[
Signal(k1=str(freq.value), k2="倒1笔", k3="RSQ状态", v1="小于0.8", v2='其他', v3='其他')
]
)
assert not factor.is_match(s)
factor = Factor(
name="单测",
signals_all=[
Signal(k1=str(freq.value), k2="倒0笔", k3="方向", v1="向上", v2='其他', v3='其他'),
Signal(k1=str(freq.value), k2="倒0笔", k3="长度", v1="大于5", v2='其他', v3='其他')
],
signals_any=[
Signal(k1=str(freq.value), k2="倒1笔", k3="RSQ状态", v1="小于0.2", v2='其他', v3='其他')
],
signals_not=[
Signal(k1=str(freq.value), k2="倒0笔", k3="三K形态", v1="顶分型", v2='其他', v3='其他'),
]
)
assert not factor.is_match(s)
def test_event():
freq = Freq.F15
s = OrderedDict()
default_signals = [
Signal(k1=str(freq.value), k2="倒0笔", k3="方向", v1="向上", v2='其他', v3='其他'),
Signal(k1=str(freq.value), k2="倒0笔", k3="长度", v1="大于5", v2='其他', v3='其他'),
Signal(k1=str(freq.value), k2="倒0笔", k3="三K形态", v1="顶分型", v2='其他', v3='其他'),
Signal(k1=str(freq.value), k2="倒1笔", k3="表里关系", v1="其他", v2='其他', v3='其他'),
Signal(k1=str(freq.value), k2="倒1笔", k3="RSQ状态", v1="小于0.2", v2='其他', v3='其他'),
]
for signal in default_signals:
s[signal.key] = signal.value
event = Event(name="单测", operate=Operate.LO, factors=[
Factor(name="测试", signals_all=[
Signal(k1=str(freq.value), k2="倒0笔", k3="长度", v1="大于5", v2='其他', v3='其他')])
], signals_all=[
Signal(k1=str(freq.value), k2="倒0笔", k3="方向", v1="向上", v2='其他', v3='其他'),
])
m, f = event.is_match(s)
assert m and f
raw = event.dump()
new_event = Event.load(raw)
m, f = new_event.is_match(s)
assert m and f
raw1 = {'name': '单测',
'operate': '开多',
'signals_all': ['15分钟_倒0笔_方向_向上_其他_其他_0'],
'factors': [{'name': '测试', 'signals_all': ['15分钟_倒0笔_长度_大于5_其他_其他_0']}]}
new_event = Event.load(raw1)
m, f = new_event.is_match(s)
assert m and f
raw1 = {'operate': '开多',
'signals_all': ['15分钟_倒0笔_方向_向上_其他_其他_0'],
'factors': [{'name': '测试', 'signals_all': ['15分钟_倒0笔_长度_大于5_其他_其他_0']}]}
new_event = Event.load(raw1)
m, f = new_event.is_match(s)
assert m and f
event = Event(name="单测", operate=Operate.LO,
factors=[
Factor(name="测试", signals_all=[
Signal('15分钟_倒0笔_长度_大于5_其他_其他_0')
]),
],
signals_any=[
Signal('15分钟_倒0笔_方向_向上_其他_其他_0'),
Signal('15分钟_倒0笔_长度_大于100_其他_其他_0')
])
m, f = event.is_match(s)
assert m and f
event = Event(name="单测", operate=Operate.LO,
factors=[
Factor(name="测试", signals_all=[
Signal(k1=str(freq.value), k2="倒0笔", k3="长度", v1="大于5", v2='其他', v3='其他')])
],
signals_not=[
Signal(k1=str(freq.value), k2="倒0笔", k3="方向", v1="向上", v2='其他', v3='其他'),
])
m, f = event.is_match(s)
assert not m and not f
event = Event(name="单测", operate=Operate.LO, factors=[
Factor(
name="测试",
signals_all=[
Signal(k1=str(freq.value), k2="倒0笔", k3="方向", v1="向上", v2='其他', v3='其他'),
Signal(k1=str(freq.value), k2="倒0笔", k3="长度", v1="大于5", v2='其他', v3='其他')]
)
])
m, f = event.is_match(s)
assert m and f
event = Event(name="单测", operate=Operate.LO, factors=[
Factor(
name="测试",
signals_all=[
Signal('15分钟_倒0笔_方向_向上_其他_其他_0'), Signal('15分钟_倒0笔_长度_任意_其他_其他_0')
]
)
])
m, f = event.is_match(s)
assert m and f
event = Event(name="单测", operate=Operate.LO, factors=[
Factor(
name="测试",
signals_all=[
Signal('15分钟_倒0笔_方向_向上_其他_其他_20'), Signal('15分钟_倒0笔_长度_任意_其他_其他_0')
]
)
])
m, f = event.is_match(s)
assert not m and not f
event = Event(name="单测", operate=Operate.LO, factors=[
Factor(
name="测试",
signals_all=[
Signal('15分钟_倒0笔_方向_向下_其他_其他_0'), Signal('15分钟_倒0笔_长度_任意_其他_其他_0')
]
)
])
m, f = event.is_match(s)
assert not m and not f
event = Event.load(
{
"name": "开多",
"operate": "开多",
"signals_all": [
"1分钟_D1_涨跌停V230331_任意_任意_任意_0",
"1分钟_D0停顿分型_BE辅助V230106_看空_强_任意_0"
],
"signals_any": [],
"signals_not": [],
"factors": [
{
"name": "SMA#40多头",
"signals_all": [
"5分钟_D1#SMA#40MO10_BS辅助V230313_看多_任意_任意_0"
],
"signals_any": [],
"signals_not": []
}
]
}
)
assert len(event.get_signals_config()) == 3