第 38 篇：协整与配对交易实战

# —— 第一块：取两只真实 ETF，接口不可用则降级造协整数据 ——
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import akshare as ak

rng = np.random.default_rng(11)

def get_etf_close(symbol: str) -> pd.Series | None:
    """拉一只 ETF 的日线收盘价，失败返回 None"""
    try:
        df = ak.fund_etf_hist_em(symbol=symbol, period="daily", adjust="qfq")
        s = pd.to_datetime(df["日期"])
        return pd.Series(df["收盘"].values, index=s, name=symbol).loc["2020-01-01":"2024-12-31"]
    except Exception:
        return None

pa = get_etf_close("510050")    # 上证50ETF
pb = get_etf_close("510300")    # 沪深300ETF

if pa is None or pb is None:    # 降级方案：造一对真正协整的「ETF」
    n = 1218
    idx = pd.bdate_range("2020-01-01", periods=n)
    lb = pd.Series(8.0 + np.cumsum(rng.normal(0, 0.012, n)), index=idx)  # 随机游走
    eps = np.zeros(n)
    for t in range(1, n):       # 平稳噪声：绕 0 往返（AR(1)）
        eps[t] = 0.90 * eps[t - 1] + rng.normal(0, 0.02)
    la = pd.Series(0.80 * lb.values + 1.2 + eps, index=idx)
    pa, pb = np.exp(la), np.exp(lb)
    print("akshare 不可用，使用造数据（保证协整）")
else:
    pa, pb = pa.align(pb, join="inner")   # 按共同交易日对齐
    print(f"akshare 数据：各 {len(pa)} 个交易日")

# —— 第二块：训练段上做检验与对冲比率估计 ——
from statsmodels.tsa.stattools import coint, adfuller
import statsmodels.api as sm

la, lb = np.log(pa), np.log(pb)
split = int(len(la) * 0.7)      # 前 70% 估计参数，后 30% 模拟交易
la_tr, lb_tr = la.iloc[:split], lb.iloc[:split]

corr = pa.pct_change().corr(pb.pct_change())
print(f"全样本日收益率相关系数：{corr:.3f}")

X = sm.add_constant(lb_tr.values)             # Engle-Granger 第一步：OLS 估 β
ols = sm.OLS(la_tr.values, X).fit()
beta = float(ols.params[1])
print(f"对冲比率 beta ≈ {beta:.3f}")

spread_tr = la_tr - beta * lb_tr              # 第二步：对残差做 ADF
_, p_adf, *_ = adfuller(spread_tr)
print(f"训练段价差 ADF：p值={p_adf:.4f} → {'平稳' if p_adf < 0.05 else '不平稳'}")

_, p_coint, _ = coint(la_tr, lb_tr)           # 一行等价于上面两步
print(f"训练段 coint 检验：p值={p_coint:.4f} → {'存在协整' if p_coint < 0.05 else '不存在协整'}")

# —— 第三块：测试段上的 z-score 信号与简化回测 ——
window = 120
spread_all = la - beta * lb                    # 用训练段的 β 套全样本价差
z = ((spread_all - spread_all.rolling(window).mean())
     / spread_all.rolling(window).std()).dropna()
z = z[z.index >= la.index[split]]              # 只在测试段交易，避免用未来数据

pos = np.zeros(len(z))                         # 仓位：+1 多价差 / -1 空价差 / 0 空仓
for i, zv in enumerate(z.values):
    if zv < -2:
        pos[i] = 1                             # 价差过低：买 A 卖 B
    elif zv > 2:
        pos[i] = -1                            # 价差过高：卖 A 买 B
    elif abs(zv) > 4:                          # 硬止损：皮筋可能断了，认错离场
        pos[i] = 0
    elif abs(zv) < 0.5:
        pos[i] = 0                             # 回归到位，止盈平仓
    else:
        pos[i] = pos[i - 1] if i > 0 else 0    # 其余时间维持原仓位
pos = pd.Series(pos, index=z.index)

ra = pa.pct_change().reindex(z.index).fillna(0)     # 两腿的日收益
rb = pb.pct_change().reindex(z.index).fillna(0)
spread_ret = (ra - beta * rb) / (1 + beta)          # 归一化到总名义 1 的价差日收益

cost = pos.diff().abs().fillna(0) * 0.001           # 每次换手按双边 0.1% 计费
strat_ret = pos.shift(1).fillna(0) * spread_ret - cost
nav = (1 + strat_ret).cumprod()

turns = int((pos.diff().abs() > 0).sum())
ann = nav.iloc[-1] ** (244 / len(strat_ret)) - 1
sharpe = (strat_ret.mean() * 244 - 0.02) / (strat_ret.std() * np.sqrt(244))
mdd = (nav / nav.cummax() - 1).min()
print(f"测试段换手次数：{turns}，年化：{ann:.3f}，夏普：{sharpe:.2f}，最大回撤：{mdd:.2%}")

# —— 第四块：价差与 z-score 可视化 ——
fig, axes = plt.subplots(2, 1, figsize=(10, 6), sharex=True)
axes[0].plot(spread_all.index, spread_all.values, lw=0.8, color="gray")
axes[0].axvline(la.index[split], color="black", ls="--", lw=1, label="训练/测试分界")
axes[0].set_title("价差（对数）：训练段估计，测试段交易")
axes[0].legend()
axes[1].plot(z.index, z.values, lw=0.8, color="steelblue")
axes[1].axhline(2, color="red", ls="--", lw=1, label="做空线 +2")
axes[1].axhline(-2, color="green", ls="--", lw=1, label="做多线 -2")
axes[1].axhline(4, color="darkred", ls=":", lw=1, label="止损线 +4")
axes[1].axhline(-0.5, color="gray", ls=":", lw=1, label="平仓区")
axes[1].axhline(0.5, color="gray", ls=":", lw=1)
axes[1].set_title("z-score：穿 ±2 进场，回 ±0.5 出场")
axes[1].legend(loc="upper left", fontsize=8, ncol=2)
plt.tight_layout()
plt.show()

全样本日收益率相关系数：0.977
对冲比率 beta ≈ 0.802
训练段价差 ADF：p值=0.0003 → 平稳
训练段 coint 检验：p值=0.0012 → 存在协整