4. 价差计算与标准化：价差序列构建、Z-score标准化、滚动窗口计算

好，咱们进入第四章。价差计算与标准化。

说实话，很多做配对交易的朋友，最后亏钱不是因为策略不对，而是价差算错了。我见过有人直接用价格相减，结果两个品种价格差了几百块，Z-score 算出来全是异常值。嗯，这其实是个坑。

这一章，我会把价差序列怎么构建、怎么标准化、怎么用滚动窗口动态更新，掰开揉碎了讲清楚。你跟着我走一遍，后面写策略心里就有底了。

4.1 价差序列的构建：从价格到价差

价差，说白了就是两个品种之间的“距离”。但怎么量这个距离，有讲究。

常见的价差构建方式有三种：

方法	公式	适用场景
简单价差	spread = P₁ - P₂	同品种、同面值（如股指期货近远月）
比率价差	spread = P₁ / P₂	价格量级差异大（如股票 vs 股票）
回归残差	spread = P₁ - β × P₂	协整关系明确，需要对冲比率

我个人习惯用回归残差。为什么？因为简单价差受价格绝对值影响太大。举个例子，茅台和五粮液，价格差几百块，你拿简单价差做 Z-score，均值漂移一下你就懵了。

构建价差序列的代码，我一般这样写：

import numpy as np
import pandas as pd
import statsmodels.api as sm

# 假设 df 包含两列：'asset1' 和 'asset2'
def build_spread(df, method='ols'):
    if method == 'simple':
        spread = df['asset1'] - df['asset2']
    elif method == 'ratio':
        spread = df['asset1'] / df['asset2']
    elif method == 'ols':
        # 用 OLS 回归估计对冲比率
        X = sm.add_constant(df['asset2'])
        model = sm.OLS(df['asset1'], X).fit()
        beta = model.params['asset2']
        spread = df['asset1'] - beta * df['asset2']
    return spread

4.2 Z-score 标准化：让价差变得可比较

价差序列建好了，但不同时间段的价差均值、标准差都不一样。你没法直接说“价差大于 5 就开仓”，因为 5 在某个时段可能是极端值，在另一个时段可能就是正常波动。

这时候就需要 Z-score 标准化。公式很简单：

Z = (当前价差 - 均值) / 标准差

Z-score 的含义是：当前价差偏离均值多少个标准差。一般 Z > 2 或 Z < -2 就认为是极端值，可以考虑反向开仓。

我曾经在回测时犯过一个错：用全量数据算均值和标准差。结果呢？Z-score 在早期全是 0 附近，后期全是极端值。为什么？因为均值漂移了。所以，标准化一定要用滚动窗口。

def zscore_spread(spread, window=60):
    rolling_mean = spread.rolling(window=window).mean()
    rolling_std = spread.rolling(window=window).std()
    zscore = (spread - rolling_mean) / rolling_std
    return zscore

4.3 滚动窗口计算：动态更新，适应市场变化

滚动窗口是价差标准化的灵魂。市场在变，价差的均值和波动率也在变。你用半年前的均值来算今天的 Z-score，那跟刻舟求剑没区别。

窗口大小怎么选？ 我一般遵循两个原则：

• 统计显著性：窗口至少要有 30-60 个数据点，否则均值和标准差不稳定。
• 交易频率：日内交易用 20-30 期，日线交易用 60-120 期。

举个例子，如果你做 5 分钟级别的配对交易，窗口设 20 期（约 1.6 小时）就够用了。太长的话，市场结构变了你还用旧参数，容易吃大亏。

滚动窗口计算的完整流程，我画了个图，你一看就明白：

你看，滚动窗口的核心就是：每次只取最近 N 个数据点，算均值和标准差，然后对当前价差做标准化。窗口不断向前滑动，参数实时更新。

4.4 实战中的几个坑

嗯，这里我要多说几句。实战中价差标准化有几个容易踩的坑：

• 窗口太短：统计不稳定，Z-score 频繁触发假信号。我曾经用 10 期窗口做 1 分钟线，结果一天交易了 50 次，手续费都亏麻了。
• 窗口太长：对市场变化反应迟钝，价差已经回归了你还以为在极端位置。
• 忽略数据质量：价差序列里如果有异常跳空，Z-score 会瞬间爆表。我一般会先做一次中位数滤波，把 3 倍标准差以外的点替换掉。

4.5 完整代码示例

最后，给你一个完整的价差计算与标准化函数。我平时做研究就用这个模板：

def spread_pipeline(df, col1='asset1', col2='asset2', window=60, method='ols'):
    """
    完整的价差计算与标准化流程
    返回：原始价差、Z-score、滚动均值、滚动标准差
    """
    # 1. 构建价差
    if method == 'simple':
        spread = df[col1] - df[col2]
    elif method == 'ratio':
        spread = df[col1] / df[col2]
    elif method == 'ols':
        X = sm.add_constant(df[col2])
        model = sm.OLS(df[col1], X).fit()
        beta = model.params[col2]
        spread = df[col1] - beta * df[col2]
    
    # 2. 滚动窗口标准化
    rolling_mean = spread.rolling(window=window).mean()
    rolling_std = spread.rolling(window=window).std()
    zscore = (spread - rolling_mean) / rolling_std
    
    # 3. 返回结果
    result = pd.DataFrame({
        'spread': spread,
        'zscore': zscore,
        'rolling_mean': rolling_mean,
        'rolling_std': rolling_std
    })
    return result

# 使用示例
result = spread_pipeline(df, window=60, method='ols')
print(result.tail())

好了，价差计算与标准化就讲到这里。你把这个流程跑通，后面做信号生成和回测就顺了。

---