4. 行情数据接入：交易所API对接与数据标准化

行情数据，说白了就是做市商的「眼睛」。没有它，你就是在黑箱里做交易。

我个人习惯把行情接入分成三层：链路层（怎么连）、解析层（怎么读）、聚合层（怎么用）。今天咱们就一层层拆开来讲。

4.1 交易所API对接：REST vs WebSocket

大部分交易所都提供两种接口：REST和WebSocket。你可能会问，为什么要有两套？

REST接口，适合做「一次性查询」。比如开盘前拉一下历史K线，或者查账户余额。它的优点是简单，缺点嘛——太慢了。一次请求几十毫秒，高频场景根本扛不住。

WebSocket，才是做市商的主力。它建立一条长连接，交易所主动往你这边推数据。延迟能压到1-5毫秒。我在项目中遇到过，用REST去轮询深度数据，结果还没轮完一轮，盘口已经变了三次。嗯，从那以后，实时行情我只用WebSocket。

下面是一个简单的WebSocket连接示例（Python）：

import websocket
import json

def on_message(ws, message):
    data = json.loads(message)
    # 这里处理行情数据
    process_tick(data)

def on_error(ws, error):
    print(f"连接异常: {error}")
    # 自动重连逻辑
    reconnect()

def on_close(ws, close_status_code, close_msg):
    print("连接关闭，准备重连...")
    reconnect()

def on_open(ws):
    print("连接成功")
    # 订阅BTC/USDT深度
    sub_msg = {
        "op": "subscribe",
        "args": ["depth.BTC-USDT"]
    }
    ws.send(json.dumps(sub_msg))

websocket.enableTrace(False)
ws = websocket.WebSocketApp(
    "wss://api.exchange.com/ws",
    on_open=on_open,
    on_message=on_message,
    on_error=on_error,
    on_close=on_close
)
ws.run_forever()

4.2 行情数据标准化处理

不同交易所，数据格式五花八门。有的用数组，有的用字典；有的时间戳是秒级，有的是毫秒级。你想想看，如果每个交易所都单独写一套处理逻辑，那代码得多乱？

我的做法是：统一内部数据模型。不管外面来的是什么，进到系统里，必须转成标准格式。

举个例子，深度数据的标准化：

class StandardDepth:
    def __init__(self):
        self.exchange = ""      # 交易所名称
        self.symbol = ""        # 交易对
        self.bids = []          # 买盘 [[price, qty], ...]
        self.asks = []          # 卖盘 [[price, qty], ...]
        self.timestamp = 0      # 毫秒时间戳
        self.seq_id = 0         # 序列号，用于去重

def normalize_depth(raw_data, exchange_name):
    """将不同交易所的深度数据转为标准格式"""
    std = StandardDepth()
    std.exchange = exchange_name
    
    if exchange_name == "Binance":
        std.symbol = raw_data['s']
        std.bids = [[float(p), float(q)] for p, q in raw_data['b']]
        std.asks = [[float(p), float(q)] for p, q in raw_data['a']]
        std.timestamp = raw_data['E']
        std.seq_id = raw_data['u']
    elif exchange_name == "OKX":
        # OKX的格式略有不同
        data = raw_data['data'][0]
        std.symbol = data['instId']
        std.bids = [[float(item[0]), float(item[1])] for item in data['bids']]
        std.asks = [[float(item[0]), float(item[1])] for item in data['asks']]
        std.timestamp = data['ts']
        std.seq_id = int(data['seqId'])
    
    return std

4.3 多源数据聚合与去重

做市商通常要接入3-5家交易所。为什么？因为单一交易所的深度不够，容易被操纵。多源聚合后，你能看到更真实的盘口。

但多源也带来了新问题：重复数据。同一笔成交，可能从两个交易所同时推送过来。如果不做去重，你的统计就会翻倍。

去重的核心是序列号。每个交易所的每一条行情，都有一个唯一的序列号。我们只需要记住每个交易所、每个交易对「最后处理过的序列号」，新来的数据如果序列号小于等于它，直接丢弃。

class Deduplicator:
    def __init__(self):
        # key: "exchange:symbol", value: last_seq_id
        self.last_seq_map = {}
    
    def is_duplicate(self, exchange, symbol, seq_id):
        key = f"{exchange}:{symbol}"
        last_seq = self.last_seq_map.get(key, 0)
        if seq_id <= last_seq:
            return True
        self.last_seq_map[key] = seq_id
        return False

# 使用示例
dedup = Deduplicator()
for raw_data in stream:
    std_data = normalize_depth(raw_data, "Binance")
    if dedup.is_duplicate(std_data.exchange, std_data.symbol, std_data.seq_id):
        continue  # 重复数据，跳过
    process_depth(std_data)

聚合的逻辑更直接：把多个交易所的深度，按价格合并到一起。比如买盘，把所有交易所的买单按价格从高到低排序，取前N档。这样你看到的深度，就是整个市场的「合成深度」。

4.4 整体架构图

下面这张图，是我在实际项目中用过的行情接入架构。你可以看到数据从交易所到最终使用的完整链路：

这张图里，从上到下依次是：交易所源、连接管理、标准化、去重排序、最终输出。每一层都有明确的职责，互不干扰。我在项目里就是这么搭的，维护起来特别省心。

4.5 性能优化要点

行情接入对性能要求极高。我总结了几条经验：

优化项	具体做法	效果
内存复用	预分配对象池，避免频繁GC	延迟降低30%
零拷贝	直接操作字节数组，减少数据复制	吞吐量提升2倍
批量处理	攒够100条再一起处理，减少锁竞争	CPU使用率下降40%
无锁队列	使用Disruptor或类似机制	延迟稳定在微秒级

好了，行情接入这块就聊到这儿。核心就三件事：连得上、读得懂、合得拢。把这三点做好，你的做市商系统就成功了一半。

---

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