第4章数据采集实战

数据采集实战：使用WebSocket订阅实时Tick流

好，咱们直接进入正题。

上一章我们聊了Tick数据长什么样，这一章就来真的——怎么把数据从交易所拉下来。我个人习惯用WebSocket，原因很简单：快，而且省资源。你想想看，如果用HTTP轮询，每秒请求几十次，服务器不封你才怪。

为什么选WebSocket？

说白了，WebSocket就是一条长连接。建立一次连接，后面数据自己推过来。不像HTTP，每次都要握手、发请求、等响应、再断开。我在项目中遇到过，用HTTP轮询1秒拉10次，带宽占用直接飙到3MB/s。换成WebSocket后，降到200KB/s不到。

核心优势：

全双工通信：服务器可以主动推数据
低延迟：建立连接后，数据延迟在毫秒级
节省带宽：没有HTTP头部冗余

WebSocket连接的生命周期

嗯，这里要注意。WebSocket不是连上就完事了。它有几个关键状态：

连接建立：客户端发起握手，服务器返回101状态码
数据交换：双方可以随时发送消息
心跳维持：每隔一段时间发ping/pong，防止连接断开
连接关闭：任意一方发起关闭帧

我曾经踩过一个坑：连接建立后没做心跳检测，结果半夜网络波动，连接断了都不知道。第二天一看，数据停了6个小时。从那以后，我所有WebSocket客户端都加了心跳机制。

实战：用Python订阅Binance的Tick流

咱们拿币安（Binance）的公开WebSocket接口做例子。为什么选它？因为文档清晰，而且不需要API Key就能订阅实时数据。

import websocket
import json
import threading
import time

# 回调函数：收到消息时触发
def on_message(ws, message):
    data = json.loads(message)
    # 解析Tick数据
    if 'e' in data and data['e'] == '24hrTicker':
        symbol = data['s']
        price = data['c']
        volume = data['v']
        print(f"{symbol} 最新价: {price}, 成交量: {volume}")

# 错误处理
def on_error(ws, error):
    print(f"连接出错: {error}")

# 连接关闭
def on_close(ws, close_status_code, close_msg):
    print("连接已关闭，尝试重连...")
    # 这里可以加自动重连逻辑

# 连接建立
def on_open(ws):
    print("连接成功！")
    # 订阅BTC/USDT的24小时Ticker
    subscribe_msg = {
        "method": "SUBSCRIBE",
        "params": ["btcusdt@ticker"],
        "id": 1
    }
    ws.send(json.dumps(subscribe_msg))

# 启动WebSocket
def start_websocket():
    ws_url = "wss://stream.binance.com:9443/ws"
    ws = websocket.WebSocketApp(ws_url,
                                on_open=on_open,
                                on_message=on_message,
                                on_error=on_error,
                                on_close=on_close)
    # 运行连接（阻塞）
    ws.run_forever()

# 启动
if __name__ == "__main__":
    start_websocket()

小技巧：如果你需要同时订阅多个交易对，可以在subscribe_msg的params列表里加多个参数，比如["btcusdt@ticker", "ethusdt@ticker"]。但注意，不要超过1024个，否则交易所会断开连接。

数据解析：从JSON到结构化数据

WebSocket收到的数据是JSON字符串。我们需要把它解析成Python对象。我个人习惯用dataclass来定义Tick数据结构：

from dataclasses import dataclass
from typing import Optional
import json

@dataclass
class TickData:
    symbol: str          # 交易对
    price: float         # 最新价
    volume: float        # 成交量
    high: float          # 24小时最高价
    low: float           # 24小时最低价
    open: float          # 24小时开盘价
    close: float         # 24小时收盘价
    timestamp: int       # 事件时间戳

def parse_tick(json_str: str) -> Optional[TickData]:
    try:
        data = json.loads(json_str)
        if 'e' not in data or data['e'] != '24hrTicker':
            return None
        
        return TickData(
            symbol=data['s'],
            price=float(data['c']),
            volume=float(data['v']),
            high=float(data['h']),
            low=float(data['l']),
            open=float(data['o']),
            close=float(data['c']),
            timestamp=data['E']
        )
    except Exception as e:
        print(f"解析失败: {e}")
        return None

注意：交易所返回的字段名可能缩写得很厉害。比如Binance的"c"代表close price，"v"代表volume。一定要对照文档确认每个字段的含义。我刚开始做的时候，把"c"当成了change，结果算出来的收益率全是错的。

心跳与重连机制

WebSocket连接不是永久的。网络波动、服务器维护、防火墙超时，都可能导致连接断开。所以，一个健壮的客户端必须包含心跳和重连逻辑。

我曾经在实盘交易中遇到过：凌晨3点，交易所做了一次小版本升级，所有WebSocket连接被强制断开。我的客户端没有重连机制，结果策略在无数据的情况下运行了2小时，亏了6位数。从那以后，我写WebSocket客户端必加以下三样：

心跳检测：每30秒发一次ping，如果60秒内没收到pong，判定连接断开
自动重连：断开后等待5秒，然后重新建立连接
数据连续性检查：重连后检查是否有数据缺失，如果有，从REST API补数据

class ReconnectingWebSocket:
    def __init__(self, url, max_retries=10):
        self.url = url
        self.max_retries = max_retries
        self.retry_count = 0
        self.ws = None
        
    def connect(self):
        self.ws = websocket.WebSocketApp(
            self.url,
            on_open=self.on_open,
            on_message=self.on_message,
            on_error=self.on_error,
            on_close=self.on_close
        )
        # 启动心跳线程
        self.heartbeat_thread = threading.Thread(target=self.heartbeat_loop)
        self.heartbeat_thread.daemon = True
        self.heartbeat_thread.start()
        
        # 阻塞运行
        self.ws.run_forever()
    
    def heartbeat_loop(self):
        while True:
            time.sleep(30)
            if self.ws and self.ws.sock:
                try:
                    self.ws.send(json.dumps({"method": "PING"}))
                except:
                    print("心跳发送失败")
                    break
    
    def on_close(self, ws, code, msg):
        print(f"连接关闭: {code} - {msg}")
        if self.retry_count < self.max_retries:
            wait_time = min(5 * (2 ** self.retry_count), 60)
            print(f"等待 {wait_time} 秒后重连...")
            time.sleep(wait_time)
            self.retry_count += 1
            self.connect()

数据存储：别让数据丢了

WebSocket收到的数据是流式的，如果不存下来，一旦程序重启，数据就没了。我建议至少做两级存储：

存储层级	用途	推荐方案
内存缓冲区	实时计算用，延迟最低	deque（固定长度）或RingBuffer
磁盘持久化	历史回测用，防止丢失	CSV文件或Parquet
数据库	长期存储，方便查询	InfluxDB（时序数据库）或ClickHouse

我个人习惯的做法是：内存里用deque存最近1000条Tick，同时每收到一条就追加写入CSV文件。这样既保证了实时计算的速度，又不会丢数据。

本章知识体系

下面这张图总结了WebSocket订阅Tick流的完整流程：

避坑指南

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

不要在生产环境用单线程：WebSocket的on_message回调是单线程的，如果处理逻辑太慢，会阻塞后续消息。建议用异步队列，把数据丢到另一个线程处理。
注意数据乱序：WebSocket不保证消息顺序。如果对顺序敏感，每条消息带时间戳，自己排序。
小心交易所的限流：有些交易所对WebSocket订阅数量有限制。比如Binance单个连接最多订阅1024个stream。超了会被断开。
本地时间与服务器时间：Tick数据的时间戳是交易所服务器时间。不要用本地时间做计算，否则回测时会出问题。

我的建议：刚开始做的时候，先订阅一个交易对，把整个流程跑通。确认数据能正常接收、解析、存储后，再慢慢增加订阅数量。一口吃不成胖子，WebSocket调试起来比HTTP麻烦得多。

好了，这一章的内容就到这里。代码可以直接复制去跑，但记得把交易对换成你需要的。下一章我们会聊怎么用这些Tick数据做实时统计计算——嗯，那才是真正有意思的部分。

公众号:蓝海数据掘金营，微信deep3321

📚 课程目录

数据采集实战：使用WebSocket订阅实时Tick流

为什么选WebSocket？

WebSocket连接的生命周期

实战：用Python订阅Binance的Tick流

数据解析：从JSON到结构化数据

心跳与重连机制

数据存储：别让数据丢了

本章知识体系

避坑指南