第三章 实时数据流处理：WebSocket连接、数据清洗与对齐、Tick级数据存储

好，咱们进入实战环节最核心的一块——实时数据流处理。

说实话，很多做量化的人，策略写得花里胡哨，但一到实盘就崩。为什么？数据流没处理好。你想想看，行情数据就像流水，你接不住、洗不干净、存不对，后面所有分析都是空中楼阁。

这一章，我带你搞定三件事：连上WebSocket、清洗对齐数据、存好Tick级数据。这三板斧练好了，你的策略才算真正有了“实时”的底气。

3.1 WebSocket连接：别用轮询，太Low了

我记得刚入行那会儿，还有人用HTTP轮询拿行情。每秒请求一次，服务器没崩，自己先崩了。现在谁还这么干？WebSocket才是正经路子。

WebSocket说白了就是一条长连接。客户端和服务器握个手，之后数据就像自来水一样，源源不断流过来。延迟低，效率高。

我习惯用Python的websockets库，轻量又好用。给你看个基础框架：

import asyncio
import websockets
import json

async def connect_market_data(uri):
    """连接行情WebSocket"""
    async for websocket in websockets.connect(uri, ping_interval=20):
        try:
            # 订阅合约
            subscribe_msg = {
                "op": "subscribe",
                "args": ["ticker.BTCUSDT"]
            }
            await websocket.send(json.dumps(subscribe_msg))
            
            # 持续接收数据
            async for message in websocket:
                yield json.loads(message)
                
        except websockets.ConnectionClosed:
            print("连接断了，准备重连...")
            continue

这里有个坑——心跳机制。我曾经遇到过一次，交易所的WebSocket服务每隔30秒不发心跳就断开。我这边没做保活，结果半夜策略断线了，第二天一看亏了不少。后来我学乖了，ping_interval设成20秒，比交易所的阈值短一点，稳得很。

3.2 数据清洗与对齐：脏数据会要了策略的命

数据进来了，但你不能直接用。为什么？因为交易所发来的数据，有时候会乱序、重复、甚至缺失。我见过最离谱的一次，某交易所的Tick数据里居然混了一条昨天的数据。

所以，清洗是必须的。我一般做三步：

1. 去重：同一笔成交，可能因为网络重传收到两次。用(交易对, 时间戳, 成交ID)做唯一键，重复的直接扔掉。
2. 排序：WebSocket数据包可能乱序到达。我习惯维护一个长度为100的滑动窗口，窗口内按时间戳排序。
3. 补缺失：如果发现某笔Tick的时间戳和前一笔间隔超过阈值（比如500ms），说明中间可能有数据丢了。这时候我会用前一笔的价格填充，或者直接标记为缺失。

给你看个清洗函数的例子：

def clean_tick(tick: dict, last_tick: dict) -> dict:
    """清洗单笔Tick数据"""
    # 去重检查
    if tick['trade_id'] == last_tick.get('trade_id'):
        return None  # 重复数据，丢弃
    
    # 时间戳检查
    if tick['timestamp'] < last_tick.get('timestamp', 0):
        return None  # 乱序数据，丢弃
    
    # 价格合理性检查
    if tick['price'] <= 0 or tick['price'] > 1e6:
        return None  # 异常价格，丢弃
    
    return tick

数据对齐呢？说白了就是让不同数据源的时间基准统一。比如你同时接了币安和OKX的行情，它们的服务器时间可能有几十毫秒的偏差。我一般用NTP同步本地时间，然后以本地时间为准，把所有Tick的时间戳都转成本地时间。

3.3 Tick级数据存储：别小看这步，它决定了你的回测质量

Tick数据有多重要？我这么说吧，没有Tick数据，你的策略回测就是隔靴搔痒。日线级别的回测，根本看不出高频交易中的滑点和冲击成本。

但Tick数据存储有个难题——数据量太大了。一个活跃的合约，一天能产生几十万笔Tick。存一年就是几亿条。怎么存？

我推荐用列式存储。比如Parquet或者InfluxDB。别用MySQL，别用CSV。为什么？因为Tick数据是时序数据，列式存储的压缩比高，查询速度快。

给你看个存储方案对比：

存储方式	写入速度	查询速度	压缩比	推荐场景
CSV	慢	慢	低	小规模测试
MySQL	中	中	中	不推荐用于Tick
Parquet	快	快	高	离线分析
InfluxDB	极快	极快	高	实时查询

我个人习惯用Parquet + ClickHouse的组合。白天实时写入ClickHouse，晚上批量转存Parquet做归档。这样既保证了实时查询的速度，又降低了存储成本。

给你看个存储代码片段：

import pandas as pd
from datetime import datetime

def save_tick_to_parquet(tick_data: list, symbol: str):
    """将Tick数据批量写入Parquet文件"""
    df = pd.DataFrame(tick_data)
    # 按日期分区存储
    date_str = datetime.now().strftime("%Y%m%d")
    file_path = f"data/{symbol}/tick_{date_str}.parquet"
    
    # 追加写入
    df.to_parquet(file_path, engine='pyarrow', compression='snappy', append=True)
    print(f"已写入 {len(df)} 条Tick数据到 {file_path}")

3.4 整体流程：一张图看懂

说了这么多，咱们把整个流程串起来。我画了张图，你看一眼就明白了：

这张图把整个流程串起来了。你从左边开始，连上WebSocket，拿到原始数据。然后清洗、对齐，最后存起来。存完之后，数据有两个去向：一是喂给实时策略引擎做交易决策，二是存到离线库供回测分析。

嗯，这里要注意——实时和离线用的是同一套数据。这样你的回测结果才能真实反映实盘表现。我见过有人回测用一套数据，实盘用另一套，结果回测赚得盆满钵满，实盘亏得底朝天。说白了，数据不一致，一切都是白搭。

好了，这一章的内容就到这儿。记住，数据是量化交易的血液，别让它堵了。