第3章实时数据采集

实时数据采集：WebSocket连接券商接口、行情数据解析与清洗、断线重连与心跳机制

做量化交易，最怕什么？

我个人觉得，不是策略亏钱，而是行情断了你都不知道。你想想看，盘口资金流向这种高频数据，差一秒可能就是天壤之别。今天我们就来聊聊实时数据采集这块硬骨头——怎么连、怎么接、怎么保证不掉链子。

一、WebSocket连接券商接口

传统的HTTP轮询，说白了就是「你问一句，它答一句」。这在低频场景下没问题，但做盘口资金流？不行。延迟太高，带宽浪费也大。

WebSocket就不一样了。它是一条长连接，建立后双方随时可以推数据。我习惯把它想象成一根水管——接上了，水就一直流，不用反复拧水龙头。

  核心要点： 券商提供的WebSocket接口，通常需要先通过REST API获取token，再用token建立WS连接。

举个例子，某券商的行情网关地址一般是 wss://push.tradex.com/ws。连接前，你得先登录拿到一个access_token。嗯，这里要注意——token有过期时间，我建议在代码里加个定时器，提前30秒刷新。

import websocket
import json

def on_message(ws, message):
    # 收到行情数据
    data = json.loads(message)
    process_tick(data)

def on_error(ws, error):
    print(f"连接异常: {error}")

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

def on_open(ws):
    # 订阅盘口数据
    sub_msg = {
        "action": "subscribe",
        "channel": "orderbook",
        "symbols": ["600519", "000858"]
    }
    ws.send(json.dumps(sub_msg))

ws = websocket.WebSocketApp(
    "wss://push.tradex.com/ws?token=YOUR_TOKEN",
    on_message=on_message,
    on_error=on_error,
    on_close=on_close
)
ws.on_open = on_open
ws.run_forever()

这段代码看起来简单，但我在项目中遇到过一个问题：run_forever() 默认不会自动重连。一旦网络抖动，连接断了，程序就傻在那了。所以，后面我们要聊心跳和重连机制。

二、行情数据解析与清洗

券商推过来的原始数据，长什么样？

我拿一个真实的盘口快照举例：

{
  "symbol": "600519",
  "timestamp": 1693123456789,
  "bids": [
    [189.50, 1200],
    [189.48, 800],
    [189.45, 1500]
  ],
  "asks": [
    [189.52, 900],
    [189.55, 1100],
    [189.58, 600]
  ],
  "last_price": 189.51,
  "volume": 23500
}

看着挺规整对吧？但实际跑起来，你会发现一堆坑：

时间戳格式不统一：有的券商用毫秒，有的用微秒，甚至还有字符串格式的。我建议统一转成Unix毫秒时间戳，方便后续对齐。
价格精度问题：有些股票价格是两位小数，有些是三位。比如科创板股票，价格可以精确到0.001元。解析时一定要用Decimal，别用float——float的精度丢失会让你算资金流时差之毫厘谬以千里。
空值或异常值：比如某档挂单量为0，但券商可能传null或者干脆不传这个字段。清洗时要做默认值填充。

我的小技巧： 写一个统一的清洗函数，把原始数据转成标准化的DataFrame结构。这样后面算资金流向、做可视化，都不用再操心数据格式问题。

def clean_orderbook(raw):
    """清洗盘口数据"""
    cleaned = {
        "symbol": raw["symbol"],
        "timestamp": int(raw["timestamp"]) // 1000,  # 统一转秒
        "bid_price_1": raw["bids"][0][0] if raw["bids"] else 0,
        "bid_volume_1": raw["bids"][0][1] if raw["bids"] else 0,
        "ask_price_1": raw["asks"][0][0] if raw["asks"] else 0,
        "ask_volume_1": raw["asks"][0][1] if raw["asks"] else 0,
        "last_price": raw.get("last_price", 0),
        "volume": raw.get("volume", 0)
    }
    return cleaned

三、断线重连与心跳机制

做实时行情，断线是常态，不是意外。

我曾经在一次实盘交易中，因为网络波动断了30秒，结果错过了关键的资金流入信号。那次之后，我彻底重构了重连逻辑。

3.1 心跳机制

说白了，就是客户端和服务器互相确认「我还活着」。一般有两种方式：

Ping/Pong帧：WebSocket协议自带的心跳。客户端发Ping，服务器回Pong。如果一段时间没收到Pong，就认为连接断了。
业务心跳：有些券商要求客户端定期发送一个自定义的心跳消息，比如 {"action": "ping"}，服务器回复 {"action": "pong"}。

我个人习惯两种都用。协议层的心跳保底，业务层的心跳做双重确认。

import threading
import time

class HeartbeatThread(threading.Thread):
    def __init__(self, ws, interval=30):
        super().__init__()
        self.ws = ws
        self.interval = interval
        self.running = True

    def run(self):
        while self.running:
            time.sleep(self.interval)
            try:
                self.ws.send(json.dumps({"action": "ping"}))
            except:
                print("心跳发送失败，连接可能已断开")
                self.running = False

3.2 断线重连策略

重连不是简单地把代码再跑一遍。我总结了一个「三步走」策略：

指数退避：第一次重连等1秒，第二次等2秒，第三次等4秒……最多等60秒。防止服务器压力过大。
重新认证：重连后要重新获取token，因为旧的token可能已经过期。
恢复订阅：重连成功后，重新发送订阅消息，把之前订阅的股票代码再订阅一遍。

注意： 重连期间丢失的行情数据是无法补回的。所以，我建议在本地维护一个「数据缺口标记」，等重连后通过REST API拉取缺失的分钟级K线做插值。虽然不精确，但比没有强。

def reconnect(ws, max_retries=10):
    retry_count = 0
    while retry_count < max_retries:
        try:
            # 重新获取token
            token = get_new_token()
            # 重新连接
            ws = websocket.WebSocketApp(
                f"wss://push.tradex.com/ws?token={token}",
                on_message=on_message,
                on_error=on_error,
                on_close=on_close
            )
            ws.on_open = on_open
            ws.run_forever()
            break
        except Exception as e:
            retry_count += 1
            wait_time = min(2 ** retry_count, 60)
            print(f"第{retry_count}次重连失败，等待{wait_time}秒...")
            time.sleep(wait_time)

四、整体架构图

说了这么多，我们来看一张完整的流程图。它把上面讲的三个环节串在了一起：

这张图把整个流程分成了四层：连接层、管理层、解析层、输出层。每一层都有明确的职责。我在实际项目中，还会在每一层加日志记录——哪个股票、什么时间、数据是否正常。出了问题，一眼就能定位。

五、避坑指南

我曾经踩过的坑：

用单线程跑WebSocket，结果心跳和数据处理互相阻塞。后来改成双线程——一个收数据，一个发心跳。
没有处理「重复订阅」。重连后如果忘了取消旧订阅，会收到两倍的数据量，导致资金流计算翻倍。我建议每次重连前先发一个unsubscribe_all。
忽略内存泄漏。行情数据每秒几十条，如果不及时清理，内存会爆。我习惯用环形缓冲区，只保留最近10万条tick。

我的建议： 开发阶段先用模拟数据测试重连逻辑。你可以写一个脚本，每30秒断开一次WebSocket服务，看看你的客户端能不能自动恢复。等模拟测试跑通了，再上实盘。

好了，实时数据采集这块就聊到这儿。记住一句话：行情数据是量化交易的血液，而WebSocket就是输送血液的血管。血管不能断，断了要能马上接上。

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

📚 课程目录

实时数据采集：WebSocket连接券商接口、行情数据解析与清洗、断线重连与心跳机制

一、WebSocket连接券商接口

二、行情数据解析与清洗

三、断线重连与心跳机制

3.1 心跳机制

3.2 断线重连策略

四、整体架构图

五、避坑指南