-
Notifications
You must be signed in to change notification settings - Fork 343
/
signal_return_ic_analysis.py
130 lines (95 loc) · 4.97 KB
/
signal_return_ic_analysis.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
# encoding: utf-8
import numpy as np
import pandas as pd
from jaqs.data import DataView
from jaqs.data import RemoteDataService
from jaqs.research import SignalDigger
import jaqs.util as jutil
from config_path import DATA_CONFIG_PATH
data_config = jutil.read_json(DATA_CONFIG_PATH)
dataview_folder = '../../output/prepared/test_signal'
def save_dataview():
ds = RemoteDataService()
ds.init_from_config(data_config)
dv = DataView()
props = {'start_date': 20150101, 'end_date': 20171001, 'universe': '000300.SH',
'fields': 'volume,turnover,float_mv,pb,total_mv',
'freq': 1}
dv.init_from_config(props, ds)
dv.prepare_data()
# for convenience to check limit reachers
dv.add_formula('limit_reached', 'Abs((open - Delay(close, 1)) / Delay(close, 1)) > 0.095', is_quarterly=False)
dv.add_formula('random', 'StdDev(volume, 20)', is_quarterly=False)
dv.add_formula('momentum', 'Return(close_adj, 20)', is_quarterly=False)
# dv.add_formula('size', '', is_quarterly=False)
dv.save_dataview(dataview_folder)
def analyze_signal():
# --------------------------------------------------------------------------------
# Step.1 load dataview
dv = DataView()
dv.load_dataview(dataview_folder)
# --------------------------------------------------------------------------------
# Step.2 calculate mask (to mask those ill data points)
trade_status = dv.get_ts('trade_status')
mask_sus = trade_status == u'停牌'.encode('utf-8')
df_index_member = dv.get_ts('index_member')
mask_index_member = ~(df_index_member > 0)
dv.add_formula('limit_reached', 'Abs((open - Delay(close, 1)) / Delay(close, 1)) > 0.095', is_quarterly=False)
df_limit_reached = dv.get_ts('limit_reached')
mask_limit_reached = df_limit_reached > 0
mask_all = np.logical_or(mask_sus, np.logical_or(mask_index_member, mask_limit_reached))
# --------------------------------------------------------------------------------
# Step.3 get signal, benchmark and price data
# dv.add_formula('illi_daily', '(high - low) * 1000000000 / turnover', is_quarterly=False)
# dv.add_formula('illi', 'Ewma(illi_daily, 11)', is_quarterly=False)
# dv.add_formula('size', 'Log(float_mv)', is_quarterly=False)
# dv.add_formula('value', '-1.0/pb', is_quarterly=False)
# dv.add_formula('liquidity', 'Ts_Mean(volume, 22) / float_mv', is_quarterly=False)
dv.add_formula('divert', '- Correlation(vwap_adj, volume, 10)', is_quarterly=False)
signal = dv.get_ts('divert').shift(1, axis=0) # avoid look-ahead bias
price = dv.get_ts('close_adj')
price_bench = dv.data_benchmark
# Step.4 analyze!
my_period = 5
obj = SignalDigger(output_folder='../../output/test_signal',
output_format='pdf')
obj.process_signal_before_analysis(signal, price=price,
mask=mask_all,
n_quantiles=5, period=my_period,
benchmark_price=price_bench,
)
res = obj.create_full_report()
# import cPickle as pickle
# pickle.dump(res, open('_res.pic', 'w'))
def analyze_event():
# --------------------------------------------------------------------------------
# Step.1 load dataview
dv = DataView()
dv.load_dataview(dataview_folder)
# --------------------------------------------------------------------------------
# Step.2 calculate mask (to mask those ill data points)
trade_status = dv.get_ts('trade_status')
mask_sus = trade_status == u'停牌'.encode('utf-8')
df_index_member = dv.get_ts('index_member')
mask_index_member = ~(df_index_member > 0)
dv.add_formula('limit_reached', 'Abs((open - Delay(close, 1)) / Delay(close, 1)) > 0.095', is_quarterly=False)
df_limit_reached = dv.get_ts('limit_reached')
mask_limit_reached = df_limit_reached > 0
mask_all = np.logical_or(mask_sus, np.logical_or(mask_index_member, mask_limit_reached))
# --------------------------------------------------------------------------------
# Step.3 get signal, benchmark and price data
dv.add_formula('new_high', 'close_adj >= Ts_Max(close_adj, 300)', is_quarterly=False)
dv.add_formula('new_high_delay', 'Delay(Ts_Max(new_high, 300), 1)', is_quarterly=False)
dv.add_formula('sig', 'new_high && (! new_high_delay)', is_quarterly=False)
signal = dv.get_ts('sig').shift(0, axis=0) # avoid look-ahead bias
price = dv.get_ts('close_adj')
price_bench = dv.data_benchmark
# Step.4 analyze!
obj = SignalDigger(output_folder=jutil.join_relative_path('../output'),
output_format='pdf')
obj.create_binary_event_report(signal, price, mask_all, 5, price_bench, periods=[5, 20, 40])
# import cPickle as pickle
# pickle.dump(res, open('_res.pic', 'w'))
if __name__ == "__main__":
save_dataview()
analyze_signal()