4. 价差计算：实时计算不同交易所之间的价差，识别套利机会

价差计算，说白了就是套利策略的“眼睛”。

你想想看，跨市场套利的核心逻辑就一句话：同一个资产，在不同地方卖不同价。我们要做的，就是实时捕捉这个差价，然后快速出手。

我个人习惯把价差计算拆成三个层次：原始价差、归一化价差、信号价差。咱们一层层剥开。

4.1 原始价差：最朴素的减法

最直接的方式，就是拿交易所A的价格减去交易所B的价格。

比如BTC/USDT，币安报价50000，OKX报价50020，那价差就是20 USDT。

def raw_spread(price_a, price_b):
    return price_a - price_b

但这里有个坑——交易对可能不一样。我在项目中遇到过，有的交易所用BTC/USDT，有的用BTC/USD，还有的用XBT/USD。直接减？那算出来的价差毫无意义。

4.2 归一化价差：消除“刻度”差异

原始价差有个问题：BTC价格50000时，20 USDT的价差算大；但价格5000时，20 USDT的价差就很大了。所以我们需要归一化。

我常用的归一化公式：

def normalized_spread(price_a, price_b):
    return (price_a - price_b) / ((price_a + price_b) / 2) * 10000  # 返回基点(bps)

用基点（bps）表示，1个基点 = 0.01%。这样不管价格高低，价差都能统一比较。

4.3 信号价差：过滤噪音，识别真机会

实时价差波动很大，直接拿原始数据做交易，你会被来回打脸。我建议加一层信号处理。

常用的方法：

• 移动平均平滑：用过去N笔价差的均值代替当前值
• 中位数滤波：剔除极端值，取中位数
• 标准差阈值：价差超过均值±2倍标准差才触发信号

import numpy as np

def signal_spread(spread_history, window=20, threshold=2):
    mean = np.mean(spread_history[-window:])
    std = np.std(spread_history[-window:])
    current = spread_history[-1]
    
    if current > mean + threshold * std:
        return 'long_spread'  # 价差过大，做空价差
    elif current < mean - threshold * std:
        return 'short_spread' # 价差过小，做多价差
    else:
        return 'no_signal'

4.4 多交易所价差矩阵

如果你同时监控3个以上交易所，手动算价差就太累了。我习惯用价差矩阵，一目了然。

交易所	币安	OKX	Bybit	Bitget
币安	0	+12.3 bps	-5.1 bps	+8.7 bps
OKX	-12.3 bps	0	-17.4 bps	-3.6 bps
Bybit	+5.1 bps	+17.4 bps	0	+13.8 bps
Bitget	-8.7 bps	+3.6 bps	-13.8 bps	0

正数表示行交易所价格高于列交易所，负数反之。一眼就能看出哪个交易所最便宜、哪个最贵。

4.5 核心逻辑流程图

下面这张图，是我做套利系统时画的价差计算核心流程。你看一遍就懂了。

4.6 实战中的几个要点

讲完理论，说点实在的。我在实盘里总结了几条经验：

1. 数据对齐是前提：两个交易所的报价时间戳必须对齐。差1秒，价差可能就变了。我一般用WebSocket推送，不用REST轮询。
2. 手续费要算进去：别只看价差。币安maker手续费0.1%，OKX 0.08%，一进一出就是0.18%。价差没超过这个数，做了就是亏。
3. 滑点不可忽视：你看到的价差是盘口价，实际成交可能滑点。我习惯在价差基础上再减5 bps作为安全垫。
4. 网络延迟是隐形杀手：我曾经因为服务器在东京，连美国交易所，价差信号到了，单子已经没了。后来把服务器放在交易所同机房，延迟从80ms降到2ms。