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

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

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

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

import yfinance as yf
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import statsmodels.api as sm

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

# 数据清洗
data = data.dropna()
log_prices = np.log(data)  # 对数变换便于解释相对变化，但不会自动使序列平稳

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

from statsmodels.tsa.stattools import coint

def enhanced_coint_test(s1, s2):
    # Engle-Granger 检验有回归方向，先按经济关系预先指定因变量
    statistic, pvalue, critical_values = coint(s1, s2)
    return statistic, pvalue, critical_values

statistic, pvalue, critical_values = enhanced_coint_test(
    log_prices['YUM'], log_prices['MCD']
)
print(f"协整检验 p 值: {pvalue:.4f}")

# 滚动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']

# 自适应布林通道
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

# 用状态机让仓位从入场持续到退出。state=1 表示做多价差：
# 做多 YUM、按滚动对冲比率做空 MCD。
positions = pd.DataFrame(0.0, index=signals.index, columns=['MCD', 'YUM'])
state = 0
for date in signals.index:
    if signals.at[date, 'exit']:
        state = 0
    elif signals.at[date, 'entry_short']:
        state = -1
    elif signals.at[date, 'entry_long']:
        state = 1
    positions.at[date, 'YUM'] = state
    positions.at[date, 'MCD'] = -state * hedge_ratio_series.at[date]

# t 日收盘后生成的信号最早在 t+1 日持有
positions = positions.shift(1).fillna(0)

# 收益计算
returns = data.pct_change().reindex(positions.index).fillna(0)
strategy_returns = (positions['MCD'] * returns['MCD'] +
                    positions['YUM'] * returns['YUM'])

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

# 断点检测（以2020年疫情为例）
import ruptures as rpt

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

# 标记结构变化点
plt.plot(spread)
for cp in change_points:
    if cp < len(clean_spread):  # ruptures 会把序列终点也作为分段边界返回
        plt.axvline(x=clean_spread.index[cp], color='r', linestyle='--')

# 交易成本模型
def apply_transaction_cost(returns, positions, cost=0.0005):
    one_way_turnover = positions.diff().abs().sum(axis=1).fillna(0)
    return returns - one_way_turnover * cost

net_returns = apply_transaction_cost(strategy_returns, positions)

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

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

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

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