4. 价差计算引擎：价差计算公式、滚动窗口计算、标准化处理
好，咱们进入价差计算引擎这个核心模块。
说实话，很多做统计套利的朋友，策略逻辑没问题，但最后亏钱就亏在价差计算这一步。不是公式用错了，就是窗口选得不对，再或者标准化处理出了偏差。我当年刚入行时也踩过这个坑——记得有一次回测曲线漂亮得不行，实盘一跑就崩，查了三天才发现是价差计算时用了未来数据。
所以这一节，咱们把价差计算的每个细节都掰开揉碎。你跟着我走一遍，保证以后不会再犯低级错误。
4.1 价差的核心公式
价差，说白了就是两个相关资产价格之间的差值。但这里有个关键点：价差不是简单的价格相减。
最常见的价差公式有两种：
| 类型 | 公式 | 适用场景 |
|---|---|---|
| 简单价差 | Spread = P₁ - P₂ | 同品种、同面值（如：两个ETF） |
| 回归价差 | Spread = P₁ - β × P₂ | 不同品种、不同波动率（如：股票对） |
我个人习惯用回归价差。为什么呢？因为两个资产的价格量级往往不一样。比如茅台和五粮液，价格差了几百块，直接相减得到的价差序列，均值可能一直在漂移，根本没法用。
β 怎么算？通常用历史数据做线性回归。但这里有个小陷阱——β 不是一成不变的。我在项目中遇到过，某个股票对的 β 在半年内从 1.2 漂移到了 0.8，如果还用固定 β，价差信号全是错的。
4.2 滚动窗口计算
好，公式有了，接下来就是怎么算。这里引入一个概念——滚动窗口。
你想想看，价差的统计特征（均值、标准差）是随时间变化的。如果用全量历史数据去算，那得到的是一个「平均状态」，根本反映不了当前的市场环境。
滚动窗口的思路很简单：只取最近 N 个数据点来计算。窗口不断向前滑动，每次只保留最新的 N 个样本。
代码实现也不复杂：
def rolling_spread(price1, price2, window=60, beta=1.0):
"""
滚动计算价差
:param price1: 资产1价格序列
:param price2: 资产2价格序列
:param window: 滚动窗口大小
:param beta: 对冲系数
:return: 价差序列
"""
spread = pd.Series(index=price1.index, dtype=float)
for i in range(window, len(price1)):
# 取窗口内的数据
p1_window = price1.iloc[i-window:i]
p2_window = price2.iloc[i-window:i]
# 计算窗口内的价差
spread.iloc[i] = p1_window.iloc[-1] - beta * p2_window.iloc[-1]
return spread这里 window 参数怎么选？嗯，这是个好问题。我一般遵循两个原则：
- 统计原则：窗口至少要有 30 个样本，否则统计结果不可靠
- 业务原则：窗口要覆盖至少一个完整的均值回归周期
举个例子，如果你做的是 5 分钟级别的期货价差，一个回归周期大概 2-3 小时，那窗口选 30-40 根 K 线就差不多。如果是日线级别的股票对，一个回归周期可能 20-30 天，窗口就选 60 天左右。
4.3 标准化处理
价差算出来了，但直接拿原始价差值去判断「是否偏离」是不行的。为什么？因为价差的绝对数值没有意义。
比如某个股票对的价差均值是 5，标准差是 2。那价差跑到 7 算不算偏离？跑到 10 呢？你没法直观判断。
所以我们需要标准化——把价差转换成 Z-score：
Z = (当前价差 - 窗口均值) / 窗口标准差
标准化后的 Z-score 有几个好处：
- 无量纲：不同品种的价差可以放在一起比较
- 直观：Z=2 意味着当前价差偏离均值 2 个标准差，这是明显的开仓信号
- 稳定：Z-score 的均值为 0，标准差为 1，方便设置统一的阈值
代码实现：
def calculate_zscore(spread, window=60):
"""
滚动计算 Z-score
:param spread: 原始价差序列
:param window: 滚动窗口
:return: Z-score 序列
"""
rolling_mean = spread.rolling(window=window).mean()
rolling_std = spread.rolling(window=window).std()
zscore = (spread - rolling_mean) / rolling_std
return zscore这里有个细节要注意：滚动均值和滚动标准差的计算，一定要用相同窗口。我见过有人用 60 天的均值，但用 30 天的标准差，结果 Z-score 忽大忽小，完全没法用。
if rolling_std == 0: zscore = 0
4.4 完整的价差计算流程
好，我们把上面三个步骤串起来，看看完整的价差计算引擎长什么样：
class SpreadEngine:
def __init__(self, window=60, beta_update_freq=5):
self.window = window
self.beta = 1.0
self.beta_update_freq = beta_update_freq # 每 N 个 tick 更新一次 beta
self.tick_count = 0
def update_beta(self, price1, price2):
"""更新对冲系数"""
# 用最近 window 个数据做线性回归
from scipy import stats
slope, _, _, _, _ = stats.linregress(price2[-self.window:],
price1[-self.window:])
self.beta = slope
def calculate(self, price1, price2):
"""计算标准化价差"""
self.tick_count += 1
# 定期更新 beta
if self.tick_count % self.beta_update_freq == 0:
self.update_beta(price1, price2)
# 1. 计算回归价差
spread = price1.iloc[-1] - self.beta * price2.iloc[-1]
# 2. 滚动窗口统计
spread_window = (price1 - self.beta * price2).iloc[-self.window:]
mean = spread_window.mean()
std = spread_window.std()
# 3. 标准化
if std == 0:
zscore = 0.0
else:
zscore = (spread - mean) / std
return {
'spread': spread,
'zscore': zscore,
'beta': self.beta,
'mean': mean,
'std': std
}这个引擎的核心逻辑就三句话：算价差 → 滚动统计 → 标准化。但每一步都有细节，每一步都可能出错。
- 价差公式：用回归价差，β 要动态更新
- 滚动窗口：窗口大小要匹配回归周期，不是越大越好
- 标准化：Z-score 让价差信号可比较、可量化
嗯，价差计算引擎就讲到这里。你把这个模块写好了，后面的信号生成和交易执行才有可靠的基础。别急着往下走，先把这段代码跑通，看看你的品种的 Z-score 长什么样——我敢打赌，你会发现一些意想不到的规律。