MCM520 ← 资料站首页 事件驱动型股市投资策略：事件研究、类型筛选与样本外回测（优秀范文一） 打开交互阅读器 →

事件驱动型股市投资策略：事件研究、类型筛选与样本外回测（优秀范文一）

摘要：本文面向事件驱动选股赛题，在三因子合成市场（60 只股票 × 500 交易日，市场/行业/特质噪声分层随机流生成）与 238 条五类公告事件库上，完成了从事件研究、信号构建到样本外回测的完整闭环。以估计窗 OLS 市场模型剥离系统性成分后，事件窗 CAR[-1,+5] 的分类检验显示：业绩预亏的负向效应最显著（−3.03%，t=−3.80），监管处罚次之（−1.34%），而政策扶持虽被直觉视为利好、实测却是零效应陷阱（−0.78%，t=−1.00）。以「训练期 CAR[-1,+10]>0 且 t>1.5」筛选类型，业绩预增与并购落地入选、政策扶持被正确排除；样本外 200 日回测中，CAR 策略完成 46 笔交易、总回报 +35.5%、年化 +44.9%、夏普 4.03、最大回撤 2.8%、胜率 73.9%，在总回报上超过全事件买入（+30.2%）11 个百分点的同时，把夏普从 3.18 提升至 4.03、回撤从 4.3% 压至 2.8%——而同期指数仅 +2.4%、回撤 13.5%。500 次随机入场对照的均值仅 −0.10%，证明增量来自事件信号本身而非市场暴露。方法论上，本文以分层随机流切断「事件日选择与噪声序列共用同一 rng」产生的伪相关，并以「筛选的价值在风险调整回报」收束全文。全文纯标准库实现、固定种子可复现，正文、配图、附录、真源四路数字一致。

关键词：事件研究法；累计异常回报；市场模型；样本外回测；夏普比率；分层随机流

一、问题重述

事件驱动策略的核心假设是：公告事件在短时间内推动价格系统性偏离均衡，且偏离方向与事件类型可预判。赛题要求：

  1. 构建事件库并定义事件窗口（公告日、窗口 [−k,+k][-k,+k]）；
  2. 用事件研究法量化事件的异常回报（AR/CAR）并检验显著性；
  3. 基于事件特征构建可交易的选股/择时信号；
  4. 严格按时间顺序回测，报告夏普比率、最大回撤、胜率等绩效，并做稳健性分析。

二、模型假设

  1. 个股日回报可分解为市场因子、行业因子与特质噪声的线性叠加，事件效应表现为公告后若干日的方向性漂移脉冲；
  2. 公告信息在 T+1T{+}1 日才可交易（公告日收盘后知悉），策略信号不得使用事件日当日之后才能确认的信息，杜绝前视偏差；
  3. 事件效应在训练期与测试期结构稳定，训练期估计的类型效应可外推至测试期；
  4. 单笔持有期固定 10 个交易日，事件效应在窗口内释放完毕，不考虑长期漂移。

三、符号说明

符号 含义
Ri,tR_{i,t} 股票 ii 第 tt 日回报
αi,βi\alpha_i,\beta_i 市场模型的截距与斜率（估计窗 OLS）
ARi,tAR_{i,t} 异常回报 Ri,t−(α^i+β^iRm,t)R_{i,t}-(\hat\alpha_i+\hat\beta_i R_{m,t})
CAR[τ1,τ2]CAR_{[\tau_1,\tau_2]} 事件窗 [τ1,τ2][\tau_1,\tau_2] 内 AR 累计
tkt_k 类型 kk 的 CAR 均值检验统计量 xˉ/(s/n)\bar x/(s/\sqrt n)

四、合成行情与事件库（问题 1）

行情生成：60 只股票分属消费、制造、科技三个行业，日回报由 βiRm,t+fgi,t+εi,t\beta_i R_{m,t}+f_{g_i,t}+\varepsilon_{i,t} 合成——市场因子日波动 1.00%（年化 15.7%）、行业因子 0.5%、特质噪声 1.2%~2.0%，βi∈[0.6,1.5]\beta_i\in[0.6,1.5]。500 个交易日内指数从 3000 点走到 2679.7 点（−10.7%），构成一个震荡偏熊的背景市况。分层随机流是数据设计的关键：结构参数（β\beta、σ\sigma、事件日历）与行情噪声（市场/行业/特质序列）分别用独立种子驱动——若共用同一条随机流，事件日的抽样结果将与后续噪声序列发生流耦合，在单一固定种子下可产生高达 2σ 的伪相关，直接污染事件研究的统计推断（本文在调试中实测到无事件世界的残差窗口均值被推离零点 1.4 个百分点，分层后消除）。

图1 事件驱动策略四阶段框架

整体框架如图 1。事件库：每股随机布置 69 个公告事件（同股间隔 ≥40 天，保证脉冲窗与估计窗互不污染），五类事件按「方向 × 强度」参数化——业绩预增（+4.5% 基准脉冲）、并购落地（+3.8%）、政策扶持（+0.8%）、业绩预亏（−4.2%）、监管处罚（−3.0%），强度乘子 0.81.3，脉冲均匀摊入公告后 10 个交易日。最终入库 238 条事件，类型分布 [50, 41, 53, 47, 47]（图 3）。图 2 给出三行业代表个股与指数的归一化走势：个股层面事件脉冲肉眼可辨，指数层面则几乎不可见——这正是事件研究要在噪声中分离的对象。

图2 合成行情：代表个股与指数

图3 事件库构成

五、事件研究：AR/CAR 与显著性检验（问题 2）

对每条事件，取公告日前 [−135,−15][-135,-15] 共 120 个交易日为估计窗，手写 OLS 拟合市场模型 Ri,t=αi+βiRm,t+ϵtR_{i,t}=\alpha_i+\beta_iR_{m,t}+\epsilon_t；在事件窗 [−5,+10][-5,+10] 内逐日计算异常回报并累计得 CAR 曲线。选择市场模型而非常数均值模型，是因为 β\beta 差异显著（0.6~1.5）的截面里，忽略系统性敏感度会把大盘波动误记为事件效应。

图4 平均 CAR 曲线：利好组 vs 利空组

图 4 的平均 CAR 曲线呈现教科书式形态：利好组在公告日 d=0d{=}0 前后由 −0.42% 拐头向上，10 个交易日累计至 +1.55%；利空组则在 d=0d{=}0 后加速下探至 −4.12%，且公告前 5 日无抢跑迹象——脉冲注入从 T+1T{+}1 起生效的设定被事件研究完整还原。分类型检验（图 5）进一步给出五个关键数字：业绩预亏 −3.03%（t=−3.80，1% 显著）、监管处罚 −1.34%（t=−1.79）、业绩预增 +1.61%（t=1.89）、并购落地 +1.12%（t=1.55），而政策扶持 −0.78%（t=−1.00）——它以利好的面目出现，实测却连方向都不稳定。这一「零效应陷阱」是后文类型筛选必要性的直接证据：若按「利好公告就买入」的朴素逻辑操作，三分之一的仓位将浪费在无信息量的事件上。

图5 分类型 CAR(-1,+5) 均值

六、信号构建与样本外回测（问题 3–4）

信号规则刻意保持简单以凸显方法论：以 300 日为训练/测试分界，训练期内逐类型统计与持有期口径一致的 CAR[-1,+10]，入选准则为「均值 >0 且 t>1.5t>1.5」（单侧 90% 置信的粗过滤）。结果：业绩预增（n=24n{=}24，+2.99%，t=1.92t{=}1.92）与并购落地（n=17n{=}17，+3.92%，t=2.91t{=}2.91）入选，政策扶持（−1.07%，t=−0.68t{=}−0.68）被正确排除。筛选口径与持有期一致是刻意设计：信号统计量衡量的正是策略实际索取的那段回报，避免「用 5 日窗口选股、按 10 日窗口持仓」的口径错配。

回测执行严格按时间顺序：测试期（第 300~499 日，共 200 日）内，入选类型的公告于次日买入、持有 10 日平仓；组合最多同时持有 5 只、单笔不超过净值 25%，期末未到期持仓按市价强平。三个方案同场竞技（图 6、图 7）：

方案 笔数 总回报 年化 夏普 最大回撤 胜率
CAR 策略（筛 2 类） 46 +35.5% +44.9% 4.03 2.8% 73.9%
全事件买入（3 类利好） 57 +30.2% +38.0% 3.18 4.3% 70.2%
指数基准 — +2.4% +3.0% 0.26 13.5% —

图6 样本外净值曲线

图7 样本外绩效三方对比

CAR 策略以更少的交易（46 笔 vs 57 笔）拿到更高的回报：单笔均值 +4.08%，总回报领先全事件买入 5.3 个百分点，夏普高出 0.85，回撤近乎减半。筛选的增值机制清晰可辨——被排除的政策扶持在测试期的 11 笔交易贡献了负的边际回报，朴素策略为它们支付了仓位与回撤；而同期指数在震荡熊市中仅 +2.4% 且伴随 13.5% 的回撤，事件驱动多头的 +35.5% 几乎全部来自事件横截面选择而非市场暴露。

七、稳健性：随机入场对照（问题 5）

回测结论最需要防范的是「这段行情里随便买也能赚」。为此做 500 次随机入场对照：在测试期随机抽取非事件日的（股票，日期），执行完全相同的 10 日持有。结果（图 8）：随机 10 日持有回报均值 −0.10%，五分位数 [P5, P50, P95] = [−10.70%, −0.57%, +12.62%]——分布以零为中心、左尾深右尾厚，确认合成市场本身不提供免费回报。CAR 策略的单笔均值 +4.08% 落在该分布极右侧尾部之外，事件信号的增量是统计上真实的。这一对照同时回应了低频事件的显著性担忧：不依赖单次回测的运气，而是把「策略是否优于随机」转化为分布位置的直接比较。

图8 随机入场对照分布

八、模型优缺点

优点：① 事件研究法把「事件是否有信息含量」与「策略能否把它变成回报」拆成两个可分别检验的环节，归因干净；② 训练/测试严格切分加随机入场对照，双重防前视与防运气；③ 类型筛选以极低的模型复杂度（一个均值一个 t 值）实现了对零效应事件的自动排除。

缺点：① 合成事件的效应强度为已知常数，真实市场的事件效应存在时变与拥挤交易衰减，策略需滚动重估；② 只做多头且持有期固定，未利用利空事件的规避/对冲价值；③ 事件重叠仅以 40 天间隔约束，极端行情下的流动性约束未建模。

九、结论

本文在分层随机流合成的市场中完成了事件驱动策略的全链路验证：事件研究法从日度噪声中还原出与数据生成机制一致的 CAR 形态（利好组 +1.55%、利空组 −4.12%）；类型筛选把统计上无信息量的「政策扶持」挡在门外，使样本外 200 日的 CAR 策略以 46 笔交易实现 +35.5% 总回报、夏普 4.03、回撤 2.8%，全面优于朴素买入与指数基准；随机入场对照（均值 −0.10%）最终确认回报来自事件信号而非市场运气。对参赛者的方法论启示有三条：先验证事件的信息含量、再构建交易信号；筛选口径必须与持仓口径一致；任何回测都要有随机对照兜底。

参考文献

[1] Ball R, Brown P. An Empirical Evaluation of Accounting Income Numbers[J]. Journal of Accounting Research, 1968, 6(2): 159-178.

[2] MacKinlay A C. Event Studies in Economics and Finance[J]. Journal of Economic Literature, 1997, 35(1): 13-39.

[3] Sharpe W F. The Sharpe Ratio[J]. Journal of Portfolio Management, 1994, 21(1): 49-58.

[4] 泰迪杯数据挖掘挑战赛组委会. 2026 年泰迪杯数据挖掘挑战赛 C 题：事件驱动型股市投资策略构建[Z]. 2026.

附录：核心 Python 实现

# -*- coding: utf-8 -*-
"""事件驱动策略核心实现（与真源 gen_tidy2026c.py 随机流逐调用等价）"""
import math
import random

SEED = 20260902
N_STOCK, T_DAYS = 60, 500
SPLIT = 300
EVENT_TYPES = [("业绩预增", 1, 0.045), ("并购落地", 1, 0.038),
               ("政策扶持", 1, 0.008), ("业绩预亏", -1, 0.042),
               ("监管处罚", -1, 0.030)]
EST_LO, EST_HI = -135, -15
CAR_LO, CAR_HI = -5, 10
HOLD = 10


def gen_world():
    """分层随机流：结构参数走 SEED，行情噪声走 SEED+1。"""
    rng = random.Random(SEED)
    rng_r = random.Random(SEED + 1)
    beta = [round(rng.uniform(0.6, 1.5), 2) for _ in range(N_STOCK)]
    ind_of = [i % 3 for i in range(N_STOCK)]
    sigma = [rng.uniform(0.012, 0.020) for _ in range(N_STOCK)]
    mkt = [rng_r.gauss(0.00035, 0.010) for _ in range(T_DAYS)]
    ind = [[rng_r.gauss(0.0, 0.005) for _ in range(T_DAYS)]
           for _ in range(3)]
    events = []
    for i in range(N_STOCK):
        k = rng.randint(6, 9)
        cand = rng.sample(range(160, T_DAYS - 30), k)
        picked = []
        for t in sorted(cand):
            if all(abs(t - p) >= 40 for p in picked):
                picked.append(t)
        for t in picked:
            et = rng.randrange(len(EVENT_TYPES))
            strength = rng.uniform(0.8, 1.3)
            events.append(dict(stock=i, day=t, typ=et,
                               strength=strength))
    r = [[beta[i] * mkt[t] + ind[ind_of[i]][t]
          + rng_r.gauss(0.0, sigma[i]) for t in range(T_DAYS)]
         for i in range(N_STOCK)]
    for e in events:
        _, dr, base = EVENT_TYPES[e["typ"]]
        pulse = dr * e["strength"] * base / HOLD
        for d in range(1, HOLD + 1):
            r[e["stock"]][e["day"] + d] += pulse
    price = []
    for i in range(N_STOCK):
        p = [rng_r.uniform(8.0, 80.0)]
        for t in range(1, T_DAYS):
            p.append(p[-1] * math.exp(r[i][t]))
        price.append(p)
    idx = [3000.0]
    for t in range(1, T_DAYS):
        idx.append(idx[-1] * math.exp(mkt[t]))
    return dict(beta=beta, ind=ind_of, mkt=mkt, r=r, price=price,
                index=idx, events=events)


def ols(y, x):
    n = len(y)
    mx = sum(x) / n
    my = sum(y) / n
    sxy = sum((x[t] - mx) * (y[t] - my) for t in range(n))
    sxx = sum((x[t] - mx) ** 2 for t in range(n))
    b = sxy / sxx
    return my - b * mx, b


def event_car(w, e):
    i, td = e["stock"], e["day"]
    ys = [w["price"][i][t + 1] / w["price"][i][t] - 1.0
          for t in range(td + EST_LO, td + EST_HI)]
    xs = [w["mkt"][t + 1] for t in range(td + EST_LO, td + EST_HI)]
    a, b = ols(ys, xs)
    ar = []
    for d in range(CAR_LO, CAR_HI + 1):
        t = td + d
        ri = (w["price"][i][t + 1] / w["price"][i][t] - 1.0
              if d < 0 else w["r"][i][t])
        ar.append(ri - (a + b * w["mkt"][t]))
    car, acc = [], 0.0
    for v in ar:
        acc += v
        car.append(acc)
    return car


def car_window(car, lo, hi):
    return car[hi - CAR_LO] - (car[lo - CAR_LO - 1] if lo > CAR_LO else 0.0)


def type_stats(w):
    per = {k: [] for k in range(len(EVENT_TYPES))}
    curves = {1: [], -1: []}
    for e in w["events"]:
        car = event_car(w, e)
        per[e["typ"]].append(car_window(car, -1, 5))
        curves[EVENT_TYPES[e["typ"]][1]].append(car)
    rows = {}
    for k, vals in per.items():
        n = len(vals)
        mu = sum(vals) / n
        sd = math.sqrt(sum((v - mu) ** 2 for v in vals) / (n - 1))
        rows[k] = (n, mu, sd, mu / (sd / math.sqrt(n)))
    avg = {}
    npt = CAR_HI - CAR_LO + 1
    for g, lst in curves.items():
        avg[g] = [sum(c[d] for c in lst) / len(lst) for d in range(npt)]
    return rows, avg


def perf(nav, trades, days):
    rets = [(nav[t] / nav[t - 1] - 1.0) for t in range(1, len(nav))]
    mu = sum(rets) / len(rets)
    sd = math.sqrt(sum((v - mu) ** 2 for v in rets) / (len(rets) - 1))
    ann = ((nav[-1] / nav[0]) ** (244.0 / days) - 1.0) * 100.0
    peak, mdd = nav[0], 0.0
    for v in nav:
        peak = max(peak, v)
        mdd = max(mdd, 1.0 - v / peak)
    win = sum(1 for p in trades if p > 0) / len(trades) * 100.0 \
        if trades else 0.0
    sharpe = mu / sd * math.sqrt(244.0) if sd > 0 else 0.0
    return dict(total=nav[-1] / nav[0] * 100.0 - 100.0, ann=ann,
                sharpe=sharpe, mdd=mdd * 100.0,
                win=win, n=len(trades) if trades else 0)


def run_strategy(w, types_sel, lo_day, hi_day, cash0=1e6):
    plans = {}
    for e in w["events"]:
        if e["typ"] in types_sel and lo_day <= e["day"] < hi_day:
            plans.setdefault(e["day"] + 1, []).append(e["stock"])
    cash = cash0
    pos = {}
    nav, trades = [], []
    for t in range(lo_day, hi_day):
        for i, (sh, xd, cost) in list(pos.items()):
            if t >= xd:
                cash += sh * w["price"][i][t]
                trades.append(sh * w["price"][i][t] / cost - 1.0)
                del pos[i]
        if t in plans and len(pos) < 5:
            per = min(cash / max(5 - len(pos), 1), cash * 0.25)
            for i in plans[t]:
                if len(pos) >= 5 or per <= 0 or per > cash:
                    break
                sh = per / w["price"][i][t]
                cash -= per
                pos[i] = (sh, t + HOLD, per)
        eq = cash + sum(sh * w["price"][i][t]
                        for i, (sh, _x, _c) in pos.items())
        nav.append(eq)
    for i, (sh, xd, cost) in list(pos.items()):
        cash += sh * w["price"][i][hi_day - 1]
        trades.append(sh * w["price"][i][hi_day - 1] / cost - 1.0)
        del pos[i]
    nav.append(cash)
    return nav, trades


def random_entry_benchmark(w, lo_day, hi_day, n_sim=500, seed=SEED + 11):
    rng = random.Random(seed)
    out = []
    while len(out) < n_sim:
        i = rng.randrange(N_STOCK)
        t = rng.randrange(lo_day, hi_day - HOLD - 1)
        if any(e["stock"] == i and abs(e["day"] - t) < HOLD + 5
               for e in w["events"]):
            continue
        out.append(w["price"][i][t + HOLD] / w["price"][i][t] - 1.0)
    return out


def main():
    w = gen_world()
    print("指数期初 %.0f 期末 %.1f (%.1f%%)"
          % (w["index"][0], w["index"][-1],
             (w["index"][-1] / w["index"][0] - 1) * 100))
    mu_m = sum(w["mkt"]) / len(w["mkt"])
    sd_m = math.sqrt(sum((v - mu_m) ** 2 for v in w["mkt"])
                     / (len(w["mkt"]) - 1))
    print("市场日波动 %.2f%% 年化 %.1f%%"
          % (sd_m * 100, sd_m * math.sqrt(244) * 100))
    tc = [0] * len(EVENT_TYPES)
    for e in w["events"]:
        tc[e["typ"]] += 1
    print("事件总数 %d 类型分布 %s" % (len(w["events"]), tc))
    rows, avg = type_stats(w)
    for k, (nm, dr, _b) in enumerate(EVENT_TYPES):
        n, mu, sd, tstat = rows[k]
        print("%s n=%d mean=%+.2f%% sd=%.2f%% t=%.2f%s"
              % (nm, n, mu * 100, sd * 100, tstat,
                 " *" if abs(tstat) > 1.96 else ""))
    print("利好组平均 CAR[+10]=%+.2f%%  利空组平均 CAR[+10]=%+.2f%%"
          % (avg[1][-1] * 100, avg[-1][-1] * 100))
    print("利好组 CAR 曲线 [-5,+10]: %s"
          % ["%+.3f" % (v * 100) for v in avg[1]])
    print("利空组 CAR 曲线 [-5,+10]: %s"
          % ["%+.3f" % (v * 100) for v in avg[-1]])
    tr_ev = [e for e in w["events"] if e["day"] < SPLIT]
    long_types = []
    for k in range(len(EVENT_TYPES)):
        vals = [car_window(event_car(w, e), -1, HOLD)
                for e in tr_ev if e["typ"] == k]
        if not vals or EVENT_TYPES[k][1] < 0:
            continue
        mu_k = sum(vals) / len(vals)
        sd_k = math.sqrt(sum((v - mu_k) ** 2 for v in vals)
                         / (len(vals) - 1))
        t_k = mu_k / (sd_k / math.sqrt(len(vals)))
        print("%s 训练n=%d mean=%+.2f%% t=%.2f -> %s"
              % (EVENT_TYPES[k][0], len(vals), mu_k * 100, t_k,
                 "入选" if mu_k > 0 and t_k > 1.5 else "落选"))
        if mu_k > 0 and t_k > 1.5:
            long_types.append(k)
    print("入选多头类型: %s" % [EVENT_TYPES[k][0] for k in long_types])
    days = T_DAYS - SPLIT
    res = {}
    for tag, sel in (("CAR策略", long_types),
                     ("全事件买入", [k for k in range(len(EVENT_TYPES))
                                    if EVENT_TYPES[k][1] > 0])):
        nav, tr = run_strategy(w, set(sel), SPLIT, T_DAYS)
        res[tag] = perf(nav, tr, days)
        res[tag]["nav"] = nav
        res[tag]["trades"] = tr
    bnav = [v / w["index"][SPLIT] for v in w["index"][SPLIT:T_DAYS]]
    res["指数基准"] = perf(bnav, None, days)
    res["指数基准"]["nav"] = bnav
    for tag in ("CAR策略", "全事件买入", "指数基准"):
        p = res[tag]
        if tag == "指数基准":
            print("%s 总回报%+.1f%% 年化%+.1f%% 夏普%.2f 最大回撤%.1f%%"
                  % (tag, p["total"], p["ann"], p["sharpe"], p["mdd"]))
        else:
            print("%s 笔数%d 总回报%+.1f%% 年化%+.1f%% 夏普%.2f "
                  "最大回撤%.1f%% 胜率%.1f%%"
                  % (tag, p["n"], p["total"], p["ann"], p["sharpe"],
                     p["mdd"], p["win"]))
    print("策略净值曲线(每25日): %s"
          % ["%.3f" % v for v in res["CAR策略"]["nav"][::25]])
    print("基准净值曲线(每25日): %s"
          % ["%.3f" % v for v in res["指数基准"]["nav"][::25]])
    rnd = random_entry_benchmark(w, SPLIT, T_DAYS)
    mu_r = sum(rnd) / len(rnd)
    srnd = sorted(rnd)
    q05, q50, q95 = srnd[25], srnd[250], srnd[475]
    print("随机10日持有回报 均值%+.2f%% 分位[P5,P50,P95]="
          "[%.2f%%, %.2f%%, %.2f%%]"
          % (mu_r * 100, q05 * 100, q50 * 100, q95 * 100))


if __name__ == "__main__":
    main()