4. 实时价差计算：用Python搞定实时行情与价差序列
做统计套利，核心就是盯住价差。
价差算不准，后面所有策略都是白搭。我见过太多人把精力花在优化进场信号上，结果价差计算那步就埋了雷。今天咱们就把这块地基打牢。
4.1 实时行情获取：选对数据源
实时行情从哪来？我个人习惯用两个渠道：
- WebSocket 推送：适合高频场景，延迟低
- REST API 轮询：适合中低频，实现简单
我建议初学者先从 REST API 入手。为什么？因为调试方便，出问题了你能一眼看到返回的数据长什么样。WebSocket 那玩意儿一旦断连重连，处理起来挺烦的。
下面是一个获取实时行情的例子，我用的是某交易所的公开接口：
import requests
import pandas as pd
import time
from datetime import datetime
def fetch_realtime_price(symbol, exchange_url):
"""
获取实时价格
symbol: 'BTCUSDT' 格式
"""
try:
url = f"{exchange_url}/api/v3/ticker/price?symbol={symbol}"
resp = requests.get(url, timeout=3)
data = resp.json()
return {
'symbol': symbol,
'price': float(data['price']),
'timestamp': datetime.now()
}
except Exception as e:
print(f"获取行情失败: {e}")
return None
# 使用示例
symbol_a = 'BTCUSDT'
symbol_b = 'ETHUSDT'
price_a = fetch_realtime_price(symbol_a, 'https://api.binance.com')
price_b = fetch_realtime_price(symbol_b, 'https://api.binance.com')
小技巧：实际项目中，我会把请求超时设成3秒。太短容易误报，太长会卡住整个循环。
4.2 价差计算：不只是简单的减法
价差公式其实不复杂。但这里有个坑——标准化。
你想想看，如果两个品种价格差了好几个数量级，比如一个几百块，一个几毛钱，直接相减有意义吗？没意义。所以我们要做标准化处理。
常用的价差计算方式有三种：
| 方法 | 公式 | 适用场景 |
|---|---|---|
| 简单价差 | P₁ - P₂ | 同品种、同数量级 |
| 对数价差 | ln(P₁) - ln(P₂) | 价格波动剧烈时 |
| 标准化价差 | (P₁ - P₂) / σ | 跨品种、跨周期 |
我个人最常用的是标准化价差。为什么呢？因为它在不同市场环境下都能保持稳定的统计特性。嗯，这里要注意——标准化用的标准差，得用滚动窗口算，不能拿全量数据算。
def calculate_spread(price_a, price_b, method='standardized', window=20):
"""
计算价差序列
"""
df = pd.DataFrame({
'price_a': price_a,
'price_b': price_b
})
if method == 'simple':
df['spread'] = df['price_a'] - df['price_b']
elif method == 'log':
df['spread'] = np.log(df['price_a']) - np.log(df['price_b'])
elif method == 'standardized':
df['raw_spread'] = df['price_a'] - df['price_b']
df['spread'] = (df['raw_spread'] - df['raw_spread'].rolling(window).mean()) / \
df['raw_spread'].rolling(window).std()
return df['spread']
注意：滚动窗口的大小很关键。窗口太小，价差信号会频繁跳动；窗口太大，反应又太迟钝。我一般从20期开始调，然后根据回测结果微调。
4.3 处理缺失值：别让数据坑了你
实时行情里，缺失值几乎是必然的。网络抖动、交易所维护、数据推送延迟……原因太多了。
我曾经遇到过一个情况：某次实盘中，一个品种的行情停了整整30秒，价差直接跳出一个巨大的异常值，触发了假信号。那次亏了不少钱。从那以后，我对缺失值处理就特别上心。
常见的处理方法：
- 前向填充：用上一个有效值填充，适合短暂中断
- 线性插值：用前后值取平均，适合缓慢变化
- 直接丢弃：如果缺失太多，干脆跳过这一帧
def handle_missing_values(df, method='ffill', max_gap=5):
"""
处理价差序列中的缺失值
max_gap: 允许的最大连续缺失数
"""
df = df.copy()
# 标记缺失值
missing_mask = df['spread'].isna()
if method == 'ffill':
df['spread'] = df['spread'].fillna(method='ffill')
elif method == 'linear':
df['spread'] = df['spread'].interpolate(method='linear')
elif method == 'drop':
# 连续缺失超过max_gap的，直接设为NaN
df['spread'] = df['spread'].where(
~(missing_mask & (missing_mask.groupby(
(~missing_mask).cumsum()).transform('count') > max_gap))
)
return df
我的经验：实盘中我一般用前向填充+最大缺失限制的组合。前向填充保证数据连续性，最大缺失限制防止用太久远的数据来填充。
4.4 实时计算循环：把一切串起来
好了，现在我们把行情获取、价差计算、缺失值处理串成一个实时循环。
def realtime_spread_monitor(symbol_a, symbol_b, exchange_url,
window=20, interval=1):
"""
实时价差监控主循环
interval: 轮询间隔（秒）
"""
price_history_a = []
price_history_b = []
while True:
# 1. 获取实时行情
pa = fetch_realtime_price(symbol_a, exchange_url)
pb = fetch_realtime_price(symbol_b, exchange_url)
if pa and pb:
price_history_a.append(pa['price'])
price_history_b.append(pb['price'])
# 2. 计算价差
if len(price_history_a) >= window:
spread = calculate_spread(
price_history_a[-window:],
price_history_b[-window:],
method='standardized',
window=window
)
# 3. 处理缺失
spread = handle_missing_values(spread)
# 4. 输出当前价差
current_spread = spread.iloc[-1]
print(f"{datetime.now()} | 价差: {current_spread:.4f}")
time.sleep(interval)
核心要点：实时计算的关键不是算法多复杂，而是稳定可靠。宁可慢一点，也别让数据出错。
4.5 本章知识体系
下面这张图，把实时价差计算的整个流程串起来了：
说白了，实时价差计算就是三个步骤的循环：拿数据 → 算价差 → 清数据。每一步都有坑，但踩过一次就记住了。
我个人觉得，做量化交易最怕的不是策略亏钱，而是策略逻辑本身有bug。价差计算这种基础模块，一定要反复验证。你想想看，如果价差都算错了，后面那些机器学习模型、进场信号还有什么意义？