第三节：数据获取实战——使用WebSocket订阅交易所实时行情，解析并存储订单簿快照

做量化交易，数据就是你的弹药。没有实时、准确的数据，再好的策略也是空中楼阁。

我个人习惯，在搭建任何交易系统之前，先把数据管道搞扎实。今天我们就来聊聊，怎么用WebSocket从交易所拿到实时的订单簿数据，然后解析、存起来。

为什么非要用WebSocket？

你可能会问：用REST API轮询不行吗？

嗯，理论上可以。但实际跑起来，你会发现几个问题：

• 延迟高：每次请求都要建立HTTP连接，来回握手，等数据到手，行情早变了。
• 浪费带宽：你每隔几百毫秒拉一次全量数据，大部分数据是重复的。
• 容易被限流：交易所对REST API的调用频率有严格限制，轮询快了直接封你IP。

WebSocket就不一样了。它是一条长连接，交易所一有变化就主动推给你。说白了，就是「订阅-推送」模式，你只管等着收数据就行。

订单簿快照 vs 增量更新

这里有个概念要先搞清楚。交易所推订单簿，通常有两种方式：

类型	说明	特点
全量快照	把当前所有买单卖单一次性推给你	数据完整，但数据量大，推送频率低
增量更新	只推送发生变化的那几档价格	数据量小，实时性高，但需要维护本地状态

我在项目中遇到过一个问题：只订阅增量更新，结果本地状态因为网络抖动和交易所对不上了。后来我学乖了——先拿一次全量快照，再在此基础上应用增量更新。每隔一段时间，再重新拉一次全量快照做校准。

实战：用Python订阅币安订单簿

我们以币安交易所为例。它的WebSocket接口文档写得还算清楚。下面是我常用的代码模板：

import json
import websocket
import pandas as pd
from datetime import datetime

# 币安订单簿WebSocket地址
WS_URL = "wss://stream.binance.com:9443/ws"

# 订阅参数：BTCUSDT的深度数据，每100ms推送一次
SUBSCRIBE_MSG = {
    "method": "SUBSCRIBE",
    "params": [
        "btcusdt@depth20@100ms"  # 20档深度，100ms更新
    ],
    "id": 1
}

def on_message(ws, message):
    """收到消息时的回调函数"""
    data = json.loads(message)
    
    # 判断是快照还是增量
    if 'lastUpdateId' in data and 'bids' in data:
        # 这是全量快照
        process_snapshot(data)
    elif 'U' in data:
        # 这是增量更新
        process_update(data)

def process_snapshot(data):
    """处理全量快照"""
    bids = pd.DataFrame(data['bids'], columns=['price', 'quantity'])
    asks = pd.DataFrame(data['asks'], columns=['price', 'quantity'])
    
    # 转成数值类型
    bids[['price', 'quantity']] = bids[['price', 'quantity']].astype(float)
    asks[['price', 'quantity']] = asks[['price', 'quantity']].astype(float)
    
    # 计算买卖压力
    bid_volume = bids['quantity'].sum()
    ask_volume = asks['quantity'].sum()
    imbalance = (bid_volume - ask_volume) / (bid_volume + ask_volume)
    
    print(f"[快照] 时间: {datetime.now()}, 失衡度: {imbalance:.4f}")
    
    # 存储到本地
    save_to_database(data)

def process_update(data):
    """处理增量更新"""
    # 这里需要维护本地订单簿状态
    # 篇幅原因，略去具体实现
    pass

def on_error(ws, error):
    print(f"WebSocket错误: {error}")

def on_close(ws, close_status_code, close_msg):
    print("WebSocket连接关闭")

def on_open(ws):
    """连接建立后，发送订阅消息"""
    ws.send(json.dumps(SUBSCRIBE_MSG))
    print("订阅请求已发送")

# 启动WebSocket
ws = websocket.WebSocketApp(
    WS_URL,
    on_open=on_open,
    on_message=on_message,
    on_error=on_error,
    on_close=on_close
)

ws.run_forever()

数据存储：别小看这一步

拿到数据之后，存哪里？怎么存？

我个人习惯分两层：

• 热存储：用Redis或内存数据库，存最近几分钟的快照，供策略实时读取。
• 冷存储：用InfluxDB或Parquet文件，存历史数据，供回测和分析用。

为什么这么分？你想想看，策略跑的时候需要毫秒级读取，如果每次都去查磁盘上的CSV文件，延迟根本扛不住。但历史数据又需要长期保存，不能全放内存里。

下面是一个简单的存储结构示例：

# 存储到CSV（简单方案）
import csv
import os

def save_to_database(data):
    """将快照保存到CSV文件"""
    filename = "orderbook_snapshots.csv"
    file_exists = os.path.isfile(filename)
    
    with open(filename, 'a', newline='') as f:
        writer = csv.writer(f)
        if not file_exists:
            writer.writerow(['timestamp', 'bid_price', 'bid_qty', 'ask_price', 'ask_qty'])
        
        timestamp = datetime.now().isoformat()
        # 取前5档买卖单
        for i in range(min(5, len(data['bids']))):
            writer.writerow([
                timestamp,
                data['bids'][i][0],
                data['bids'][i][1],
                data['asks'][i][0],
                data['asks'][i][1]
            ])

知识体系总览

下面这张图，帮你理清整个数据获取流程的逻辑：

几个容易踩的坑

最后，分享几个我踩过的坑，希望能帮你省点时间：

1. 心跳检测不能省：交易所的WebSocket连接如果长时间没数据，会自动断开。记得每隔30秒发一次ping。
2. 数据对齐问题：全量快照和增量更新的时间戳可能不同步。我习惯用交易所的event time做对齐，而不是本地时间。
3. 异常数据处理：偶尔会收到价格为零或数量为负的脏数据。解析时一定要做合法性校验，否则策略会出bug。
4. 重连策略：网络波动导致断连是常态。我一般用指数退避策略重连，第一次等1秒，第二次等2秒，第三次等4秒，最多等60秒。