3 分钟快速生成代码
输入想法,AI 即刻生成可运行代码
在聚宽(JoinQuant)平台上进行 ETF 策略的长期回测时,确保回测结果的真实性和有效性至关重要。以下是实现长期回测、避免未来函数、合理设置交易成本以及处理调仓信号的完整指南与实战代码。
在传统的「前复权」回测模式中,历史价格是基于未来某一天的复权因子计算出来的,这会引入隐蔽的未来函数。对于长期回测,必须在 initialize 中开启动态复权(真实价格)模式:
set_option('use_real_price', True)
开启后,历史时刻交易使用当时的真实价格,而通过 history 或 attribute_history 获取的历史数据则是基于当前回测逻辑日期的前复权价格,从而完美避免未来函数。
在使用 get_price 或 get_bars 获取历史数据时,结束时间 end_date 或 end_dt 切勿超过当前逻辑时间 context.current_dt,否则会引入未来数据。建议在策略中开启避免未来数据模式:
set_option("avoid_future_data", True)
set_order_cost 进行单独设置。set_slippage 设置合理的滑点(如固定滑点或百分比滑点)以模拟真实市场。以下是一个实用的 ETF 动量轮动策略源码。策略在每月第一个交易日运行,计算候选 ETF 的过去 20 日收益率,买入表现最好的 ETF,并严格扣除交易成本与滑点。
# 导入聚宽函数库
import jqdata
import pandas as pd
import numpy as np
def initialize(context):
# 1. 设定沪深300作为基准
set_benchmark('000300.XSHG')
# 2. 开启真实价格模式(动态复权),防范未来函数
set_option('use_real_price', True)
# 开启避免未来数据检测
set_option("avoid_future_data", True)
# 3. 设置交易成本:ETF不收印花税,买入/卖出佣金设为万分之三,无最低五元限制
set_order_cost(OrderCost(open_tax=0, close_tax=0,
open_commission=0.0003, close_commission=0.0003,
close_today_commission=0, min_commission=0), type='fund')
# 4. 设置滑点:使用百分比滑点,单边 0.1%
set_slippage(PriceRelatedSlippage(0.002), type='fund')
# 5. 定义候选 ETF 资金池(例如:黄金ETF、沪深300ETF、纳指ETF、标普500ETF、中证500ETF)
g.etf_pool = [
'518880.XSHG', # 华安黄金易ETF
'510300.XSHG', # 华泰柏瑞沪深300ETF
'513100.XSHG', # 广发纳斯达克100ETF
'513500.XSHG', # 博时标普500ETF
'510500.XSHG' # 南方中证500ETF
]
# 6. 设置定时运行:每月第一个交易日 09:30 运行调仓
run_monthly(handle_etf_rotation, monthday=1, time='09:30', reference_security='000300.XSHG')
def handle_etf_rotation(context):
# 获取当前可用资金
cash = context.portfolio.available_cash
# 存储各 ETF 的动量得分(过去20个交易日的涨幅)
momentum_scores = {}
for etf in g.etf_pool:
# 获取过去21天的数据(不包含今天,避免未来函数)
hist = attribute_history(etf, count=21, unit='1d', fields=['close'], skip_paused=True)
if len(hist) < 21:
continue
# 计算 20 日收益率
price_now = hist['close'].iloc[-1]
price_past = hist['close'].iloc[0]
return_rate = (price_now - price_past) / price_past
momentum_scores[etf] = return_rate
if not momentum_scores:
return
# 找出涨幅最大的 ETF
best_etf = max(momentum_scores, key=momentum_scores.get)
best_score = momentum_scores[best_etf]
# 获取当前持仓
current_positions = list(context.portfolio.long_positions.keys())
# 调仓逻辑:如果表现最好的 ETF 涨幅大于 0,且当前未持有,则全仓轮动
if best_score > 0:
# 卖出不在最强池中的其他持仓
for etf in current_positions:
if etf != best_etf:
order_target(etf, 0)
log.info("卖出表现较弱的 ETF: %s" % etf)
# 买入最强的 ETF
if best_etf not in context.portfolio.long_positions:
# 重新获取卖出后的可用现金
available_cash = context.portfolio.available_cash
order_value(best_etf, available_cash)
log.info("买入最强动量 ETF: %s,20日收益率: %.2f%%" % (best_etf, best_score * 100))
else:
# 如果所有 ETF 动量均为负,则空仓避险
for etf in current_positions:
order_target(etf, 0)
log.info("市场走弱,所有候选 ETF 动量为负,清仓避险。")
在进行 5 年以上的长期回测时,市场会经历单边牛市、单边熊市、震荡市等不同阶段。建议在回测结束后,利用聚宽的归因分析和风险指标进行多维度评估: