如何在 JoinQuant 中围绕财报公告日构建事件驱动型交易策略？

这是一个非常经典的事件驱动策略（Event-Driven Strategy），在量化交易中通常被称为盈余公告后漂移策略（PEAD, Post-Earnings Announcement Drift）。

该策略的核心逻辑是：当公司发布好于预期的财报（如净利润大幅增长）后，股价往往不会一步到位，而是在接下来的几天或几周内持续上涨。

策略设计思路

1. 事件触发：每天检查前一个交易日发布财报的公司。
2. 筛选条件：
   • 业绩增长：净利润同比增长率（Year-on-Year）超过一定阈值（例如 30%）。
   • 基本面过滤：剔除亏损股、剔除 ST 股、剔除停牌股。
   • 估值过滤：PE（市盈率）在合理范围内（例如 0 < PE < 60），防止买入估值过高的股票。
3. 交易逻辑：
   • 买入：在财报发布后的第一个交易日开盘买入。
   • 卖出：持有固定天数（如 20 天）后卖出，或者当股票不再满足持仓条件时卖出。
4. 资金管理：等权重买入，设置最大持仓数量。

策略代码实现

以下是一个完整的 JoinQuant 策略代码。你可以直接复制到聚宽的回测环境中运行。

# -*- coding: utf-8 -*-
from jqdata import *

def initialize(context):
    """
    初始化函数，设定基准、手续费、全局变量等
    """
    # 设定沪深300作为基准
    set_benchmark('000300.XSHG')
    # 开启动态复权模式(真实价格)
    set_option('use_real_price', True)
    # 过滤掉order系列API产生的比error级别低的log
    log.set_level('order', 'error')
    
    # --- 策略参数设置 ---
    g.growth_threshold = 30.0  # 净利润同比增长率阈值(%)
    g.pe_max = 60              # 最大市盈率
    g.pe_min = 0               # 最小市盈率
    g.max_hold_days = 20       # 最大持仓天数
    g.max_stocks = 10          # 最大持仓股票数量
    
    # 记录持仓天数的字典 {security: days}
    g.hold_days = {} 
    
    # 每天开盘前运行
    run_daily(before_market_open, time='before_open')
    # 每天开盘时运行
    run_daily(market_open, time='open')

def before_market_open(context):
    """
    开盘前筛选股票
    """
    # 获取当前回测日期的前一个交易日
    # 我们寻找昨天发布财报的股票，今天开盘买入
    prev_date = context.previous_date
    
    # 1. 查询财务数据
    # 查询条件：
    # - 昨天发布的财报 (pubDate == prev_date)
    # - 净利润同比增长率 > 阈值
    q = query(
        indicator.code,
        indicator.inc_net_profit_year_on_year, # 净利润同比增长率
        valuation.pe_ratio
    ).filter(
        indicator.pubDate == prev_date,
        indicator.inc_net_profit_year_on_year > g.growth_threshold,
        valuation.pe_ratio > g.pe_min,
        valuation.pe_ratio < g.pe_max
    ).order_by(
        indicator.inc_net_profit_year_on_year.desc() # 按增长率降序排列
    )
    
    df = get_fundamentals(q)
    
    # 2. 过滤 ST、停牌、退市股票
    current_data = get_current_data()
    target_list = []
    
    if not df.empty:
        raw_list = list(df['code'])
        for stock in raw_list:
            # 过滤 ST 和 停牌
            if not current_data[stock].is_st and \
               not current_data[stock].paused and \
               '退' not in current_data[stock].name:
                target_list.append(stock)
    
    # 限制每日买入候选名单长度，避免资金过于分散
    g.buy_list = target_list[:5] 
    
    # 更新持仓天数
    for stock in list(g.hold_days.keys()):
        g.hold_days[stock] += 1

def market_open(context):
    """
    开盘交易逻辑
    """
    # --- 卖出逻辑 ---
    # 遍历当前持仓
    for stock in list(context.portfolio.positions.keys()):
        # 如果持仓天数超过设定值，或者股票停牌/ST（风控），则卖出
        days_held = g.hold_days.get(stock, 0)
        
        should_sell = False
        
        # 条件1: 持仓时间到期
        if days_held >= g.max_hold_days:
            should_sell = True
            log.info("股票 %s 持仓达到 %d 天，止盈/止损卖出" % (stock, days_held))
            
        # 条件2: 也可以加入止损逻辑，例如亏损超过10%卖出
        # cost = context.portfolio.positions[stock].avg_cost
        # price = context.portfolio.positions[stock].price
        # if price < cost * 0.9:
        #     should_sell = True
        
        if should_sell:
            order_target_value(stock, 0)
            if stock in g.hold_days:
                del g.hold_days[stock]

    # --- 买入逻辑 ---
    # 计算当前还可以买入多少只股票
    current_positions_count = len(context.portfolio.positions)
    available_slots = g.max_stocks - current_positions_count
    
    if available_slots > 0 and len(g.buy_list) > 0:
        # 计算每只股票分配的资金
        # 注意：这里简单处理，用可用资金除以剩余仓位。实际中可能需要更复杂的资金管理。
        cash = context.portfolio.available_cash
        if cash > 0:
            # 避免资金过小无法买入
            cash_per_stock = cash / available_slots
            
            for stock in g.buy_list:
                if current_positions_count >= g.max_stocks:
                    break
                
                if stock not in context.portfolio.positions:
                    order_value(stock, cash_per_stock)
                    # 记录买入，初始化持仓天数为0
                    g.hold_days[stock] = 0
                    current_positions_count += 1
                    log.info("财报利好，买入: %s" % stock)

def after_code_changed(context):
    """
    模拟交易中更新代码后执行
    """
    g.growth_threshold = 30.0
    g.max_hold_days = 20

代码关键点解析

1. 获取财报发布日期 (indicator.pubDate)：

   • 这是策略的核心。get_fundamentals 查询中的 indicator.pubDate 字段代表财报的实际公告日期。
   • filter(indicator.pubDate == context.previous_date)：这行代码的意思是“筛选出昨天刚刚发布财报的股票”。因为我们在回测中通常是在 T+1 日（今天）根据 T 日（昨天）晚间发布的消息进行交易。

2. 财务指标选择 (indicator.inc_net_profit_year_on_year)：

   • 我们使用了 inc_net_profit_year_on_year（归属于母公司所有者的净利润同比增长率）。这是衡量公司业绩增长最直接的指标。
   • 你也可以替换为 inc_revenue_year_on_year（营业收入增长率）或 roe（净资产收益率）等。

3. 持仓周期管理 (g.hold_days)：

   • PEAD 效应通常有时效性。我们使用一个字典 g.hold_days 来记录每只股票持有了多少天。
   • 在 before_market_open 中每天加 1。
   • 在 market_open 中检查，如果超过 g.max_hold_days 就卖出。这是一种简单的时间止盈止损策略。

4. 防坑指南（未来函数）：

   • 聚宽的 get_fundamentals 默认查询的是 date 参数指定日期当时可见的数据。
   • 代码中使用 context.previous_date 作为筛选条件，模拟的是在今天开盘前，看到了昨天收盘后发布的公告，因此不存在未来函数。

如何优化这个策略？

如果你想进一步提高策略的收益风险比，可以考虑以下优化方向：

1. 结合预期数据：
   • 单纯的增长不一定是利好，如果增长低于分析师预期，股价反而会跌。
   • 可以使用聚宽的分析师预期数据（需购买数据权限），计算 SUE（Standardized Unexpected Earnings，标准化预期外盈利），只买入实际业绩远超预期的股票。
2. 叠加技术指标：
   • 在财报发布后，要求股价必须站上 20 日均线才买入，避免买入“见光死”（利好出尽下跌）的股票。
3. 区分财报类型：
   • 年报和一季报通常在 4 月密集发布，可能会导致资金分散。可以针对不同季度的财报设置不同的阈值。
4. 市场环境过滤：
   • 在大盘（如沪深300指数）处于下跌趋势时，即使个股财报好也可能补跌，此时可以空仓或降低仓位。

Q&A

Q: 为什么使用 indicator.pubDate 而不是 statDate?
A: statDate 是财报的统计截止日期（如 2023-12-31），而 pubDate 是财报实际对外公布的日期（如 2024-03-15）。我们在做交易时，必须等到信息公开（pubDate）后才能根据该信息交易。如果使用 statDate 进行回测，会导致严重的“未来函数”问题（即在 1 月份就用到了 3 月份才发布的数据）。

Q: 这个策略适合什么频率的回测？
A: 适合 按天 (Daily) 回测。因为财报数据是按天更新的，盘中分钟级别的波动对财报选股逻辑影响较小。

Q: 为什么有时候选不出股票？
A: 财报发布有季节性。上市公司通常在 4 月（年报+一季报）、8 月（中报）、10 月（三季报）密集发布财报。在其他月份（如 5、6、7 月），可能连续很多天都没有公司发布财报，这是正常现象。