Pairs Trading（配对交易与协整）

日期: 2026-07-29 方向: 量化 / 统计套利 / Alpha 阶段: Phase 2 - 统计套利与Alpha Research (Day 89-102) 标签: #量化策略 #套利 #协整 #Cointegration #PairsTrading

---

今日目标

---

一、理论与模型

1.1 为什么不是相关性而是协整

资深量化从业者最常见的混淆：Correlation ≠ Cointegration。

关键认知：相关性高不代表能做配对交易。两条独立随机游走（I(1)）可能短期相关性 > 0.95，但价差是 $I(1)$ 的，不平稳，做空价差只会爆仓。

1.2 协整的形式化定义

设 $X_t, Y_t$ 都是 $I(1)$（一阶单整，即一阶差分平稳），如果存在 $\beta \neq 0$ 使得：

$$ Z_t = Y_t - \beta X_t \sim I(0) \quad \text{（平稳）} $$

则称 $(X_t, Y_t)$ 协整，$\beta$ 称为协整向量。

Granger表示定理：若 $X_t, Y_t$ 协整，则一定存在误差修正模型 (VECM)：

$$ \Delta Y_t = \alpha_y (Y_{t-1} - \beta X_{t-1}) + \sum \gamma_i \Delta Y_{t-i} + \sum \delta_i \Delta X_{t-i} + \epsilon_t $$

其中 $\alpha_y < 0$ 是回归速度（误差修正系数）。

1.3 Engle-Granger 两步法

Step 1：对 $Y_t = \alpha + \beta X_t + Z_t$ 做 OLS 回归，得到残差 $\hat Z_t$ Step 2：对 $\hat Z_t$ 做 ADF 单位根检验

ADF（Augmented Dickey-Fuller）：检验 $H_0: Z_t$ 含单位根（非平稳） vs $H_1$ 平稳

$$ \Delta Z_t = \rho Z_{t-1} + \sum_{i=1}^{p} \phi_i \Delta Z_{t-i} + \epsilon_t $$

检验统计量：$\tau = \hat \rho / \text{SE}(\hat \rho)$，与 Dickey-Fuller 临界值比较（注意：不是标准t分布）。

1.4 半衰期（Half-Life）

价差的回归速度：拟合 $\Delta Z_t = \theta + \kappa Z_{t-1} + \epsilon_t$，则

$$ \text{Half-Life} = -\frac{\ln 2}{\kappa} $$

经验法则：

• 半衰期 < 1 天：高频均值回归，胜率高但容量小
• 半衰期 1-30 天：日内/日间统计套利，最佳区间
• 半衰期 > 100 天：太慢，资金占用成本高，可能伪协整

1.5 Hurst 指数

$$ \text{Hurst} = H, \quad \text{var}(Z_{t+\tau} - Z_t) \propto \tau^{2H} $$

• $H < 0.5$：均值回归（适合做配对）
• $H = 0.5$：随机游走
• $H > 0.5$：趋势（不适合做配对）

---

二、直觉与陷阱

陷阱 1：ADF 的 I(1) 误判

坑：测试 ADF 时若价格本身是 I(2)（例如累积通胀），残差也可能"伪平稳"。 解法：先对每个序列单独做 ADF，确认都是 I(1)，再做协整。

陷阱 2：β 估计偏差（OLS 内生性）

OLS 的 $\beta$ 不对称：把 X 回归到 Y 和把 Y 回归到 X 得到的 $\beta$ 不同（除非 R² = 1）。 解法：用 TLS（Total Least Squares）/ Johansen 方法对称估计；或两个方向都跑一遍取均值。

陷阱 3：协整破坏 (Regime Change)

加密世界协整破坏频繁：

• 2022/05 LUNA-UST 死亡螺旋：UST/USDC 历史协整 6 个月，3 天破灭
• 2022/06 stETH/ETH：3AC + Celsius 流动性危机使 stETH 折价 8%，半衰期模型完全失效
• 2023/03 USDC 脱钩：SVB 倒闭使 USDC/USDT 协整对在 36 小时内偏离 6%

解法：

1. 滚动协整检验（每周重新校准）
2. 设置硬止损（如 4σ 强平）
3. 监控基本面变化（项目方公告、链上储备）

陷阱 4：数据频率失配

5 分钟数据与日线协整结论可能完全相反。经验：用决策频率的 2-5 倍频率做检验（日内策略用 5 分钟数据，日间策略用日线）。

陷阱 5：幸存者偏差与多重检验

扫描 100 个交易对找协整，按 5% 显著性水平至少 5 对会"虚假协整"。 解法：Bonferroni 校正：$\alpha_{adjusted} = \alpha / N$；或样本外验证（in-sample 找对，out-of-sample 验证）。

---

三、代码实现

3.1 完整 Pairs Trading 引擎

# pairs.py
"""
Pairs Trading Engine for Crypto Markets
- Cointegration discovery via Engle-Granger
- Half-life estimation
- Z-score signal generation
- Walk-forward backtest with risk controls
"""

import numpy as np
import pandas as pd
import statsmodels.api as sm
from statsmodels.tsa.stattools import adfuller, coint
from typing import Tuple, Optional
import warnings
warnings.filterwarnings('ignore')


# ----------------------------- 数据获取 -----------------------------
def fetch_binance_klines(symbol: str, interval: str = '1d', limit: int = 1000) -> pd.DataFrame:
    """从 Binance 公开 API 拉取 K 线（无需 API key）"""
    import requests
    url = "https://api.binance.com/api/v3/klines"
    r = requests.get(url, params={'symbol': symbol, 'interval': interval, 'limit': limit})
    cols = ['open_time', 'open', 'high', 'low', 'close', 'volume',
            'close_time', 'qav', 'trades', 'tb_base', 'tb_quote', 'ignore']
    df = pd.DataFrame(r.json(), columns=cols)
    df['close'] = df['close'].astype(float)
    df['ts'] = pd.to_datetime(df['close_time'], unit='ms')
    return df.set_index('ts')['close']


# ----------------------------- 协整检验 -----------------------------
def engle_granger_test(y: pd.Series, x: pd.Series) -> dict:
    """Engle-Granger 两步法协整检验"""
    # Step 1: OLS y = α + β x + ε
    X = sm.add_constant(x.values)
    model = sm.OLS(y.values, X).fit()
    alpha, beta = model.params[0], model.params[1]
    residuals = y.values - (alpha + beta * x.values)

    # Step 2: ADF on residuals
    adf_stat, p_value, _, _, crit_values, _ = adfuller(residuals, autolag='AIC')

    return {
        'alpha': alpha,
        'beta': beta,
        'residuals': pd.Series(residuals, index=y.index),
        'adf_stat': adf_stat,
        'p_value': p_value,
        'crit_5pct': crit_values['5%'],
        'cointegrated': adf_stat < crit_values['5%']
    }


def estimate_half_life(spread: pd.Series) -> float:
    """估计均值回归半衰期：ΔZ_t = θ + κ Z_{t-1} + ε"""
    spread_lag = spread.shift(1).dropna()
    spread_diff = spread.diff().dropna()
    spread_lag, spread_diff = spread_lag.align(spread_diff, join='inner')

    X = sm.add_constant(spread_lag.values)
    model = sm.OLS(spread_diff.values, X).fit()
    kappa = model.params[1]
    if kappa >= 0:
        return np.inf  # 不收敛
    return -np.log(2) / kappa


def hurst_exponent(ts: pd.Series, max_lag: int = 100) -> float:
    """Hurst 指数（用于判断均值回归特性）"""
    lags = range(2, max_lag)
    tau = [np.std(np.subtract(ts.values[lag:], ts.values[:-lag])) for lag in lags]
    poly = np.polyfit(np.log(lags), np.log(tau), 1)
    return poly[0]


# ----------------------------- 信号生成 -----------------------------
def generate_zscore_signals(spread: pd.Series, lookback: int = 60,
                            entry_z: float = 2.0, exit_z: float = 0.5,
                            stop_z: float = 4.0) -> pd.DataFrame:
    """
    Z-score 配对交易信号
    - |Z| > entry_z 入场（反向开仓）
    - |Z| < exit_z 平仓
    - |Z| > stop_z 止损（避免协整破坏）
    """
    mu = spread.rolling(lookback).mean()
    sigma = spread.rolling(lookback).std()
    z = (spread - mu) / sigma

    signal = pd.Series(0, index=spread.index)
    position = 0

    for i, zt in enumerate(z):
        if pd.isna(zt):
            continue
        # 止损优先级最高
        if abs(zt) > stop_z:
            position = 0
        elif position == 0:
            if zt > entry_z:
                position = -1  # 价差太高，做空 spread (short Y, long X)
            elif zt < -entry_z:
                position = 1   # 价差太低，做多 spread (long Y, short X)
        elif position != 0 and abs(zt) < exit_z:
            position = 0
        signal.iloc[i] = position

    return pd.DataFrame({'spread': spread, 'zscore': z, 'signal': signal})


# ----------------------------- 回测 -----------------------------
def backtest_pair(y: pd.Series, x: pd.Series, beta: float,
                  signals: pd.DataFrame, fee_bps: float = 5,
                  capital: float = 100_000) -> pd.DataFrame:
    """简化回测：每次开仓 capital/2 在每条腿"""
    rets_y = y.pct_change().fillna(0)
    rets_x = x.pct_change().fillna(0)

    # 配对收益：long Y short β X (signal=1) or vice versa
    pair_ret = signals['signal'].shift(1) * (rets_y - beta * rets_x)
    # 交易成本：每次仓位变动按双边收费
    turnover = signals['signal'].diff().abs().fillna(0)
    cost = turnover * (fee_bps / 10_000) * 2  # 两条腿
    net_ret = pair_ret - cost

    equity = capital * (1 + net_ret).cumprod()
    return pd.DataFrame({
        'gross_ret': pair_ret,
        'cost': cost,
        'net_ret': net_ret,
        'equity': equity
    })


def performance_metrics(returns: pd.Series, annualize: int = 365) -> dict:
    """评估指标"""
    cum = (1 + returns).prod() - 1
    ann_ret = (1 + returns.mean()) ** annualize - 1
    ann_vol = returns.std() * np.sqrt(annualize)
    sharpe = ann_ret / ann_vol if ann_vol > 0 else 0

    equity = (1 + returns).cumprod()
    dd = (equity / equity.cummax() - 1).min()

    return {
        'cum_return': cum,
        'ann_return': ann_ret,
        'ann_vol': ann_vol,
        'sharpe': sharpe,
        'max_drawdown': dd,
    }


# ----------------------------- 主流程 -----------------------------
if __name__ == '__main__':
    # 真实加密对：BTC vs ETH (历史经典协整对，但 2021 后协整减弱)
    print("=" * 60)
    print("Fetching BTC/ETH from Binance...")
    btc = fetch_binance_klines('BTCUSDT', '1d', 730)
    eth = fetch_binance_klines('ETHUSDT', '1d', 730)
    df = pd.DataFrame({'BTC': btc, 'ETH': eth}).dropna()
    print(f"Sample size: {len(df)} days")

    # 用对数价格（标准做法，避免水平偏差）
    log_btc = np.log(df['BTC'])
    log_eth = np.log(df['ETH'])

    # 协整检验
    result = engle_granger_test(log_eth, log_btc)
    print(f"\nEngle-Granger Test:")
    print(f"  beta (hedge ratio): {result['beta']:.4f}")
    print(f"  ADF stat: {result['adf_stat']:.4f}")
    print(f"  p-value: {result['p_value']:.4f}")
    print(f"  5% critical: {result['crit_5pct']:.4f}")
    print(f"  Cointegrated? {result['cointegrated']}")

    # 半衰期
    hl = estimate_half_life(result['residuals'])
    print(f"  Half-life: {hl:.2f} days")

    # Hurst
    h = hurst_exponent(result['residuals'])
    print(f"  Hurst: {h:.3f} (< 0.5 means mean-reverting)")

    # 信号生成
    signals = generate_zscore_signals(result['residuals'], lookback=60,
                                      entry_z=2.0, exit_z=0.5, stop_z=4.0)

    # 回测
    perf = backtest_pair(log_eth, log_btc, result['beta'], signals, fee_bps=5)
    metrics = performance_metrics(perf['net_ret'])

    print(f"\nBacktest Results:")
    for k, v in metrics.items():
        print(f"  {k}: {v:.4f}")

3.2 多对协整扫描器

def scan_cointegrated_pairs(price_df: pd.DataFrame,
                             p_threshold: float = 0.05,
                             min_half_life: float = 1,
                             max_half_life: float = 30) -> pd.DataFrame:
    """
    扫描所有交易对找协整对
    Bonferroni 校正后的 p 值阈值
    """
    n = len(price_df.columns)
    n_pairs = n * (n - 1) // 2
    p_adj = p_threshold / n_pairs  # Bonferroni

    results = []
    for i, sym_a in enumerate(price_df.columns):
        for sym_b in price_df.columns[i+1:]:
            log_a = np.log(price_df[sym_a].dropna())
            log_b = np.log(price_df[sym_b].dropna())
            common = log_a.index.intersection(log_b.index)
            if len(common) < 200:
                continue
            res = engle_granger_test(log_a.loc[common], log_b.loc[common])
            if res['p_value'] < p_adj:
                hl = estimate_half_life(res['residuals'])
                if min_half_life <= hl <= max_half_life:
                    results.append({
                        'pair': f"{sym_a}/{sym_b}",
                        'p_value': res['p_value'],
                        'beta': res['beta'],
                        'half_life': hl,
                    })
    return pd.DataFrame(results).sort_values('p_value')

---

四、真实数据/案例

案例 1：stETH / ETH 协整破坏（2022 年 6 月）

教训：

• 协整模型基于"赎回机制有效"假设，但 Shapella 升级前 stETH 不能 1:1 赎回，依赖 Curve 流动性
• 在 6 月 13 日，做多 spread (long stETH, short ETH) 的策略浮亏 7%；若 leverage 5x 则爆仓
• 正确做法：检验依赖的市场结构假设（如赎回机制是否激活）

案例 2：3AC 的 GBTC 套利崩溃（2021）

3AC 长期做多 GBTC（灰度 BTC 信托）vs 现货 BTC 的协整对，假设折价/溢价均值回归。 2021 年 GBTC 折价持续扩大到 -45%，3AC 杠杆爆仓贡献了崩盘。

核心认知：流动性溢价/折价的"均值"可能不是 0，而是结构性偏移。

案例 3：BTC/ETH 协整的衰减

---

五、CEX vs DEX 策略差异

DeFi 原生 Pairs 机会：

1. stETH/ETH：Lido 质押 + Curve 流动性
2. rETH/ETH：Rocket Pool 质押
3. wstETH/stETH：包装代币与原生代币的轻微偏移
4. GMX-V2 GLP vs spot：流动性提供者代币的内在价值与市价

---

六、风险管理

6.1 配对策略特有风险

6.2 容量分析

日均 ADV (Average Daily Volume) 是容量上限
单笔仓位 ≤ ADV × 5%
策略总容量 ≈ ADV × 10%

例：BTC/ETH 在 Binance ADV ~$30B/day，单对策略容量上限约 $3B/day（实际受集中度限制更小）。

6.3 风控参数推荐

---

七、关键速查

协整检验决策树

Step 1: 序列 P_x, P_y 都是 I(1)?
    ├─ 否 → 先差分至平稳后再考虑
    └─ 是 → Step 2

Step 2: ADF(残差) p < 0.05?
    ├─ 否 → 不协整，放弃
    └─ 是 → Step 3

Step 3: Half-Life ∈ [1, 30] days?
    ├─ 否 → 太慢/太快，跳过
    └─ 是 → Step 4

Step 4: Hurst < 0.4?
    ├─ 否 → 信号弱，谨慎
    └─ 是 → 入选策略池

Python 函数速查

关键阈值

---

八、面试题

Q1：你怎么区分相关性和协整？为什么配对交易必须用协整？

答：

• 相关性衡量收益率同向变动幅度，是短期统计指标，可能由共同趋势虚假抬高
• 协整衡量价格水平的长期均衡关系，要求残差平稳（I(0)）
• 反例：两个独立随机游走可能短期相关性 > 0.9，但价差是 I(1)，做配对永远不回归
• 配对交易本质是做空 spread 的均值回归，要求 spread 平稳，即协整条件

Q2：你发现一对资产 ADF p-value = 0.03，half-life = 8 天，Hurst = 0.45。但你的策略亏钱。可能的原因？

答：

1. 样本外失效：in-sample 协整在 out-of-sample 破坏（regime change），如 LUNA/UST
2. 多重检验偏差：扫描 100 对中 5 对会"虚假协整"，没做 Bonferroni 校正
3. 执行成本：half-life 8 天意味着平均持仓 8 天，但回合策略转手频繁，被 fee 吞噬
4. β 估计不稳定：滚动 β 漂移会引入额外噪声
5. 流动性错配：一条腿冲击成本 > spread 利润
6. funding 成本：长腿/短腿 funding 不对称（DeFi 永续）

Q3：如何判断协整关系是否被破坏？

答：

• 实时监控指标：
  1. 滚动 ADF p-value（每天重计算 60-day 窗口），若 > 0.10 持续 5 天则警报
  2. 滚动 β 飘移（β 突变 > 30% 警报）
  3. spread 突破 4σ（结构断点）
• 基本面监控：
  • 协议升级公告（Shapella）、储备金公告（USDC）
  • Token unlock、监管事件
• 统计断点检验：Bai-Perron 多重断点检验，Chow 检验
• 风控规则：硬止损 4σ + 最大持仓时间 = 3 × half-life

Q4：为什么用对数价格而不是原始价格做协整检验？

答：

• 对数价格差 = 对数收益，量纲一致便于比较跨币种
• 对数价格的方差更稳定（异方差小）
• 残差经济意义：$\log(Y) - \beta \log(X)$ 是相对偏离百分比
• 对加密这种波动率几个数量级的资产更鲁棒

Q5：Engle-Granger vs Johansen 各适合什么场景？

答：

• Engle-Granger：两变量、关系明确（known dependent var）、计算简单。但 β 不对称（取决于哪个做被解释）
• Johansen：多变量（≥3）、检验秩（rank）即同时存在多少协整关系。VAR 框架，更严谨但样本要求 200+
• 加密实战：扫描阶段用 EG（快），最终入选配对用 Johansen 双重确认；3 资产以上篮子（如 LST 多腿）必须用 Johansen

---

明日预告

Day 90: Mean Reversion 深度 — Ornstein-Uhlenbeck 过程、Kalman 滤波动态对冲比、最优执行边界（Avellaneda-Stoikov）。今天的协整是基础，明天的 OU 是更精细的统计模型。