行情接口：WebSocket实时行情接入、REST API快照获取、数据解析与缓存

做量化交易，行情接口就是你的眼睛。

眼睛不好使，再牛的策略也是白搭。我个人习惯把行情接入分成三个层次：实时流、快照、本地缓存。今天咱们就把这三层彻底讲透。

一、为什么需要两套接口？

你想想看，WebSocket和REST API，一个像水龙头，一个像照相机。

WebSocket是长连接，数据主动推过来。适合做盘口实时更新，毫秒级延迟。但有个问题——万一断线重连，中间的数据就丢了。

REST API是一次性请求，拿到的是当前时刻的全量快照。适合做初始化、补数据、校验。但频繁请求会被交易所封IP。

所以实战中，我都是两者结合：

• 启动时，先拉一次REST快照，拿到基准数据
• 然后开启WebSocket，接收增量更新
• 本地维护一个订单簿，用增量去更新快照

二、WebSocket实时行情接入

先说说WebSocket。这东西其实不复杂，就是一个全双工的TCP连接。

我早期踩过一个坑——直接用浏览器那套ws库，结果在Python异步环境里各种报错。后来换了websockets库，配合asyncio，才算稳定下来。

2.1 连接与心跳

交易所一般要求你定时发心跳包。比如每5秒发一个ping，服务器回一个pong。如果连续几次没收到，就该重连了。

import asyncio
import websockets
import json

async def subscribe_depth(symbol):
    uri = "wss://api.example.com/ws"
    async with websockets.connect(uri) as ws:
        # 订阅深度数据
        sub_msg = {
            "op": "subscribe",
            "args": [f"depth.{symbol}"]
        }
        await ws.send(json.dumps(sub_msg))
        
        while True:
            try:
                data = await asyncio.wait_for(ws.recv(), timeout=10)
                # 解析数据...
                process_depth(data)
            except asyncio.TimeoutError:
                # 超时了，发个ping
                await ws.send("ping")

2.2 数据格式与解析

交易所返回的深度数据，通常是这样的：

{
    "e": "depthUpdate",
    "s": "BTCUSDT",
    "b": [["50000.00", "1.5"], ["49999.00", "2.0"]],
    "a": [["50001.00", "0.8"], ["50002.00", "1.2"]]
}

b是买单（bids），a是卖单（asks）。每个数组里，第一个是价格，第二个是数量。

注意：这里给的是增量数据，不是全量。你需要用这些增量去更新本地的订单簿。

三、REST API快照获取

WebSocket启动之前，必须先拿一次快照。不然你都不知道当前盘口长什么样。

3.1 请求快照

import requests

def get_snapshot(symbol):
    url = f"https://api.example.com/api/v1/depth?symbol={symbol}&limit=100"
    resp = requests.get(url)
    data = resp.json()
    return data

limit参数一般可以设20、50、100。我建议设100，够用又不至于数据量太大。

3.2 快照与增量的合并

拿到快照后，怎么和WebSocket的增量合并？

说白了，就是维护一个本地的订单簿字典：

class OrderBook:
    def __init__(self, snapshot):
        self.bids = {price: qty for price, qty in snapshot['b']}
        self.asks = {price: qty for price, qty in snapshot['a']}
    
    def update(self, delta):
        # 更新买单
        for price, qty in delta['b']:
            if float(qty) == 0:
                self.bids.pop(price, None)
            else:
                self.bids[price] = float(qty)
        # 更新卖单
        for price, qty in delta['a']:
            if float(qty) == 0:
                self.asks.pop(price, None)
            else:
                self.asks[price] = float(qty)

注意：如果某个价格的数量变成0，说明这一档已经撤单了，要删掉。

四、数据解析与缓存

数据拿到手，不能直接用。得解析成你策略能吃的格式。

4.1 解析成标准结构

我个人习惯把盘口数据转成这样的结构：

{
    "symbol": "BTCUSDT",
    "timestamp": 1700000000000,
    "bids": [
        {"price": 50000.0, "qty": 1.5},
        {"price": 49999.0, "qty": 2.0}
    ],
    "asks": [
        {"price": 50001.0, "qty": 0.8},
        {"price": 50002.0, "qty": 1.2}
    ]
}

这样策略模块直接拿来用，不用关心底层是WebSocket还是REST。

4.2 本地缓存策略

缓存不是随便存一下就行。要考虑几个问题：

• 内存占用：一个订单簿可能几千档，几十个币种同时监控，内存压力不小
• 更新频率：毫秒级更新，不能每次都写磁盘
• 断线恢复：程序重启后，能不能从缓存恢复？

我的做法是：

• 内存里用字典维护最新订单簿
• 每5秒把快照序列化后存到Redis
• 程序启动时，先从Redis恢复，再拉REST快照做校验

五、整体架构图

下面这张图，是我做行情系统时的标准架构。你看一眼就明白了。

你看，整个流程很清晰：交易所同时提供WebSocket和REST两路数据，经过解析模块处理后，一路进缓存做持久化，一路进本地订单簿供策略实时使用。

六、实战中的几个坑

最后分享几个我踩过的坑，你遇到了能少走弯路：

1. WebSocket断线重连——一定要加指数退避。第一次等1秒，第二次2秒，第三次4秒...不然服务器会认为你在攻击它。
2. 数据时间戳——交易所给的时间戳是服务器时间，和你本地时间可能有偏差。我习惯用交易所的时间戳，不用本地时间。
3. 内存泄漏——订单簿如果只增不减，内存迟早爆掉。记得定期清理过期数据，比如只保留最近100档。
4. 日志记录——每笔行情都打日志？别傻了，一秒几千条，磁盘直接写爆。我一般只记录异常情况和关键快照。

好了，行情接口这块就聊到这儿。记住一句话：WebSocket管快，REST管准，缓存管稳。三者配合，你的交易系统才能跑得又稳又快。

---