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
| def get_previous_N_trading_date(date,counts=1,exchange='SHSE'):
"""
获取end_date前N个交易日,end_date为datetime格式，不包括date日期
:param date：目标日期
:param counts：历史回溯天数，默认为1，即前一天
"""
date = pd.Timestamp(date)
previous_N_trading_date = get_trading_dates(exchange=exchange, start_date=date-datetime.timedelta(days=max(counts+30,counts*3)), end_date=date)[-counts-1]
return previous_N_trading_date
def get_normal_stocks(date,new_days=365):
"""
获取目标日期date的A股代码（剔除停牌股、ST股、次新股（365天））
:param date：目标日期
:param new_days:新股上市天数，默认为365天
"""
if isinstance(date,str) and len(date)==10:
date = datetime.datetime.strptime(date,"%Y-%m-%d")
elif isinstance(date,str) and len(date)>10:
date = datetime.datetime.strptime(date,"%Y-%m-%d %H:%M:%S")
df_code = get_instrumentinfos(sec_types=SEC_TYPE_STOCK, fields='symbol, listed_date, delisted_date', df=True)
all_stocks = [code for code in df_code[(df_code['listed_date']<=date-datetime.timedelta(days=new_days))&(df_code['delisted_date']>date+datetime.timedelta(days=31))].symbol.to_list() if code[:6]!='SHSE.9' and code[:6]!='SZSE.2']
history_ins = get_history_instruments(symbols=all_stocks, start_date=date, end_date=date, fields='symbol,sec_level, is_suspended', df=True)
all_stocks = list(history_ins[(history_ins['sec_level']==1) & (history_ins['is_suspended']==0)]['symbol'])
all_stocks = [code for code in all_stocks if code[:8]!='SHSE.688']
all_stocks_str = ','.join(all_stocks)
return all_stocks,all_stocks_str
def history_new(context,security,frequency,start_time,end_time,fields,skip_suspended=True,fill_missing=None,adjust=ADJUST_PREV, adjust_end_time='backtest_end_time', df=True):
"""
分区间获取数据（以避免超出数据限制）(start_time和end_date为字符串,fields需包含eob和symbol,单字段)
:param ：参数同history()参数一致，adjust_end_time默认为回测结束时间：context.backtest_end_time
"""
if adjust_end_time=='backtest_end_time':
adjust_end_time = context.backtest_end_time
Data = pd.DataFrame()
if frequency=='1d':
trading_date = pd.Series(get_trading_dates(exchange='SZSE', start_date=start_time, end_date=end_time))
else:
trading_date = history('SHSE.000300', frequency=frequency, start_time=start_time, end_time=end_time, fields='eob', skip_suspended=skip_suspended, fill_missing=fill_missing, adjust=adjust, adjust_end_time=adjust_end_time, df=df)
trading_date = trading_date['eob']
space = 5
if len(trading_date)<=space:
Data = history(security, frequency=frequency, start_time=start_time, end_time=end_time, fields=fields, skip_suspended=skip_suspended, fill_missing=fill_missing, adjust=adjust, adjust_end_time=adjust_end_time, df=df)
else:
for n in range(int(np.ceil(len(trading_date)/space))):
start = n*space
end = start+space-1
if end>=len(trading_date):
data = history(security, frequency=frequency, start_time=trading_date.iloc[start], end_time=trading_date.iloc[-1], fields=fields, skip_suspended=skip_suspended, fill_missing=fill_missing, adjust=adjust, adjust_end_time=adjust_end_time, df=df)
else:
data = history(security, frequency=frequency, start_time=trading_date.iloc[start], end_time=trading_date.iloc[end], fields=fields, skip_suspended=skip_suspended, fill_missing=fill_missing, adjust=adjust, adjust_end_time=adjust_end_time, df=df)
if len(data)>=33000:
print('请检查返回数据量，可能超过系统限制，缺少数据！！！！！！！！！！')
Data = pd.concat([Data,data])
Data.drop_duplicates(keep='first',inplace=True)
if len(Data)>0:
Data = Data.set_index(['eob','symbol'])
Data = Data.unstack()
Data.columns = Data.columns.droplevel(level=0)
return Data
def on_backtest_finished(context, indicator):
print('*'*50)
print('回测已完成，请通过右上角“回测历史”功能查询详情。')
|
