第4章价差计算与实时监控

第4章：价差计算与实时监控

做价差交易，核心就一句话：盯着两个价格之间的差距，等它偏离正常范围再动手。但怎么定义这个差距？怎么实时盯着它？怎么判断它是不是该回归了？这些才是真功夫。

我刚开始做价差策略时，以为就是简单的P1减P2。后来发现，事情远没那么简单。今天我把这些坑和经验都摊开来讲。

4.1 价差公式定义：Spread = P1 - P2

价差公式看起来简单：Spread = P1 - P2。但这里有个关键问题——P1和P2到底是什么价格？

我个人习惯用最新成交价（Last Price），因为它反应最快。但要注意，如果两个品种的流动性差异很大，用最新价可能会被瞬间的异常成交带偏。

价差计算的三种常见方式：

最新价差：LastPrice(P1) - LastPrice(P2)，反应最快，但容易受噪声影响
中间价差：(Bid1+Ask1)/2 - (Bid2+Ask2)/2，更稳定，但有一定延迟
加权价差：考虑买卖盘深度，计算更精确，但计算量大

我在项目中遇到过一个问题：用最新价差做监控，结果频繁触发交易信号。后来一查，原来是某个合约的最后一笔成交是异常单。从那以后，我改用中间价差做监控，最新价差只做辅助参考。

我的建议：生产环境中，用中间价差做主监控，最新价差做辅助。这样既保证了速度，又过滤了噪声。

4.2 实时价差序列构建

有了价差公式，下一步就是构建实时序列。说白了，就是把每一时刻的价差记录下来，形成一个时间序列。

嗯，这里要注意：序列的采样频率很关键。你想想看，高频交易中，毫秒级别的变化都可能影响决策。

# Python示例：实时价差序列构建
class SpreadMonitor:
    def __init__(self, window_size=1000):
        self.spread_series = deque(maxlen=window_size)
        self.timestamp_series = deque(maxlen=window_size)
    
    def update(self, price1, price2, timestamp):
        spread = price1 - price2
        self.spread_series.append(spread)
        self.timestamp_series.append(timestamp)
        
        # 实时计算统计量
        if len(self.spread_series) > 10:
            mean = np.mean(self.spread_series)
            std = np.std(self.spread_series)
            z_score = (spread - mean) / std
            return z_score
        return None

我曾经犯过一个低级错误：用固定长度的列表存储价差序列，结果内存爆了。后来改用deque（双端队列），设置最大长度，问题就解决了。

避坑指南：实时序列一定要考虑内存管理。我建议用环形缓冲区或者deque，设置合理的窗口大小。别贪心存太多数据，够用就行。

4.3 价差均值回归特性检验

价差交易能赚钱，前提是价差具有均值回归特性。怎么检验？我一般用两个方法：

ADF检验（单位根检验）：检验序列是否平稳。p值小于0.05，说明存在均值回归特性。
半衰期计算：衡量价差偏离后回归到均值需要多长时间。半衰期越短，交易机会越多。

# Python示例：ADF检验和半衰期计算
from statsmodels.tsa.stattools import adfuller
import numpy as np

def check_mean_reversion(spread_series):
    # ADF检验
    adf_result = adfuller(spread_series)
    p_value = adf_result[1]
    
    # 半衰期计算
    spread_lag = spread_series[:-1]
    spread_diff = np.diff(spread_series)
    # 用线性回归估计半衰期
    slope = np.polyfit(spread_lag, spread_diff, 1)[0]
    half_life = -np.log(2) / slope if slope < 0 else np.inf
    
    return {
        'p_value': p_value,
        'is_stationary': p_value < 0.05,
        'half_life': half_life
    }

我个人的经验是：半衰期在5-20个时间单位之间的价差最适合做高频交易。太短了来不及反应，太长了资金占用成本高。

实战要点：

ADF检验的p值要小于0.05，最好小于0.01
半衰期建议在5-20个采样周期之间
定期重新检验，因为市场结构会变化

4.4 价差波动率计算

波动率决定了你的交易区间。波动率大的时候，价差偏离幅度大，机会多；波动率小的时候，要收紧交易区间。

我常用的波动率计算方法：

# Python示例：实时波动率计算
class VolatilityCalculator:
    def __init__(self, window=20):
        self.window = window
        self.returns = deque(maxlen=window)
    
    def update(self, spread):
        if len(self.returns) > 0:
            ret = (spread - self.returns[-1]) / self.returns[-1]
            self.returns.append(ret)
        else:
            self.returns.append(spread)
        
        if len(self.returns) >= self.window:
            # 计算历史波动率
            hist_vol = np.std(self.returns) * np.sqrt(252 * 6.5 * 3600)  # 年化
            # 计算EWMA波动率（更敏感）
            lambda_param = 0.94
            ewma_vol = self._ewma_volatility()
            return hist_vol, ewma_vol
        return None, None
    
    def _ewma_volatility(self):
        # EWMA波动率计算
        weights = np.array([(1-0.94) * 0.94**i for i in range(len(self.returns))])
        weights = weights / weights.sum()
        mean = np.average(self.returns, weights=weights)
        var = np.average((self.returns - mean)**2, weights=weights)
        return np.sqrt(var) * np.sqrt(252 * 6.5 * 3600)

为什么用EWMA？因为历史波动率反应太慢，市场突然变化时跟不上。EWMA给近期数据更高权重，能更快捕捉到波动率变化。

我的经验：波动率计算窗口不要太大，20-30个采样点就够了。太大反而会钝化，错过交易机会。

知识体系总览

下面这张图把价差计算与实时监控的核心逻辑串起来了：

这张图把价差计算的核心流程串起来了。从价差公式定义开始，到实时序列构建，再到均值回归检验，最后用波动率来指导交易决策。每一步都环环相扣。

总结一下：

价差公式选中间价差更稳定
实时序列用deque管理内存
均值回归检验看ADF和半衰期
波动率用EWMA更敏感

这些基础打牢了，后面讲交易策略和风控才能用得上。别急，一步一步来。

公众号:蓝海资料掘金营，微信deep3321

📚 课程目录