3 分钟快速生成代码
输入想法,AI 即刻生成可运行代码
在量化交易中,**未来函数(Future Function)**是回测最大的敌人。如果在回测中无意间引入了“未来的数据”(例如在早上9:30就获取了当天的收盘价,或者在财报实际发布前就使用了其财务指标),会导致回测收益率异常虚高,但在实盘中却会产生严重亏损。
本文将详细介绍如何在聚宽(JoinQuant)平台中通过 set_option("avoid_future_data", True) 开启防未来数据模式,并针对盘前选股和盘中低吸策略,解析如何正确处理收盘价、财报发布日期和信号时点。
聚宽提供了系统级的防未来数据检测功能。你只需要在策略的 initialize 初始化函数中加入以下代码:
def initialize(context):
# 开启避免未来数据模式
set_option("avoid_future_data", True)
context.current_dt)之后的数据(例如在 10:00 获取 15:00 的分钟 K 线),系统会直接抛出 FutureDataError 异常并停止运行。get_call_auction 集合竞价数据),系统会自动将返回结果中的未来数据剔除,而不会中断策略。在盘中(如 14:30)运行低吸策略时,绝对不能直接获取“当天的收盘价”。
get_price 或 get_bars 获取当天的日线 close,此时拿到的往往是尚未确定的价格,甚至在某些不规范的写法下会拿到 15:00 的最终收盘价。history 或 attribute_history 获取昨天及以前的日线数据(默认不包含当天)。get_current_data() 获取当前切片数据,或使用 get_bars 获取当前分钟的最新 Bar。# 安全获取昨日收盘价与当前最新价
def handle_data(context, data):
security = '000001.XSHE'
# 1. 获取昨日收盘价(安全,无未来)
hist = attribute_history(security, 1, '1d', ['close'])
last_close = hist['close'][-1]
# 2. 获取当前最新价(安全,无未来)
current_data = get_current_data()
current_price = current_data[security].last_price
财务数据是未来函数的重灾区。上市公司披露财报通常有延时,例如 2023 年第三季度的财报,可能在 2023 年 10 月 25 日才正式发布。如果在 10 月 1 日的回测中就使用了该季报数据,就引入了未来函数。
date 参数:调用 get_fundamentals 时,传入 date=context.current_dt(或默认不传,系统会自动处理为当前时间的前一天)。系统会严格查找上市公司在**该日期之前(含当天)**实际已经发布披露的最新财报,确保无未来数据。statDate 参数:如果直接指定 statDate='2023q3',查询的是该季度的财务报表。在实际未披露该报表的历史回测时点使用此参数,会直接带入未来数据。若必须使用,请配合 get_history_fundamentals 并指定 watch_date 参数。# 安全查询财务数据示例
def before_trading_start(context):
# 查询当前时点已经公开披露的市值和估值数据
q = query(valuation.code, valuation.pe_ratio).filter(valuation.pe_ratio < 15)
# 传入当前日期,确保不获取未披露的未来财报
df = get_fundamentals(q, date=context.current_dt.date())
策略的运行频率(天/分钟/Tick)和执行时间点决定了数据的可用性:
handle_data 中,data 默认是昨天的天数据。如果策略在 9:30 运行,此时无法获取当天的任何日线级 K 线数据。data 是上一分钟的分钟数据。例如在 09:31:00 运行,拿到的 data 是 09:30:00~09:30:59 的 Bar。不要试图在 09:31 去获取 09:35 的数据。以下是一个结合了盘前选股与盘中低吸的安全策略模板,严格开启了 avoid_future_data 并规范了数据获取时点:
import jqdata
def initialize(context):
# 1. 开启防未来数据模式(系统级安全保障)
set_option("avoid_future_data", True)
set_option('use_real_price', True)
set_benchmark('000300.XSHG')
# 2. 设定定时任务
# 盘前 09:00 进行选股
run_daily(before_market_open, time='09:00')
# 盘中 14:50 进行低吸/平仓决策
run_daily(market_decisions, time='14:50')
def before_market_open(context):
# 盘前选股:严格使用 date=context.previous_date,避免获取当天未公布的财报或未开盘的数据
q = query(valuation.code).filter(
valuation.market_cap > 100, # 市值大于100亿
valuation.pe_ratio < 20 # 低市盈率
)
# 严格限制在昨日收盘后可见的数据
df = get_fundamentals(q, date=context.previous_date)
g.target_list = list(df['code'])[:10]
log.info(f"今日待监控股票池: {g.target_list}")
def market_decisions(context):
# 盘中决策:14:50 运行
for security in g.target_list:
# 获取昨日收盘价(安全,不包含今天)
hist = attribute_history(security, 1, '1d', ['close'])
if len(hist) == 0: continue
last_close = hist['close'][-1]
# 获取当前最新切片价格(安全,无未来)
current_data = get_current_data()
if current_data[security].paused:
continue
current_price = current_data[security].last_price
# 低吸信号:当前价格相比昨日收盘价下跌超过 3%
if current_price < last_close * 0.97:
cash = context.portfolio.available_cash
if cash > 10000:
order_value(security, 10000)
log.info(f"触发低吸信号,买入 {security},当前价: {current_price}")
在聚宽平台编写策略时,请时刻牢记:“只用过去,不看未来”。通过开启 set_option("avoid_future_data", True),配合规范的 get_fundamentals(..., date=context.current_dt) 和 attribute_history 调用,可以彻底杜绝未来数据泄漏,让你的回测结果经得起实盘的检验。