2026年重磅升级已全面落地！欢迎加入专注财经数据与量化投研的【数据科学实战】知识星球！您将获取持续更新的《财经数据宝典》与《量化投研宝典》，双典协同提供系统化指引；星球内含300篇以上独有高质量文章，深度覆盖策略开发、因子分析、风险管理等核心领域，内容基本每日更新；同步推出的「量化因子专题教程」系列（含完整可运行代码与实战案例），系统详解因子构建、回测与优化全流程，并实现日更迭代。我们持续扩充独家内容资源，全方位赋能您的投研效率与专业成长。无论您是量化新手还是资深研究者，这里都是助您少走弯路、事半功倍的理想伙伴，携手共探数据驱动的投资未来！

引言

统计套利（Statistical Arbitrage，简称 StatArb）常被视为量化金融中的"黑箱"——复杂、神秘，似乎只有机构玩家才能驾驭。然而，其核心思想其实非常直观：识别并利用数学相关资产之间的临时定价偏差。

本文将带你深入理解一个经典的 Lead-Lag 策略——假设某些"领导者"股票（如大型科技巨头）的价格变动可以预测相关"目标"资产的短期走势。我们不依赖主观判断，而是用 Python 构建一个完整的量化交易流程，涵盖信号生成、收益计算和策略可视化。

无论你是量化交易新手还是希望系统学习策略开发的 Python 爱好者，这篇文章都将为你打下坚实的基础。

核心概念解析

什么是 Lead-Lag 策略？

Lead-Lag 策略的核心假设是：某些资产的价格变动会"领先"于另一些资产。通过计算残差——即目标资产的预期价格与实际价格之间的差异——我们可以生成一个均值回归的交易信号。

关键技术要点

1. 1. 协方差比率：量化领导者与目标资产之间的关系强度
2. 2. 滚动置信区间：作为波动率过滤器，避免在市场噪音过高时交易
3. 3. 向量化实现：利用 Pandas 和 NumPy 的高效计算能力

环境准备

首先，我们需要配置开发环境并导入必要的库：

import osimport warningsimport numpy as npimport pandas as pdimport matplotlib.pyplot as pltimport yfinance as yf# 忽略警告信息，保持输出简洁warnings.filterwarnings("ignore")# 定义需要安装的包required_packages = ["yfinance", "statsmodels", "seaborn", "scikit-learn"]print("环境准备完成！")

构建股票池

我们选取纳斯达克 100 指数中市值最高的一篮子股票作为研究对象：

# 定义时间范围START = "2020-01-01"END = "2023-11-22"# 股票池：纳斯达克 100 高市值股票symbols = [    "AAPL", "MSFT", "AMZN", "TSLA", "GOOG", "GOOGL", "META", "NVDA", "PEP",    "AVGO", "ADBE", "COST", "PYPL", "AMD", "QCOM", "INTC", "TXN", "CHTR",    "TMUS", "ISRG", "SBUX", "AMGN", "INTU", "ADP", "CSX", "ADI", "MU",    "ZM", "MAR", "GILD", "MELI", "WDAY", "CTSH", "PANW", "REGN", "LRCX",    "BKNG", "EXC", "ADSK", "KLAC", "TEAM"]# 使用 yfinance 下载价格数据price_df = yf.download(symbols, start=START, end=END)["Adj Close"]# 删除含有缺失值的列valid_cols_df = price_df.dropna(axis=1)# 计算周期收益率periodic_rets = valid_cols_df.pct_change().dropna()# 计算累计收益率cumulative_rets = (1 + periodic_rets).cumprod() - 1print(f"成功加载 {len(valid_cols_df.columns)} 只股票的数据")

相关性分析

统计套利的第一步是找出与目标资产高度相关的"领导者"股票：

# 设置基准股票和滚动窗口参数BASE_TICKER = "AAPL"SMOOTH_WINDOW = 24  # 平滑窗口ROLL_WINDOW = SMOOTH_WINDOW * 10  # 滚动相关性窗口# 存储高相关性的候选股票lead_candidates = []# 遍历所有股票，筛选相关性大于 0.75 的股票for symbol in symbols:    if symbol == BASE_TICKER:        continue    # 计算与基准股票的相关系数    corr_val = valid_cols_df[symbol].corr(valid_cols_df[BASE_TICKER])    # 只保留相关性绝对值大于 0.75 的股票    if abs(corr_val) >= 0.75:        lead_candidates.append(symbol)        print(f"{symbol} 与 {BASE_TICKER} 的相关系数: {corr_val:.3f}")print(f"\n共筛选出 {len(lead_candidates)} 只候选领导者股票")

信号生成核心逻辑

这是策略的核心部分。我们通过以下步骤生成交易信号：

步骤 1：计算 EMA 偏差

# 定义目标股票和参数TARGET = "AAPL"MA_WINDOW = 21  # EMA 窗口ARB_WINDOW = 126  # 套利窗口（约 6 个月）LEADS = ["MSFT", "GOOGL", "META"]  # 领导者股票# 创建工作数据框arb_df = periodic_rets.copy()arb_df["price"] = valid_cols_df[TARGET]# 计算目标股票的 EMA 和偏差arb_df[f"ema_{TARGET}"] = arb_df[TARGET].ewm(span=MA_WINDOW).mean()arb_df[f"ema_d_{TARGET}"] = arb_df[TARGET] - arb_df[f"ema_{TARGET}"]arb_df[f"ema_d_{TARGET}"].fillna(0, inplace=True)print("目标股票 EMA 偏差计算完成")

步骤 2：计算领导者与目标的协方差比率

# 对每个领导者股票进行处理for lead in LEADS:    # 计算领导者的 EMA 和偏差    arb_df[f"ema_{lead}"] = arb_df[lead].ewm(span=MA_WINDOW).mean()    arb_df[f"ema_d_{lead}"] = arb_df[lead] - arb_df[f"ema_{lead}"]    arb_df[f"ema_d_{lead}"].fillna(0, inplace=True)    # 计算滚动相关系数    arb_df[f"{lead}_corr"] = arb_df[f"ema_d_{lead}"].rolling(ARB_WINDOW).corr(        arb_df[f"ema_d_{TARGET}"]    )    # 计算协方差比率（类似于回归系数）    cov_series = arb_df[f"ema_d_{lead}"].rolling(ARB_WINDOW).cov(        arb_df[f"ema_d_{TARGET}"]    )    var_series = arb_df[f"ema_d_{lead}"].rolling(ARB_WINDOW).var()    arb_df[f"{lead}_covr"] = cov_series / var_series    # 计算预测值和残差    arb_df[f"{lead}_proj"] = arb_df[f"ema_d_{lead}"] * arb_df[f"{lead}_covr"]    arb_df[f"{lead}_residual"] = arb_df[f"{lead}_proj"] - arb_df[f"ema_d_{TARGET}"]print("协方差比率和残差计算完成")

步骤 3：生成加权信号

# 初始化累加器accum_weights = 0accum_delta = 0# 对每个领导者的信号进行相关性加权for symbol in LEADS:    # 使用相关系数的绝对值作为权重    corr_mag = arb_df[f"{symbol}_corr"].fillna(0).abs()    accum_weights += corr_mag    # 加权残差    arb_df[f"{symbol}_weighted"] = arb_df[f"{symbol}_residual"].fillna(0) * corr_mag    accum_delta += arb_df[f"{symbol}_weighted"]# 避免除以零weights = accum_weights.replace(0, 1)# 生成最终信号arb_df[f"{TARGET}_signal"] = accum_delta / weightsprint("交易信号生成完成")

置信区间计算

为了过滤噪音，我们计算信号的滚动置信区间：

# 获取信号列signal_col = arb_df[f"{TARGET}_signal"]# 计算滚动均值和标准差rolling_mean = signal_col.rolling(ARB_WINDOW).mean().fillna(0)rolling_std = signal_col.rolling(ARB_WINDOW).std().fillna(0)# 计算 95% 置信区间（使用 1.96 作为 Z 值）arb_df["ci_lower"] = signal_col.fillna(0) + (rolling_mean - 1.96 * rolling_std)arb_df["ci_upper"] = signal_col.fillna(0) + (rolling_mean + 1.96 * rolling_std)arb_df["ci_spread"] = arb_df["ci_upper"] - arb_df["ci_lower"]# 填充缺失值arb_df.fillna(0, inplace=True)print("置信区间计算完成")

交易信号生成

根据信号和置信区间生成具体的买卖指令：

# 定义交易阈值long_thresh = 0.0025   # 做多阈值short_thresh = -0.0025  # 做空阈值conf_spread_thresh = 0.15  # 置信区间宽度阈值max_qty = 1  # 最大交易数量# 初始化订单列arb_df["orders"] = 0# 获取信号序列sig_series = arb_df[f"{TARGET}_signal"]next_signal = sig_series.shift(-1)# 做多入场条件：# 1. 当前信号高于做多阈值# 2. 下一期信号将回落至阈值以下（信号穿越）# 3. 信号在置信区间内# 4. 置信区间宽度足够窄long_entry = (    (sig_series > long_thresh) &    (next_signal <= long_thresh) &    (sig_series < arb_df["ci_upper"]) &    (sig_series > arb_df["ci_lower"]) &    (arb_df["ci_spread"] < conf_spread_thresh))# 做空入场条件（与做多相反）short_entry = (    (sig_series < short_thresh) &    (next_signal >= short_thresh) &    (sig_series < arb_df["ci_upper"]) &    (sig_series > arb_df["ci_lower"]) &    (arb_df["ci_spread"] < conf_spread_thresh))# 生成订单arb_df.loc[long_entry, "orders"] += max_qtyarb_df.loc[short_entry, "orders"] -= max_qtyprint(f"做多信号数量: {long_entry.sum()}")print(f"做空信号数量: {short_entry.sum()}")

收益计算

计算策略的盈亏情况：

# 筛选有交易的记录subset_mask = long_entry | short_entrychanges = arb_df.loc[subset_mask, ["price", "orders", f"{TARGET}_signal"]].copy()# 计算持仓changes["position"] = changes["orders"].cumsum()# 计算开仓价格changes["price_open"] = changes["price"].shift(1)# 计算收益changes["pnl"] = (changes["price"] - changes["price_open"]) * np.sign(changes["position"].shift(1))changes["pnl"].fillna(0, inplace=True)# 计算收益率changes["pnl_rets"] = changes["pnl"] / changes["price_open"].abs()changes["pnl_rets"].fillna(0, inplace=True)# 计算累计收益率cumulative_pnl = changes["pnl_rets"].cumsum()print(f"累计收益率: {cumulative_pnl.iloc[-1]:.2%}")

策略可视化

最后，我们将价格走势、交易信号和累计收益可视化：

# 创建三行子图fig, axes = plt.subplots(3, 1, figsize=(16, 12), gridspec_kw={"height_ratios": (3, 1, 1)})ax_top, ax_mid, ax_bot = axesstart_idx = ARB_WINDOW# 上图：价格走势和交易点位ax_top.plot(arb_df.iloc[start_idx:].index, arb_df["price"].iloc[start_idx:],             color="g", lw=1.25, label=f"{TARGET} 价格")ax_top.plot(arb_df.loc[long_entry].index, arb_df.loc[long_entry, "price"],            "^", markersize=12, color="blue", label="买入")ax_top.plot(arb_df.loc[short_entry].index, arb_df.loc[short_entry, "price"],            "v", markersize=12, color="red", label="卖出")ax_top.set_ylabel("价格 ($)")ax_top.legend(loc="upper left")ax_top.set_title(f"统计套利策略 - {TARGET}", fontsize=14)# 中图：累计收益ax_mid.plot(changes.index, changes["pnl_rets"].cumsum(), color="b", label="累计收益")ax_mid.set_ylabel("累计收益率 (%)")ax_mid.legend(loc="upper left")ax_mid.set_title("策略累计收益", fontsize=14)# 下图：交易信号ax_bot.plot(arb_df.iloc[start_idx:].index, arb_df[f"{TARGET}_signal"].iloc[start_idx:],             label="信号", alpha=0.75, color="g")ax_bot.axhline(0, color="black", linestyle="--", linewidth=1)ax_bot.axhline(short_thresh, color="r", linestyle="--", label=f"做空阈值 ({short_thresh})")ax_bot.axhline(long_thresh, color="b", linestyle="--", label=f"做多阈值 ({long_thresh})")ax_bot.fill_between(arb_df.iloc[start_idx:].index,                     arb_df["ci_lower"].iloc[start_idx:],                     arb_df["ci_upper"].iloc[start_idx:],                     color="gray", alpha=0.3, label="置信区间")ax_bot.legend(loc="lower left")ax_bot.set_title("交易信号", fontsize=14)plt.tight_layout()plt.show()

策略优化方向

这个基础版本的统计套利策略可以从以下几个方面进行优化：

1. 1. 参数调优：使用网格搜索或贝叶斯优化寻找最佳的窗口长度和阈值
2. 2. 风险管理：加入止损和止盈机制
3. 3. 交易成本：考虑滑点和手续费的影响
4. 4. 多资产组合：将策略扩展到多个标的
5. 5. 机器学习增强：使用 Kalman 滤波器或神经网络优化信号

总结

本文详细介绍了如何用 Python 从零构建一个统计套利策略，核心内容包括：

1. 1. 数据准备：使用 yfinance 获取股票数据，并进行清洗和预处理
2. 2. 相关性筛选：通过滚动相关系数筛选与目标资产高度相关的领导者股票
3. 3. 信号生成：利用 EMA 偏差和协方差比率构建交易信号
4. 4. 风险过滤：使用滚动置信区间过滤高噪音时期的信号
5. 5. 收益计算：基于向量化操作高效计算策略盈亏
6. 6. 可视化分析：通过多层图表直观展示策略表现

虽然这是一个教学性质的简化模型，但它展示了量化策略开发的完整流程。在实际应用中，还需要考虑更多因素，如交易成本、市场冲击、风险控制等。

掌握这些基础知识后，你就可以在此基础上不断迭代优化，构建属于自己的量化交易系统。

参考文章

1. 加入专注于财经数据与量化投研的知识星球【数据科学实战】，获取本文完整研究解析、代码实现细节。

财经数据与量化投研知识社区

2026年全面升级已落地！【数据科学实战】知识星球核心权益如下：

1. 1. 双典系统赋能：获赠《财经数据宝典》与《量化投研宝典》完整文档，凝练多年实战经验，构建系统化知识框架；
2. 2. 量化因子日更教程（2026重磅新增）：每日更新「量化因子专题教程」，配套完整可运行代码与实战案例，深度拆解因子构建、回测与优化全流程；
3. 3. 量化文章专题教程库：300+篇星球独有高质量教程式文章，系统覆盖策略开发、因子研究、风险管理等核心领域，内容基本每日更新，并配套精选学习资料与实战参考；
4. 4. PyBroker实战课程：赠送《PyBroker-入门及实战》视频课程，手把手教学，快速掌握量化策略开发技能；
5. 5. 财经数据支持：定期更新国内外财经数据，为策略研发提供精准、可靠的数据基础；
6. 6. 顶尖学者与行业专家分享：年度邀请学术界博士与业界资深专家开展前沿论文精讲与实战案例分享，不少于4场，直击研究前沿与产业实践；专家直连答疑：与核心开发者及领域专家实时互动，高效解决投研实战难题；
7. 7. 专业社群与专属福利：加入高质量交流社群，获取课程折扣及更多独家资源。

星球已沉淀丰富内容生态——涵盖量化文章专题教程库、因子日更系列、高频数据集、PyBroker实战课程、专家深度分享与实时答疑服务。无论您是初探量化的学习者，还是深耕领域的从业者，这里都是助您少走弯路、高效成长的理想平台。诚邀加入，共探数据驱动的投资未来！

好文推荐