第3章生产者实战

3、生产者实战：用Python连接交易所WebSocket，抓取实时Tick数据

好，咱们进入实战环节了。

前面两章讲了架构设计和数据模型，说白了都是纸上谈兵。这一章，咱们真刀真枪地干——用Python连上交易所的WebSocket，把实时Tick数据抓下来。

我个人习惯，做流处理项目第一步永远是先搞定数据源。数据进不来，后面再牛的计算引擎也是白搭。你想想看，是不是这个理？

3.1 为什么选WebSocket，而不是REST？

很多新手会问：用REST接口轮询不行吗？

行是行，但效率差太远了。

REST是请求-响应模式，你得不停地问交易所「有新的Tick吗？有新的Tick吗？」。每次问都要建立HTTP连接、传输请求头、等待响应。对于毫秒级的行情数据，这种轮询方式根本跟不上节奏。

WebSocket不一样。它是全双工通信，连接建立后，交易所主动把数据推给你。你什么都不用干，躺着收数据就行。

我记得有一次帮客户优化行情接入，原来用REST轮询，延迟在200ms左右。换成WebSocket后，直接降到5ms以内。嗯，差距就是这么大。

  核心结论： 实时行情必须用WebSocket。REST只适合查历史数据或者做快照。

3.2 准备工作：选交易所和装库

咱们以币安（Binance）为例。为什么选它？

文档清晰，WebSocket接口稳定
数据量大，适合练手
我做过好几个项目都用它，坑基本都踩过了

需要安装的Python库：

pip install websocket-client==1.6.1
pip install json          # 标准库，一般不用装
pip install logging       # 标准库，一般不用装

这里要注意版本。websocket-client 1.x和0.x的API差别很大。我曾经因为版本不对，调试了整整一个下午。所以建议直接指定版本安装。

3.3 核心代码：连接WebSocket并订阅Tick

先看完整代码，我再逐行解释。

import websocket
import json
import logging

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

# 币安WebSocket地址
# 这里订阅BTC/USDT的逐笔交易数据
WS_URL = "wss://stream.binance.com:9443/ws/btcusdt@trade"

def on_message(ws, message):
    """收到消息时的回调函数"""
    try:
        data = json.loads(message)
        tick = {
            'symbol': data['s'],        # 交易对
            'price': float(data['p']),   # 成交价
            'quantity': float(data['q']), # 成交量
            'timestamp': data['T'],       # 成交时间（毫秒）
            'trade_id': data['t']         # 成交ID
        }
        logger.info(f"收到Tick: {tick}")
        # 这里可以对接Kafka或Redis，后面章节会讲
    except Exception as e:
        logger.error(f"解析消息失败: {e}")

def on_error(ws, error):
    """错误回调"""
    logger.error(f"WebSocket错误: {error}")

def on_close(ws, close_status_code, close_msg):
    """连接关闭回调"""
    logger.warning(f"连接关闭: {close_msg}")

def on_open(ws):
    """连接建立回调"""
    logger.info("WebSocket连接已建立")
    # 可以在这里发送订阅消息（如果需要更复杂的订阅）
    # 币安的简单订阅直接在URL里指定了

def start_websocket():
    """启动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(
        reconnect=5,  # 断线后5秒重连
        ping_interval=30,  # 每30秒发一次心跳
        ping_timeout=10    # 心跳超时10秒
    )

if __name__ == "__main__":
    logger.info("启动实时行情采集...")
    start_websocket()

3.4 代码逐段解析

1. 连接地址的玄机

你看这个URL：wss://stream.binance.com:9443/ws/btcusdt@trade

它其实包含了订阅信息。币安支持在URL里直接指定要订阅的流。这里btcusdt@trade就是订阅BTC/USDT的逐笔交易数据。

如果你想订阅多个交易对，可以用组合流：

# 订阅多个交易对
WS_URL = "wss://stream.binance.com:9443/stream?streams=btcusdt@trade/ethusdt@trade"

2. 回调函数的设计

WebSocket是事件驱动的。你定义了四个回调：

on_open：连接建立时触发。我一般在这里打日志，确认连接成功。
on_message：收到数据时触发。这是核心，数据解析都在这里。
on_error：出错时触发。网络波动、数据异常都会走到这里。
on_close：连接关闭时触发。记录关闭原因，方便排查。

3. 数据解析的坑

币安返回的原始数据是这样的：

{
  "e": "trade",
  "E": 1672515782136,
  "s": "BTCUSDT",
  "t": 123456789,
  "p": "16500.00",
  "q": "0.001",
  "T": 1672515782136,
  "m": true
}

注意几个细节：

p和q是字符串，不是数字。必须用float()转换。
T是毫秒级时间戳，Python的datetime处理时需要除以1000。
m表示是否为主动卖出，这个字段在计算买卖方向时很有用。

我曾经踩过的坑： 直接拿字符串当数字用，结果下游计算均价时全错了。后来加了一层类型校验，才把问题解决。记住：交易所返回的数值字段，一定要显式转换类型。

3.5 断线重连：生产环境必须考虑的事

WebSocket连接不是100%稳定的。网络抖动、交易所维护、防火墙超时，都可能导致断线。

我见过最夸张的一次，某交易所一天断线了20多次。如果没有重连机制，数据就全丢了。

run_forever方法里的参数就是干这个的：

参数	作用	建议值
reconnect	断线后多少秒自动重连	5秒（太短会频繁重连，太长会丢数据）
ping_interval	心跳间隔，保持连接活跃	30秒
ping_timeout	心跳超时，超时后主动断开	10秒

小技巧： 如果发现频繁断线，可以加一个指数退避策略。第一次断线等5秒，第二次等10秒，第三次等20秒...这样能避免在交易所短暂故障时疯狂重连。

3.6 知识体系：WebSocket实时数据采集的核心逻辑

下面这张图，是我自己总结的实时数据采集流程。你看一遍，基本就明白整个链路了。

这张图的核心逻辑很简单：交易所推送 → Python接收解析 → 回调处理 → 发往下游。但真正生产环境里，断线重连这个环节往往是最容易出问题的。

3.7 实战中的几个避坑指南

避坑1：不要在主线程里做耗时操作

on_message回调是在WebSocket的IO线程里执行的。如果你在里面写数据库、发HTTP请求，会阻塞消息接收。

我建议的做法：回调里只做轻量解析，然后把数据丢进队列或内存缓冲区，另起线程去消费。

from queue import Queue
import threading

tick_queue = Queue(maxsize=10000)

def on_message(ws, message):
    data = json.loads(message)
    tick_queue.put(data)  # 只入队，不做其他操作

def consumer():
    while True:
        tick = tick_queue.get()
        # 在这里做数据库写入、Kafka发送等耗时操作
        process_tick(tick)

# 启动消费者线程
threading.Thread(target=consumer, daemon=True).start()

避坑2：注意内存泄漏

如果下游处理速度跟不上数据产生速度，队列会越积越多。我曾经遇到过队列积压到几百万条，直接把内存撑爆了。

解决方案：给队列设置最大长度，超出的数据直接丢弃或写入临时文件。

警告： 生产环境一定要监控队列积压情况。我习惯每5秒打印一次队列大小，一旦超过阈值就报警。

避坑3：日志别打太多

Tick数据每秒可能来几百条。如果你每条都打日志，磁盘很快就满了。

我的做法：每100条打一条汇总日志，或者只在错误时打日志。正常数据走文件日志，错误数据走控制台。

3.8 小结

这一章，咱们把生产者这端搞定了。

你学会了：

为什么WebSocket比REST更适合实时行情
如何用Python连接币安WebSocket
数据解析的注意事项和类型转换
断线重连的配置和避坑指南

代码虽然不长，但每一行都是生产环境验证过的。你直接拿去用，基本不会出大问题。

嗯，数据源有了，下一件大事就是怎么把这些数据高效地分发出去。不过那是后面章节的事了。

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

📚 课程目录

3、生产者实战：用Python连接交易所WebSocket，抓取实时Tick数据

3.1 为什么选WebSocket，而不是REST？

3.2 准备工作：选交易所和装库

3.3 核心代码：连接WebSocket并订阅Tick

3.4 代码逐段解析

3.5 断线重连：生产环境必须考虑的事

3.6 知识体系：WebSocket实时数据采集的核心逻辑

3.7 实战中的几个避坑指南

3.8 小结