【历史文档】高阶技巧-开箱实盘即用，批量测试因子的实盘策略模板
由qxiao创建，最终由small_q 被浏览 467 用户
更新
本文内容对应旧版平台与旧版资源，其内容不再适合最新版平台，请查看新版平台的使用说明
新版量化开发IDE（AIStudio）：
https://bigquant.com/wiki/doc/aistudio-aiide-NzAjgKapzW
新版模版策略：
https://bigquant.com/wiki/doc/demos-ecdRvuM1TU
新版数据平台：
https://bigquant.com/data/home
https://bigquant.com/wiki/doc/dai-PLSbc1SbZX
新版表达式算子：
https://bigquant.com/wiki/doc/dai-sql-Rceb2JQBdS
新版因子平台：
https://bigquant.com/wiki/doc/bigalpha-EOVmVtJMS5
导语
大家在做AI策略时，很纠结训练集用几年，预测集用几年？ST如何过滤？大盘风控如何加？如何只取部分数据进行训练？本文解决了AI策略研发中最常见的一些疑问，形成了一个模板策略，大家只需要把精力放到挖掘有效的因子即可。
本策略也可以作为因子挖掘的模板，通过回测的各项指标即可发现回测的这组因子在过去一段时间是否有效。
本文重点解决了以下几点：
1、在一般的AI策略中，训练集的范围是固定的，如何利用过去最近几年的数据作为训练集，来滚动预测未来一段时间的收益率？滚动回测结束后，如何计算整体回测的收益曲线、夏普指标等？
2、开发策略的时候，会碰到某些因子只想用于过滤数据，不作为训练的因子，该怎么处理？
3、如何在策略中结合大盘风控，股票最大持有天数限制等对策略进行改进？买入的股票突然变成了ST如何处理？
滚动回测
使用自定义运行模块，采用训练三年，预测一年的方式，回测时间段为2014年到2022年共8年多。自定义运行模块的主要代码如下：
其中，m1为训练集的代码列表模块，m2为预测集的代码列表模块，通过自定义运行模块，每次指定训练集时间为预测集的前三年
def bigquant_run(bq_graph, inputs):
test_years = ['2014','2015','2016','2017','2018','2019','2020','2021','2022']
parameters_list = []
#指定训练集和预测集时间段
for i in test_years:
train_start_date = str(int(i) -3)+'-01'+'-01'
train_end_date = str(int(i) - 1)+'-12'+'-31'
test_start_date = i+'-01'+'-01'
if i == test_years[-1]:
#最后一年手动指定回测结束时间
test_end_date = i+'-05'+'-11'
else:
test_end_date = i+'-12'+'-31'
parameters = {'m1.start_date':train_start_date,
'm1.end_date':train_end_date,
'm9.start_date':test_start_date,
'm9.end_date':test_end_date,
}
parameters_list.append({'parameters': parameters})
print(len(parameters_list), parameters_list)
def run(parameters):
try:
print(parameters)
return g.run(parameters)
except Exception as e:
print('ERROR --------', e)
return None
results = T.parallel_map(run, parameters_list, max_workers=4, remote_run=True, silent=False, backend="threading")
return results
滚动回测结果的评估指标
首先从自定义运行模块（m30）中获取回测结果并拼接在一起，然后通过empyrical包计算各种指标，具体代码如下：
df = pd.DataFrame()
#收集自定义运行模块的回测结果数据
for i in range(len(m30.result)):
tmp = m30.result[i]['m4'].raw_perf.read()
df = df.append(tmp[['returns','benchmark_period_return']])
#利用empyrical计算各种指标
import empyrical
def get_stats(returns, benchmark_period_return):
return_ratio = empyrical.cum_returns_final(returns)
annual_return_ratio = empyrical.annual_return(returns)
sharp_ratio = empyrical.sharpe_ratio(returns,0.035/252)
return_volatility = empyrical.annual_volatility(returns)
max_drawdown = empyrical.max_drawdown(returns)
benchmark_returns = (benchmark_period_return+1)/(benchmark_period_return+1).shift(1)-1
alpha, beta =empyrical.alpha_beta_aligned(returns, benchmark_returns)
return {
'return_ratio': return_ratio,
'annual_return_ratio': annual_return_ratio,
'beta': beta,
'alpha': alpha,
'sharp_ratio': sharp_ratio,
'return_volatility': return_volatility,
'max_drawdown': max_drawdown,
'收益回测比': abs(annual_return_ratio / max_drawdown)
}
d=get_stats(df['returns'], df['benchmark_period_return'])
df1=pd.DataFrame.from_dict(d,orient='index')
df1.T
输出效果如下：
同时，可以通过T.plot方法画出收益曲线：
利用因子进行数据过滤，但是因子不参与模型训练
可以采用两个输入特征列表模块的方法来实现，如下样例，在m3中输入需要参与训练的因子特征，输出连接到m5 StockRanker训练模块。另外一个输入特征列表模块填入过滤数据需要的特征，此模块输入为m3，然后输出给m15 基础特征抽取模块，在m21 数据过滤中就可以使用m6中的特征因子进行数据过滤了。
本样例中通过market_cap_float_0因子来选出小市值的股票：
回测过程结合大盘风控，ST股票处理，股票最大持有天数限制
大盘风控
此样例中我们首先计算出大盘5日跌幅：
然后在回测主函数中，判断触发风控则全部清仓：
today = data.current_dt.strftime('%Y-%m-%d')
# 按日期过滤得到今日的预测数据
ranker_prediction = context.ranker_prediction[
context.ranker_prediction.date == today]
if len(ranker_prediction)==0:
return
#大盘风控模块，读取风控数据
benckmark_risk=ranker_prediction['bm_0'].values[0]
#当risk为1时，市场有风险，全部平仓，不再执行其它操作
if benckmark_risk > 0:
for instrument in positions.keys():
context.order_target(context.symbol(instrument), 0)
return
买入后突然被ST
虽然可以通过股票过滤模块过滤掉买入当天ST的股票，但是偶尔也会碰到买入后在持仓过程中突然变成ST的场景，我们可以通过表instruments_CN_STOCK_A中股票的名称来判断是否变成ST了，如果发现变成ST了，直接卖出。
首先在m22数据源模块填入表名instruments_CN_STOCK_A，然后在m20输入特征列表中填入要获取的字段name
然后在主回测模块中判断股票名称是否含有‘ST’或者‘退’，如果有就需要卖出：
st_stock_list = []
for instrument in positions.keys():
try:
instrument_name = ranker_prediction[ranker_prediction.instrument==instrument].name.values[0]
# 如果股票状态变为了st或者退市 则卖出
if 'ST' in instrument_name or '退' in instrument_name:
if instrument in stock_sold:
continue
if data.can_trade(context.symbol(instrument)):
context.order_target(context.symbol(instrument), 0)
st_stock_list.append(instrument)
cash_for_sell -= positions[instrument]
except:
continue
if st_stock_list!=[]:
# print(today,'持仓出现st股/退市股',st_stock_list,'进行卖出处理')
stock_sold += st_stock_list
股票最大持有天数限制
如果不想股票持仓时间太长，可以从持仓信息中获取股票的建仓日期last_sale_date，然后再结合交易日历表判断持仓天数是否超限，主要代码如下：
current_stopdays_stock = []
positions_lastdate = {e.symbol:p.last_sale_date for e,p in context.portfolio.positions.items()}
# 不是建仓期（在前hold_days属于建仓期）
if not is_staging:
for instrument in positions.keys():
#如果上面的止盈止损已经卖出过了，就不要重复卖出以防止产生空单
if instrument in stock_sold:
continue
# 今天和上次交易的时间相隔hold_days就全部卖出
#使用交易天数
dt = pd.to_datetime(D.trading_days(end_date = today).iloc[-context.options['hold_days']].values[0])
if pd.to_datetime(positions_lastdate[instrument].strftime('%Y-%m-%d')) <= dt and data.can_trade(context.symbol(instrument)):
context.order_target_percent(context.symbol(instrument), 0)
current_stopdays_stock.append(instrument)
cash_for_sell -= positions[instrument]
if len(current_stopdays_stock)>0:
stock_sold += current_stopdays_stock
策略链接
https://bigquant.com/experimentshare/7d301cf59b88449da20c7a1d72a7472b
\