6. 协整检验实战：Engle-Granger两步法、Johansen检验的Python实现
协整检验，说白了就是帮我们判断两个或多个价格序列之间，是否存在长期稳定的线性关系。在基差交易里，这就是核心中的核心——没有协整关系，统计套利就是空中楼阁。
我个人习惯把协整检验分成两类来理解：一类是检验两个品种之间的协整关系，用Engle-Granger两步法；另一类是检验多个品种之间的协整关系，用Johansen检验。今天我们就来手撕这两种方法的Python实现。
6.1 为什么需要协整检验？
你想想看，我们做基差交易，本质上是赌价差会回归均值。但如果两个品种的价格走势根本就是各走各的，那价差就没有均值可回归了。协整检验就是帮我们确认：这两个品种是不是真的「绑在一起」的。
我在项目中遇到过这样的情况：螺纹钢和热卷看起来走势很像，但一跑协整检验，发现根本不协整。如果当时直接上套利策略，后果可想而知。
核心要点：协整 ≠ 相关。两个高度相关的序列，可能完全不协整。比如两个随机游走序列，相关系数可能很高，但它们的价差会发散，无法回归。
6.2 Engle-Granger两步法
这个方法名字听着唬人，其实逻辑很直白。就两步：
- 第一步：用OLS回归估计协整系数
- 第二步：对残差进行单位根检验（ADF检验）
如果残差是平稳的，那这两个序列就协整。就这么简单。
6.2.1 Python实现
import numpy as np
import pandas as pd
import statsmodels.api as sm
from statsmodels.tsa.stattools import adfuller
def engle_granger_test(y1, y2):
"""
Engle-Granger两步法协整检验
y1, y2: 两个价格序列
"""
# 第一步：OLS回归
X = sm.add_constant(y2)
model = sm.OLS(y1, X).fit()
residuals = model.resid
# 第二步：对残差做ADF检验
adf_stat, p_value, _, _, critical_values, _ = adfuller(residuals, maxlag=1)
return {
'cointegrating_vector': [1, -model.params[1]],
'intercept': model.params[0],
'adf_statistic': adf_stat,
'p_value': p_value,
'critical_values': critical_values,
'is_cointegrated': p_value < 0.05
}
# 示例：检验螺纹钢和热卷的协整关系
# 假设我们有价格数据
np.random.seed(42)
n = 500
# 构造协整序列
error = np.random.randn(n)
y2 = np.cumsum(np.random.randn(n)) # 随机游走
y1 = 0.8 * y2 + error # 协整关系
result = engle_granger_test(y1, y2)
print(f"ADF统计量: {result['adf_statistic']:.4f}")
print(f"p值: {result['p_value']:.4f}")
print(f"是否协整: {result['is_cointegrated']}")
避坑指南：我曾经在实盘数据上吃过亏——ADF检验的滞后阶数选择很关键。默认的AIC准则有时会选太多滞后，导致检验功效下降。我建议至少试3个不同的滞后阶数，看结果是否稳健。
6.3 Johansen检验
当我们要检验三个或更多品种之间的协整关系时，Engle-Granger两步法就不够用了。这时候需要上Johansen检验。
Johansen检验的核心思想是：通过向量误差修正模型（VECM），同时检验多个协整关系。它给出两个统计量：迹统计量和最大特征值统计量。
6.3.1 Python实现
from statsmodels.tsa.vector_ar.vecm import coint_johansen
def johansen_test(data, det_order=1, k_ar_diff=1):
"""
Johansen协整检验
data: 多列价格序列，DataFrame格式
det_order: 确定性趋势项，0-无，1-常数项，2-趋势项
k_ar_diff: 差分滞后阶数
"""
result = coint_johansen(data, det_order, k_ar_diff)
# 迹统计量
trace_stat = result.trace_stat
trace_crit = result.trace_stat_crit_vals
# 最大特征值统计量
max_eigen_stat = result.max_eigen_stat
max_eigen_crit = result.max_eigen_stat_crit_vals
# 协整向量
coint_vecs = result.evec
return {
'trace_statistic': trace_stat,
'trace_critical_values': trace_crit,
'max_eigen_statistic': max_eigen_stat,
'max_eigen_critical_values': max_eigen_crit,
'cointegrating_vectors': coint_vecs,
'rank': result.r
}
# 示例：检验三个品种的协整关系
# 构造数据
np.random.seed(42)
n = 500
e1 = np.random.randn(n)
e2 = np.random.randn(n)
e3 = np.random.randn(n)
# 构造两个协整关系
y1 = e1
y2 = 0.5 * y1 + e2
y3 = 0.3 * y1 + 0.2 * y2 + e3
data = pd.DataFrame({
'y1': np.cumsum(y1),
'y2': np.cumsum(y2),
'y3': np.cumsum(y3)
})
result = johansen_test(data)
print("迹统计量:", result['trace_statistic'])
print("迹统计量临界值(95%):", result['trace_critical_values'][:, 1])
print("协整秩:", result['rank'])
6.4 两种方法的对比
| 对比维度 | Engle-Granger两步法 | Johansen检验 |
|---|---|---|
| 适用场景 | 两个品种 | 多个品种（≥2） |
| 协整关系数量 | 最多1个 | 可检测多个 |
| 计算复杂度 | 低 | 较高 |
| 小样本表现 | 偏差较大 | 相对稳健 |
| Python实现 | statsmodels + adfuller | statsmodels的coint_johansen |
重要提醒：Johansen检验对滞后阶数非常敏感。我建议用信息准则（AIC/BIC）选择滞后阶数，然后做敏感性分析。如果不同滞后阶数下结论不一致，那就要小心了——可能数据本身就不适合做协整分析。
6.5 实战中的注意事项
- 数据频率要一致：别拿日线数据和小时数据混着做协整检验，那会出大问题。
- 样本量要足够：我个人建议至少250个数据点（约1年交易日），太少了检验功效不够。
- 结构性断点：如果样本期内发生了政策变化或市场结构变化，协整关系可能已经变了。我习惯用滚动窗口检验，看看协整关系是否稳定。
- 不要过度拟合：协整检验通过不代表一定能赚钱。我见过太多人拿着历史数据跑出协整关系，实盘一跑就崩——因为样本外协整关系可能已经失效了。
6.6 知识体系图
嗯，到这里协整检验的核心内容就讲完了。记住一点：协整检验只是工具，不是目的。我们真正要的是找到稳定可交易的价差关系。下次当你跑完协整检验看到p值小于0.05时，别急着高兴——先问问自己：这个关系在样本外还能不能站得住？