配对交易：从均值回归中获取市场中性收益

如果无论市场涨、跌还是横盘你都能赚钱呢？配对交易是对冲基金和量化机构使用的市场中性策略，从两个相关资产的相对表现中提取收益。下面讲它如何运作，以及如何避开那些让业余者爆仓的错误。

什么是配对交易？

核心思路： 找出两只历史上同步波动的股票。当它们背离时，赌它们会重新收敛。

交易做法：

• 做多落后的那只 （表现落后的股票）
• 做空领先的那只 （表现超前的股票）
• 等待收敛 （均值回归）
• 同时平掉两边仓位 当价差回到均值时

示例（百事 vs 可口可乐）：

• 历史上 PEP 与 KO 的价差维持在很窄的区间（业务相似，同属一个行业）
• 本月：KO 涨 8%，PEP 涨 2%（出现背离！）
• 交易： 做空 KO，做多 PEP（赌价差收窄）
• 下月：KO 涨 1%，PEP 涨 5%（收敛）
• 结果： KO 空头亏 1%，PEP 多头赚 3% = 净赚 2%（市场中性！）

市场中性：为什么重要

配对交易的妙处： 你不必在意市场的方向。

关键洞察： 你赚的是 相对 表现，而非绝对表现。这正是对冲基金钟爱配对交易的原因——它在任何市场环境下都成立。

相关性 vs 协整（关键区别）

多数业余配对交易者用的是相关性。 这是错的，而且会让你亏钱。

相关性（不该用的）

它衡量什么： 两只股票同步波动的紧密程度 ——在同一时间点上.

示例：

• SPY 与 QQQ 的相关性达 0.95（每日同步波动）
• 问题： 它们可以永远一起上涨，没有均值回归。
• 结果： 你做空 QQQ、做多 SPY，两只一直涨，价差永远不收敛。你亏钱。

它为何失效： 高相关性并不意味着两者之间的价差稳定。它们可以无限期地越走越远。

协整（你该用的）

它衡量什么： 衡量的是两只股票之间的 价差 是否具有均值回归性（平稳）。

技术定义： 两个非平稳时间序列（价格），其某个线性组合（价差）是平稳的。

大白话： PEP 和 KO 可以一直往上涨，但 它们之间的差额 会维持在一个可预测的区间内，并回归均值。

配对交易方法论（分步）

第 1 步：寻找候选配对

去哪里找：

• 同一行业： PEP/KO（饮料）、XOM/CVX（石油）、JPM/BAC（银行）
• 商业模式相近： SBUX/MCD（餐饮）、HD/LOW（家居建材）
• ETF 成分股： XLE（能源）或 XLF（金融）的前几大持仓
• 直接竞争者： 正面竞争的公司（WMT/TGT、BA/LMT）

快速筛选：

• 相关性 > 0.7（单靠它不够，但是个好起点）
• 市值相近（相差不超过 3 倍）
• 流动性好（各自日成交量 > 100 万股）

第 2 步：检验协整

Engle-Granger 两步法：

1. 做线性回归： 股票 A = β × 股票 B + 残差
   • 由此求出最优对冲比率（β）
2. 检验残差的平稳性： 使用增广迪基-富勒（ADF）检验
   • 若 p 值 < 0.05：存在协整！（价差具有均值回归性）
   • 若 p 值 > 0.05：不存在协整（别交易这一对）

Python 代码示例（检验 PEP/KO）：

import pandas as pd
import numpy as np
from statsmodels.tsa.stattools import adfuller
from statsmodels.regression.linear_model import OLS

# Load price data (assume you have this)
pep = data['PEP']  # Pepsi prices
ko = data['KO']    # Coca-Cola prices

# Step 1: Run regression to find hedge ratio
model = OLS(pep, ko).fit()
hedge_ratio = model.params[0]  # Beta (how many shares of KO per 1 share of PEP)
residuals = model.resid        # The spread

# Step 2: Test residuals for stationarity (ADF test)
adf_result = adfuller(residuals)
p_value = adf_result[1]

print(f"Hedge Ratio: {hedge_ratio:.4f}")
print(f"ADF p-value: {p_value:.4f}")

if p_value < 0.05:
    print("✅ Pair is COINTEGRATED - Good for pairs trading!")
else:
    print("❌ Pair is NOT cointegrated - Do not trade!")

第 3 步：计算价差（Z 分数）

确认协整之后，跟踪价差并做标准化：

价差公式：

Z 分数（标准化后的价差）：

解读：

• Z = 0： 价差处于历史均值（合理水平）
• Z = +2： 价差高于均值 2 个标准差（PEP 相对 KO 偏贵）
• Z = -2： 价差低于均值 2 个标准差（PEP 相对 KO 偏便宜）

第 4 步：入场规则

保守做法（±2 个标准差）：

激进做法（±1.5 个标准差）： 交易次数更多，但每笔的优势更小。

第 5 步：离场规则

止盈：

• Z 穿越零轴： 价差回到均值（保守）
• Z 达到 ±0.5： 抓住大部分行情，避免来回打脸（中庸）
• Z 反向移动 1 个标准差： 例如在 Z=+2.5 入场，Z=+1.5 离场（激进）

止损：

• Z 超过 ±3.0： 价差可能已经破裂（配对关系发生改变）
• 基于时间： 30 天内未收敛就离场（考虑机会成本）
• 协整关系破裂： 若滚动 ADF 的 p 值 > 0.10，立即离场

第 6 步：配对交易的仓位管理

金额中性做法：

• 总仓位规模： 在这一对上承担组合 1% 至 2% 的风险
• 等金额： 做多 1 万美元 PEP，做空 1 万美元 KO（尚未按对冲比率调整）
• 按对冲比率调整： 若对冲比率为 1.2，则每做多 1 万美元 PEP，就做空 1.2 万美元 KO

示例（10 万美元组合承担 1% 风险）：

• 每笔交易风险：1,000 美元
• 止损：Z = ±3.0（价差约扩大 10%）
• 仓位规模：每边 1 万美元（波动 10% = 亏损 1,000 美元）
• 做多 PEP：10,000 / 每股 150 = 67 股
• 做空 KO：12,000 / 每股 60 = 200 股（对冲比率 1.2）

Real 示例： PEP/KO (2020-2024 Backtest)

设置

• 配对： 百事（PEP）vs 可口可乐（KO）
• 区间： 2020 年 1 月至 2024 年 12 月（5 年）
• 回看窗口： 用 60 天滚动窗口计算价差
• 入场： Z 分数 ± 2.0
• 离场： Z 分数穿越 0
• 止损： Z 分数 ± 3.0 或满 30 天

结果

关键结论：

• 绝对回报更低： 7.3% 对 12.3%（意料之中——市场中性策略放弃了贝塔）
• 风险调整后好得多： 夏普 1.42 对 0.87（提升 63%）
• 回撤极小： -8.2% 对 -23.9%（安然睡过了新冠崩盘！）
• 波动率低： 4.8% 对 18.3%（波动小四倍）
• 真正的市场中性： 与 SPY 相关性仅 0.12（带来分散化收益）

Python 实现（完整代码）

# Pairs Trading Strategy - Full Implementation
# Author: Plan My Retire Finance University

import pandas as pd
import numpy as np
import yfinance as yf
from statsmodels.tsa.stattools import adfuller
from statsmodels.regression.linear_model import OLS
import matplotlib.pyplot as plt

class PairsTrading:
    """
    Statistical arbitrage pairs trading strategy

    Parameters:
    -----------
    window : int
        Lookback window for spread calculation (default 60 days)
    entry_z : float
        Z-score threshold for entry (default ±2.0)
    exit_z : float
        Z-score for exit (default 0.0, mean reversion)
    stop_z : float
        Z-score stop loss (default ±3.0)
    """

    def __init__(self, window=60, entry_z=2.0, exit_z=0.0, stop_z=3.0):
        self.window = window
        self.entry_z = entry_z
        self.exit_z = exit_z
        self.stop_z = stop_z

    def test_cointegration(self, stock_a, stock_b):
        """Test if two stocks are cointegrated"""
        # Run regression
        model = OLS(stock_a, stock_b).fit()
        hedge_ratio = model.params[0]
        residuals = model.resid

        # ADF test on residuals
        adf_result = adfuller(residuals)
        p_value = adf_result[1]

        return {
            'cointegrated': p_value < 0.05,
            'p_value': p_value,
            'hedge_ratio': hedge_ratio,
            'residuals': residuals
        }

    def calculate_spread(self, stock_a, stock_b, hedge_ratio):
        """Calculate price spread between pairs"""
        return stock_a - (hedge_ratio * stock_b)

    def calculate_zscore(self, spread):
        """Calculate rolling z-score of spread"""
        mean = spread.rolling(window=self.window).mean()
        std = spread.rolling(window=self.window).std()
        zscore = (spread - mean) / std
        return zscore

    def generate_signals(self, stock_a, stock_b, hedge_ratio):
        """Generate trading signals"""
        spread = self.calculate_spread(stock_a, stock_b, hedge_ratio)
        zscore = self.calculate_zscore(spread)

        signals = pd.DataFrame(index=stock_a.index)
        signals['spread'] = spread
        signals['zscore'] = zscore
        signals['position'] = 0

        # Entry signals
        signals.loc[zscore > self.entry_z, 'position'] = -1  # Short spread
        signals.loc[zscore < -self.entry_z, 'position'] = 1  # Long spread

        # Exit signals (mean reversion)
        # Close position when zscore crosses exit threshold
        for i in range(1, len(signals)):
            if signals['position'].iloc[i-1] != 0:
                # Check exit conditions
                if abs(signals['zscore'].iloc[i]) < abs(self.exit_z):
                    signals['position'].iloc[i] = 0
                elif abs(signals['zscore'].iloc[i]) > self.stop_z:
                    signals['position'].iloc[i] = 0  # Stop loss
                else:
                    # Hold position
                    signals['position'].iloc[i] = signals['position'].iloc[i-1]

        return signals

    def backtest(self, stock_a, stock_b, signals, hedge_ratio, initial_capital=100000):
        """Backtest pairs trading strategy"""
        # Calculate returns for each leg
        returns_a = stock_a.pct_change()
        returns_b = stock_b.pct_change()

        # Portfolio returns
        portfolio_returns = []

        for i in range(1, len(signals)):
            position = signals['position'].iloc[i-1]

            if position == 1:  # Long spread (long A, short B)
                ret = returns_a.iloc[i] - (hedge_ratio * returns_b.iloc[i])
            elif position == -1:  # Short spread (short A, long B)
                ret = -(returns_a.iloc[i] - (hedge_ratio * returns_b.iloc[i]))
            else:
                ret = 0

            portfolio_returns.append(ret)

        # Calculate equity curve
        portfolio_returns = pd.Series(portfolio_returns, index=signals.index[1:])
        equity_curve = initial_capital * (1 + portfolio_returns).cumprod()

        # Performance metrics
        total_return = (equity_curve.iloc[-1] / initial_capital) - 1
        years = (equity_curve.index[-1] - equity_curve.index[0]).days / 365.25
        cagr = (1 + total_return) ** (1/years) - 1
        volatility = portfolio_returns.std() * np.sqrt(252)
        sharpe = cagr / volatility if volatility > 0 else 0

        # Max drawdown
        cumulative = (1 + portfolio_returns).cumprod()
        running_max = cumulative.expanding().max()
        drawdown = (cumulative - running_max) / running_max
        max_dd = drawdown.min()

        # Trade statistics
        trades = (signals['position'].diff() != 0).sum() / 2  # Pairs of entries/exits
        wins = (portfolio_returns > 0).sum()
        losses = (portfolio_returns < 0).sum()
        win_rate = wins / (wins + losses) if (wins + losses) > 0 else 0

        return {
            'equity_curve': equity_curve,
            'total_return': total_return,
            'cagr': cagr,
            'volatility': volatility,
            'sharpe': sharpe,
            'max_drawdown': max_dd,
            'total_trades': int(trades),
            'win_rate': win_rate
        }

# Usage Example
if __name__ == "__main__":
    # Download data
    start = '2020-01-01'
    end = '2024-12-31'

    pep = yf.download('PEP', start=start, end=end)['Adj Close']
    ko = yf.download('KO', start=start, end=end)['Adj Close']

    # Initialize strategy
    strategy = PairsTrading(window=60, entry_z=2.0)

    # Test cointegration
    coint_result = strategy.test_cointegration(pep, ko)
    print(f"Cointegration p-value: {coint_result['p_value']:.4f}")
    print(f"Cointegrated: {coint_result['cointegrated']}")
    print(f"Hedge Ratio: {coint_result['hedge_ratio']:.4f}")

    if coint_result['cointegrated']:
        # Generate signals
        signals = strategy.generate_signals(pep, ko, coint_result['hedge_ratio'])

        # Backtest
        results = strategy.backtest(pep, ko, signals,
                                   coint_result['hedge_ratio'])

        # Print results
        print("\n" + "="*50)
        print("PAIRS TRADING BACKTEST RESULTS (PEP/KO)")
        print("="*50)
        print(f"Total Return:    {results['total_return']:>10.2%}")
        print(f"CAGR:            {results['cagr']:>10.2%}")
        print(f"Volatility:      {results['volatility']:>10.2%}")
        print(f"Sharpe Ratio:    {results['sharpe']:>10.2f}")
        print(f"Max Drawdown:    {results['max_drawdown']:>10.2%}")
        print(f"Total Trades:    {results['total_trades']:>10}")
        print(f"Win Rate:        {results['win_rate']:>10.2%}")
        print("="*50)
    else:
        print("❌ Pairs are NOT cointegrated. Do not trade!")

配对关系何时会破裂

警告信号

1. 协整关系减弱： 滚动 ADF 的 p 值升到 0.10 以上（不再显著）
2. 对冲比率不稳定： 60 天对冲比率偏离长期均值 >20%
3. 相关性下降： 60 天滚动相关性跌破 0.5
4. 业务分化： 其中一家公司改变战略、进入新市场，或被收购
5. 反复止损： 连续 3 笔以上交易触及止损

近期配对破裂的示例

发现破裂时该怎么做：

• 立即平掉所有未结仓位
• 停止交易该配对 3 至 6 个月
• 恢复交易前重新检验协整

最终要点

1. 市场中性 = 分散化： 与市场相关性 0.1 至 0.2（在只做多失效时依然管用）
2. 回报更低，夏普更好： 年化通常 7% 至 12%，但夏普 > 1.0（曲线平滑）
3. 协整至关重要： 只看相关性会亏钱。要做正规检验。
4. 均值回归需要时间： 平均每笔 15 至 30 天。要有耐心。
5. 配对会破裂： 每月监测协整关系。一旦减弱就离场。
6. 交易成本很重要： 每股成本需在 0.005 美元以内，否则优势荡然无存。
7. 不适合小账户： 需要保证金、卖空权限，以及 2.5 万美元以上资金（PDT 规则）
8. 与动量策略互补： 与双动量搭配，实现真正的分散

接下来： 趋势跟随系统——如何用移动平均线、突破和基于 ATR 的仓位管理去骑乘大级别趋势。回报高于配对交易，但波动也更大。