第4章实时数据流处理

第四章：实时数据流处理

WebSocket连接、数据帧解析、心跳机制与重连策略——这几个词听起来是不是有点唬人？别急，我带你一步步拆解。说白了，实时数据流处理就是你和交易所之间的一条「专线电话」，你得知道怎么拨号、怎么听懂对方说话、怎么保持通话不断。

4.1 WebSocket连接与订阅

先聊聊WebSocket。我个人习惯把它比作「长连接水管」——你打开水龙头，水就一直流，不用每次喝水都去拧一次。这和HTTP轮询完全不一样，HTTP就像你每次喝水都要去井边打一桶，累不累？

在美股Level2行情里，主流数据商（比如Polygon、IEX、NASDAQ）都提供WebSocket接口。连接代码其实不复杂，我贴一段我常用的Python示例：

import websocket
import json

def on_message(ws, message):
    print(f"收到数据: {message}")

def on_error(ws, error):
    print(f"出错了: {error}")

def on_close(ws, close_status_code, close_msg):
    print(f"连接关闭: {close_msg}")

def on_open(ws):
    # 订阅AAPL的Level2数据
    subscribe_msg = {
        "action": "subscribe",
        "params": {
            "symbols": "AAPL,MSFT,GOOGL"
        }
    }
    ws.send(json.dumps(subscribe_msg))
    print("订阅请求已发送")

ws = websocket.WebSocketApp(
    "wss://你的数据商地址/v1/level2",
    on_open=on_open,
    on_message=on_message,
    on_error=on_error,
    on_close=on_close
)

ws.run_forever()

我的小经验：订阅时别一股脑把所有股票都塞进去。我曾经一次订阅了500只股票，结果本地内存直接爆了。建议分批订阅，每次不超过50只。

4.2 数据帧（Frame）解析

数据来了，但它是「加密电报」——你得知道怎么解码。Level2行情的数据帧通常包含这些字段：

字段名	类型	说明
symbol	string	股票代码
bid_price	float	买一价
bid_size	int	买一量
ask_price	float	卖一价
ask_size	int	卖一量
timestamp	long	交易所时间戳（纳秒级）

你想想看，这些字段每秒可能更新几百次。我刚开始做的时候，直接用Python的dict来存，结果GC（垃圾回收）频繁触发，延迟飙升。后来改用namedtuple，性能提升明显。

from collections import namedtuple

Level2Frame = namedtuple('Level2Frame', 
    ['symbol', 'bid_price', 'bid_size', 'ask_price', 'ask_size', 'timestamp'])

def parse_frame(raw_json):
    data = json.loads(raw_json)
    return Level2Frame(
        symbol=data['symbol'],
        bid_price=float(data['bid_px']),
        bid_size=int(data['bid_sz']),
        ask_price=float(data['ask_px']),
        ask_size=int(data['ask_sz']),
        timestamp=int(data['ts'])
    )

  核心要点：解析速度就是金钱。每慢1毫秒，你的策略可能就比别人晚一步。建议用struct或protobuf做二进制解析，比JSON快10倍以上。

4.3 心跳机制

为什么要心跳？说白了就是告诉对方「我还活着」。交易所那边如果长时间没收到你的消息，会认为你掉线了，主动断开连接。

我见过最坑的情况：有个同事没做心跳，结果每天下午2点准时断连，查了三天才发现是交易所的30秒超时机制。嗯，这里要注意——不同数据商的心跳间隔不一样：

NASDAQ: 每15秒发一次ping
NYSE: 每20秒发一次ping
Polygon: 每30秒发一次ping

实现心跳其实就几行代码：

import threading
import time

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

# 在on_open里启动心跳线程
def on_open(ws):
    # ... 订阅逻辑 ...
    t = threading.Thread(target=heartbeat, args=(ws, 15))
    t.daemon = True
    t.start()

避坑指南：我曾经用time.sleep(15)做心跳，结果某次网络抖动，sleep被阻塞了30秒，连接就断了。建议用threading.Timer或者异步方式，更可靠。

4.4 重连策略

网络不可能100%稳定。断连了怎么办？重连呗。但怎么重连是个技术活——你不能像傻瓜一样每秒重试一次，那样会被数据商封IP。

我常用的策略是「指数退避 + 随机抖动」：

import random
import time

def reconnect_with_backoff(ws_url, max_retries=10):
    base_delay = 1  # 初始1秒
    for attempt in range(max_retries):
        try:
            ws = websocket.WebSocketApp(ws_url, ...)
            ws.run_forever()
            break  # 连接成功就退出
        except Exception as e:
            delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
            print(f"第{attempt+1}次重连，等待{delay:.2f}秒...")
            time.sleep(delay)
    else:
        print("重连失败次数过多，请检查网络或数据商状态")

为什么加随机抖动？你想想看，如果所有客户端都在同一时间重连，数据商的服务器会瞬间被打爆。加个0-1秒的随机值，大家错开，你好我也好。

  我的血泪教训：有一次生产环境断连，重连策略没做好，结果10个实例同时重连，把数据商的API打出了429（Too Many Requests）。后来我加了限流和抖动，再也没出过问题。

4.5 知识体系总览

说了这么多，我画张图帮你理清思路：

这张图把整个流程串起来了。你从左边开始，先建立WebSocket连接，然后订阅你感兴趣的股票，接着就是源源不断的数据帧。在数据流过程中，心跳机制一直在后台运行，一旦发现连接断了，重连策略立刻启动。

个人建议：刚开始做的时候，别追求完美。先跑通最简单的连接-订阅-解析流程，再慢慢加上心跳和重连。一口吃不成胖子，对吧？

好了，这一章的内容就到这里。记住：实时数据流处理的核心就三件事——连得上、听得懂、断不了。把这三点做好，你的Level2行情系统就稳了。

📚 课程目录

第四章：实时数据流处理

4.1 WebSocket连接与订阅

4.2 数据帧（Frame）解析

4.3 心跳机制

4.4 重连策略

4.5 知识体系总览