相关性分析：皮尔逊相关系数、协整检验、滚动窗口相关性、实战案例

各位同学，今天我们来聊聊跨品种套利中最核心的一步——相关性分析。

说实话，很多人做套利一上来就找价格走势长得像的品种，然后直接开干。我见过太多这样的案例了，结果呢？要么是伪相关，要么是结构突变后亏得底掉。嗯，这里面的门道，我们今天一次性说清楚。

皮尔逊相关系数：最基础，但别乱用

皮尔逊相关系数，说白了就是衡量两个变量线性关系的强弱。值在-1到1之间，越接近±1，线性关系越强。

公式长这样：

ρ(X,Y) = Cov(X,Y) / (σX * σY)

但在实际项目中，我一般不会直接用价格算相关系数。为什么？因为价格序列通常是非平稳的，算出来的相关系数可能虚高。我习惯用收益率序列来算。

来看一段Python代码：

import numpy as np
import pandas as pd

# 假设df有两列：'RB'和'HC'，分别是螺纹钢和热卷的日收益率
corr = df['RB'].corr(df['HC'])
print(f"皮尔逊相关系数: {corr:.4f}")

# 滚动窗口计算
rolling_corr = df['RB'].rolling(window=60).corr(df['HC'])

我个人习惯，先用60个交易日的滚动窗口扫一遍。如果相关系数长期稳定在0.7以上，才考虑进一步分析。

协整检验：这才是套利的灵魂

相关系数高不代表能套利。你想想看，两只股票都跟着大盘涨跌，相关系数可能很高，但它们之间没有稳定的价差关系，没法做配对交易。

协整检验解决的就是这个问题——它检验两个时间序列是否存在长期稳定的线性关系。

最常用的是Engle-Granger两步法：

1. 第一步：用OLS回归估计协整系数
2. 第二步：对残差进行单位根检验（ADF检验）

代码实现：

import statsmodels.api as sm
from statsmodels.tsa.stattools import adfuller

# 假设price1和price2是价格序列
def cointegration_test(price1, price2):
    # 第一步：OLS回归
    X = sm.add_constant(price2)
    model = sm.OLS(price1, X).fit()
    residuals = model.resid
    
    # 第二步：ADF检验残差
    adf_stat, p_value, _, _, critical_values, _ = adfuller(residuals, maxlag=1)
    
    return p_value, model.params[1]  # 返回p值和协整系数

p_val, hedge_ratio = cointegration_test(df['RB'], df['HC'])
print(f"协整检验p值: {p_val:.4f}")

滚动窗口相关性：捕捉结构变化

市场是动态的。两个品种的相关性不会一成不变。我见过螺纹钢和热卷的相关性从0.9掉到0.3，只用了两个月。如果你用全样本算一个固定值，那基本等于刻舟求剑。

滚动窗口分析就是解决这个问题的。核心参数有两个：

• 窗口大小：太短容易受噪音干扰，太长又反应迟钝。我个人习惯用60-120个交易日
• 步长：一般用1，每天更新

来看一个完整的滚动分析框架：

def rolling_analysis(df, window=60):
    results = pd.DataFrame(index=df.index)
    
    # 滚动相关系数
    results['rolling_corr'] = df['ret1'].rolling(window).corr(df['ret2'])
    
    # 滚动协整检验
    results['coint_pvalue'] = np.nan
    for i in range(window, len(df)):
        sub1 = df['price1'].iloc[i-window:i]
        sub2 = df['price2'].iloc[i-window:i]
        p_val, _ = cointegration_test(sub1, sub2)
        results['coint_pvalue'].iloc[i] = p_val
    
    return results

实战案例：螺纹钢与热卷套利

好了，理论说完了，我们来看一个完整的实战案例。我拿螺纹钢（RB）和热卷（HC）来演示，这两个品种在黑色系里相关性很高，但价差经常出现偏离。

第一步：数据准备

import tushare as ts
# 获取近3年日线数据
rb = ts.get_k_data('RB9999', start='2021-01-01', end='2024-01-01')
hc = ts.get_k_data('HC9999', start='2021-01-01', end='2024-01-01')

# 合并数据
data = pd.DataFrame({
    'RB': rb['close'].values,
    'HC': hc['close'].values
}, index=pd.to_datetime(rb['date']))

# 计算收益率
data['ret_RB'] = data['RB'].pct_change()
data['ret_HC'] = data['HC'].pct_change()

第二步：相关性扫描

我一般会先看全样本相关系数，再看滚动窗口。如果两者差异不大，说明结构稳定。

full_corr = data['ret_RB'].corr(data['ret_HC'])
print(f"全样本相关系数: {full_corr:.4f}")

# 滚动60天
rolling_corr = data['ret_RB'].rolling(60).corr(data['ret_HC'])
print(f"滚动相关系数均值: {rolling_corr.mean():.4f}")
print(f"滚动相关系数标准差: {rolling_corr.std():.4f}")

第三步：协整检验与价差计算

p_val, hedge_ratio = cointegration_test(data['RB'], data['HC'])
print(f"协整p值: {p_val:.4f}, 对冲比率: {hedge_ratio:.4f}")

# 计算价差
data['spread'] = data['RB'] - hedge_ratio * data['HC']

# 标准化价差
data['zscore'] = (data['spread'] - data['spread'].rolling(60).mean()) / data['spread'].rolling(60).std()

第四步：交易信号生成

# 当zscore超过±2时开仓
data['signal'] = 0
data.loc[data['zscore'] > 2, 'signal'] = -1  # 做空价差
data.loc[data['zscore'] < -2, 'signal'] = 1  # 做多价差

# 当zscore回归到0附近平仓
data.loc[abs(data['zscore']) < 0.5, 'signal'] = 0

知识体系总览

下面这张图，是我自己整理的相关性分析知识框架。你把它存下来，以后做任何套利分析，按这个流程走基本不会出错。

最后说一句，相关性分析不是一劳永逸的事。我每周都会重新跑一遍滚动协整检验，看看结构有没有变化。市场在变，你的模型也得跟着变。