1. 配对交易基础

任务目标

选取两只具有长期均衡关系的股票，通过协整检验确认其配对有效性，计算价差序列并设定交易信号，完成基础配对交易策略的回测框架搭建。

环境依赖

需要 numpy、pandas、statsmodels、matplotlib。若需获取真实行情数据，可接入 yfinance 或本地CSV数据源；本例使用内置模拟数据保证可复现性。

核心实现

import numpy as np
import pandas as pd
import statsmodels.api as sm
from statsmodels.tsa.stattools import coint, adfuller

# ---------- 1. 生成模拟价格序列（真实场景替换为行情数据） ----------
np.random.seed(42)
n = 500
# 共同随机游走趋势项
common = np.cumsum(np.random.randn(n)) + 100
# 两只股票围绕共同趋势波动，价差为平稳过程
spread_true = 0.5 * np.sin(np.arange(n) / 20) + np.random.randn(n) * 0.3
price_a = common + spread_true
price_b = common - spread_true + 2  # 常数偏移不影响协整

df = pd.DataFrame({'A': price_a, 'B': price_b})

# ---------- 2. 协整检验 ----------
# 方法1：Engle-Granger 两步法（使用 statsmodels 的 coint 函数）
score, pvalue, _ = coint(df['A'], df['B'])
print(f"协整检验 p-value: {pvalue:.4f}")
is_cointegrated = pvalue < 0.05
print(f"是否存在协整关系（5%显著性水平）: {is_cointegrated}")

# 方法2：对价差序列做 ADF 平稳性检验（EG两步法的第二步）
# 先通过 OLS 估计协整系数
X = sm.add_constant(df['B'])
ols_model = sm.OLS(df['A'], X).fit()
hedge_ratio = ols_model.params['B']  # 对冲比率
spread = df['A'] - hedge_ratio * df['B']  # 价差序列

adf_stat, adf_pvalue, _ = adfuller(spread, autolag='AIC')
print(f"价差序列 ADF p-value: {adf_pvalue:.4f}")

# ---------- 3. 交易信号生成与回测 ----------
spread_mean = spread.mean()
spread_std = spread.std()
entry_threshold = 1.5  # 开仓阈值（单位：标准差）
exit_threshold = 0.5   # 平仓阈值

# 生成信号：Z-score 超过阈值开仓，回归均值附近平仓
zscore = (spread - spread_mean) / spread_std

position = 0  # 1=做多价差（多A空B），-1=做空价差（空A多B）
signals = []
for z in zscore:
    if position == 0:
        if z > entry_threshold:
            position = -1  # 价差过高，做空价差
        elif z < -entry_threshold:
            position = 1   # 价差过低，做多价差
    else:
        if abs(z) < exit_threshold:
            position = 0   # 价差回归，平仓
    signals.append(position)

df['signal'] = signals
df['spread'] = spread
df['zscore'] = zscore

# 回测收益计算（简化：每期价差变化 × 持仓方向）
df['spread_change'] = df['spread'].diff().shift(-1)  # 下一期价差变化
df['strategy_return'] = df['signal'] * df['spread_change']
cumulative_return = df['strategy_return'].fillna(0).cumsum()

# ---------- 4. 输出关键结果 ----------
print(f"对冲比率 (hedge ratio): {hedge_ratio:.4f}")
print(f"价差均值: {spread_mean:.4f}, 标准差: {spread_std:.4f}")
print(f"交易次数: {(df['signal'].diff() != 0).sum()}")
print(f"策略累计收益（价差单位）: {cumulative_return.iloc[-2]:.4f}")  # 最后一行无收益

# 可视化（可选）
import matplotlib.pyplot as plt
fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(10, 6))
ax1.plot(df.index, df['A'], label='Stock A')
ax1.plot(df.index, df['B'], label='Stock B')
ax1.set_ylabel('Price')
ax1.legend()
ax2.plot(df.index, zscore, label='Z-score', color='purple')
ax2.axhline(entry_threshold, color='red', linestyle='--', label='Entry')
ax2.axhline(-entry_threshold, color='red', linestyle='--')
ax2.axhline(exit_threshold, color='green', linestyle='--', label='Exit')
ax2.axhline(-exit_threshold, color='green', linestyle='--')
ax2.set_ylabel('Z-score')
ax2.legend()
plt.tight_layout()
plt.show()

运行方式

直接运行上述Python脚本。若使用真实数据，将 df 替换为包含两列价格数据的DataFrame，并确保索引为日期时间类型。

验证方法

• 协整检验通过条件：coint 函数返回的 p-value < 0.05，且价差序列的 ADF 检验 p-value < 0.05，说明两只股票存在长期均衡关系。
• 信号有效性：观察输出的交易次数应大于0，说明策略在样本期内触发了开平仓。
• 回测合理性：策略累计收益应围绕0波动（模拟数据中价差为均值回归过程，策略应能捕获部分回归收益），且收益曲线应呈现平稳上升趋势而非单边亏损。

常见问题

• 协整检验不通过：尝试更换股票对或延长数据窗口；也可对价格取对数后再检验。
• 信号过于频繁：适当增大 entry_threshold（如2.0）或增加 exit_threshold 与开仓阈值的距离。
• 实盘注意事项：本例未考虑交易成本、滑点和借贷费用，实盘需在 strategy_return 中扣除相应成本。