第45讲：案例·基础配对交易算法 (Example: Basic Pairs Trading Algorithm)

💡

查看全集：💎Quantopia量化分析56讲

一、核心概念解析

1.1 策略本质理解

配对交易通过捕捉两个相关资产的价格偏差获利。如同用弹簧连接的两个小球：

当价差拉伸时（弹簧被拉长）建仓

当价差恢复时（弹簧回弹）平仓

核心假设：价差具有均值回归特性

价差弹簧示意图

1.2 协整性数学表达

协整关系可表示为：

Yt=α+βXt+ϵtY_t = \alpha + \beta X_t + \epsilon_tYt=α+βXt+ϵt

其中残差项ϵt\epsilon_tϵt满足：

ϵt=Yt−(α+βXt)\epsilon_t = Y_t - (\alpha + \beta X_t)ϵt=Yt−(α+βXt)

需通过ADF检验验证ϵt\epsilon_tϵt的平稳性（p<0.05）

1.3 动态对冲演进

# 传统静态对冲（2010年前主流）
beta = 固定值

# 滚动窗口对冲（2010-2015常用）
rolling_beta = 30日滚动回归系数

# 状态空间模型（现代方法）
from statsmodels.tsa.statespace.sarimax import SARIMAX
model = SARIMAX(Y, exog=X, order=(1,0,0)).fit()

二、完整策略实现

2.1 数据获取与预处理

import yfinance as yf
import pandas as pd

# 获取数据（以麦当劳与百胜集团为例）
symbols = ['MCD', 'YUM']
data = yf.download(symbols, start='2018-01-01', end='2023-12-31')['Close']

# 数据清洗
data = data.dropna()
log_prices = np.log(data)  # 对数处理平稳化

# 可视化验证
data.plot(subplots=True, figsize=(12,6), title='Price Series')

2.2 协整检验进阶

from statsmodels.tsa.stattools import coint

def enhanced_coint_test(s1, s2):
    # 执行三次检验防止假阳性
    p1 = coint(s1, s2)[1]
    p2 = coint(s2, s1)[1]
    p3 = coint(s1.diff().dropna(), s2.diff().dropna())[1]
    return max(p1, p2), min(p1, p2), p3

p_max, p_min, p_diff = enhanced_coint_test(log_prices['MCD'],
                                         log_prices['YUM'])
print(f"最大协整p值: {p_max:.4f}, 差分序列p值: {p_diff:.4f}")

2.3 动态价差构建

# 滚动120天计算对冲比率
rolling_period = 120
hedge_ratios = []

for i in range(rolling_period, len(log_prices)):
    X = sm.add_constant(log_prices['MCD'].iloc[i-rolling_period:i])
    model = sm.OLS(log_prices['YUM'].iloc[i-rolling_period:i], X).fit()
    hedge_ratios.append(model.params['MCD'])

# 创建对冲比率序列
hedge_ratio_series = pd.Series(
    index=log_prices.index[rolling_period:],
    data=hedge_ratios
)

# 计算动态价差
spread = log_prices['YUM'] - hedge_ratio_series * log_prices['MCD']

2.4 信号生成优化

# 自适应布林通道
def adaptive_bollinger(series, window=30):
    ma = series.rolling(window).mean()
    std = series.rolling(window).std()
    upper = ma + std*(1 + 0.1*(window//30))  # 波动率放大
    lower = ma - std*(1 + 0.1*(window//30))
    return (series - ma)/std, upper, lower

zscore, upper, lower = adaptive_bollinger(spread.dropna())

# 信号规则
signals = pd.DataFrame(index=zscore.index)
signals['entry_short'] = zscore > 1.5
signals['entry_long'] = zscore < -1.5
signals['exit'] = abs(zscore) < 0.5

2.5 回测增强模块

# 头寸管理
positions = pd.DataFrame(index=signals.index)
positions['MCD'] = 0
positions['YUM'] = 0

# 信号执行（考虑交易延迟）
positions.loc[signals['entry_short'], 'MCD'] = 1
positions.loc[signals['entry_short'], 'YUM'] = -1
positions.loc[signals['entry_long'], 'MCD'] = -1
positions.loc[signals['entry_long'], 'YUM'] = 1
positions = positions.shift(1)  # 避免前瞻偏差

# 收益计算
returns = log_prices.pct_change().dropna()
strategy_returns = (positions['MCD'] * returns['MCD'] +
                    positions['YUM'] * returns['YUM'])

# 可视化
(1 + strategy_returns).cumprod().plot(title='Enhanced Strategy Performance')

三、关键问题诊断

3.1 协整关系破裂场景

# 断点检测（以2020年疫情为例）
from changepoint import Pelt

model = Pelt("rbf").fit(spread.values.reshape(-1,1))
change_points = model.predict(pen=10)

# 标记结构变化点
plt.plot(spread)
for cp in change_points:
    plt.axvline(x=spread.index[cp], color='r', linestyle='--')

3.2 交易成本影响分析

# 交易成本模型
def apply_transaction_cost(returns, positions, cost=0.0005):
    trades = positions.diff().abs()
    return returns - trades * cost

net_returns = apply_transaction_cost(strategy_returns, positions)

四、实战练习题

题目1：动态对冲验证

获取CVX（雪佛龙）和XOM（埃克森美孚）2015-2020年数据：

比较固定对冲比率与滚动对冲比率的表现差异

计算滚动窗口为60日与120日的夏普比率对比

题目2：市场状态过滤

在原有策略中加入：

# 使用SPY判断市场趋势
spy = yf.download('SPY', start='2018-01-01')['Close']
market_up = spy.pct_change(30) > 0.03

# 只在市场震荡期交易
signals['valid'] = ~market_up

五、风险控制方案

5.1 止损机制

# 动态跟踪止损
max_drawdown = 0.05  # 5%最大回撤
peak = (1 + strategy_returns).cumprod().cummax()
drawdown = (peak - (1 + strategy_returns).cumprod())/peak
signals['stop_loss'] = drawdown > max_drawdown

5.2 头寸规模优化

# 波动率倒数加权
volatility = returns.rolling(30).std()
position_size = 1 / volatility
position_size = position_size.div(position_size.sum(axis=1), axis=0)

通过本案例，您可系统掌握配对交易的实现要点与优化方向。建议在QuantConnect平台进行实时市场验证，并参考《Pairs Trading: Quantitative Methods and Analysis》深化理论理解。

第44讲：配对交易入门 (Introduction to Pairs Trading)

第46讲：案例·高级配对交易算法 (Example: Advanced Pairs Trading Algorithm)