数据采集与清洗：交易所API对接、WebSocket实时数据流、数据去重与对齐、缺失值处理、Tick数据与K线数据

做市商系统最核心的命脉是什么？

我个人觉得，不是策略，不是算法，而是数据。你想想看，如果喂给系统的数据本身就是错的、乱的、延迟的，那再牛逼的策略也是白搭。这一章，我们就来聊聊数据采集与清洗的那些坑和实战经验。

交易所API对接：从REST到WebSocket的进化

早期我做量化的时候，大部分交易所只提供REST API。说白了就是你去请求一次，它给你一次数据。这种方式做做回测还行，但做市商是高频交易，你每秒钟要请求几十次，不仅延迟高，还容易被交易所限流甚至封IP。

我记得有一次，我负责对接某二线交易所，REST请求频率稍微高了一点，对方直接返回了429状态码。嗯，那天的交易基本就废了。

所以现在的主流做法是：REST API只用来做查询账户、下单等低频操作，实时行情必须走WebSocket。

下面是一个简单的WebSocket行情订阅代码示例，我用的是Python的websockets库：

import asyncio
import websockets
import json

async def subscribe_ticker():
    uri = "wss://api.example.com/ws"
    async with websockets.connect(uri) as websocket:
        # 订阅BTC/USDT的ticker
        subscribe_msg = {
            "method": "subscribe",
            "params": ["ticker.BTCUSDT"],
            "id": 1
        }
        await websocket.send(json.dumps(subscribe_msg))
        
        while True:
            response = await websocket.recv()
            data = json.loads(response)
            # 这里处理实时行情数据
            print(f"最新价格: {data['data']['last']}")
            
asyncio.run(subscribe_ticker())

这里有个细节要注意：心跳机制。很多交易所的WebSocket连接如果长时间没有消息，会自动断开。我建议每隔30秒发送一次ping消息，保持连接活跃。

WebSocket实时数据流：断线重连与数据缓冲

WebSocket连接不是100%稳定的。网络抖动、交易所服务器重启、甚至你本地路由器抽风，都可能导致断线。我曾经在凌晨三点被电话叫醒，就是因为WebSocket断了没重连，系统用了半小时前的价格做报价，直接亏了六位数。

从那以后，我设计了一套断线重连+数据缓冲机制：

1. 指数退避重连：第一次断线等1秒重连，第二次等2秒，第三次等4秒...最多等60秒。避免频繁重连把交易所服务器打挂。
2. 本地数据缓冲：断线期间，用本地缓存的最新价格做报价。但要注意，如果断线超过5秒，必须暂停做市，因为价格可能已经变了。
3. 重连后数据对齐：重连成功后，先拉取最近100笔成交数据，和本地数据做对比，确保没有遗漏。

数据去重与对齐：别让重复数据毁了你的策略

数据去重，听起来很简单对吧？但实际做起来坑很多。

为什么会重复？常见原因有两个：一是WebSocket推送了重复的消息（交易所的bug），二是你同时订阅了多个数据源（比如同时连了主站和备用站）。

我习惯用(交易所ID + 交易对 + 时间戳 + 成交ID)作为唯一键来做去重。注意，单纯用时间戳不行，因为同一毫秒可能有多次成交。

def deduplicate_tick(tick, seen_set):
    # 生成唯一键
    key = f"{tick['exchange']}_{tick['symbol']}_{tick['timestamp']}_{tick['trade_id']}"
    if key in seen_set:
        return False  # 重复数据，丢弃
    seen_set.add(key)
    return True

数据对齐就更讲究了。做市商系统通常需要同时处理多个交易所的数据，但不同交易所的时间戳精度不一样。有的精确到毫秒，有的精确到微秒，还有的干脆只给秒级时间戳。

我的做法是：统一将时间戳对齐到毫秒级。微秒级的截断，秒级的补零。然后以本地NTP服务器的时间为基准，计算每个交易所数据的延迟，打上延迟标签。

交易所	原始时间戳精度	对齐后格式	典型延迟
交易所A	微秒	毫秒（截断）	5-10ms
交易所B	毫秒	毫秒（不变）	10-20ms
交易所C	秒	毫秒（补000）	50-100ms

缺失值处理：宁可空着，不要乱填

缺失值在数据清洗里是个老生常谈的问题。但在做市商场景下，处理方式完全不同。

比如，某交易所的深度数据突然缺失了卖一档。你能用前一秒的卖一档来填充吗？绝对不能！因为价格可能已经变了，你用旧数据做报价，就是在送钱给市场。

我的原则是：对于实时交易数据，缺失就是缺失，不要填充。直接标记为无效数据，让策略层自己决定怎么处理。

但对于历史数据回测，可以适当填充。我常用的方法：

• 线性插值：适用于价格缓慢变化的情况，比如5分钟K线的中间缺失值。
• 前向填充：适用于成交量等累积型数据，用上一个有效值填充。
• 直接丢弃：如果缺失比例超过5%，干脆把这段数据扔掉，避免污染回测结果。

Tick数据与K线数据：两种视角，两种用途

Tick数据是逐笔成交数据，每一笔交易都记录。K线数据是聚合数据，把一段时间内的tick数据压缩成一根K线。

做市商系统里，两者各有用途：

• Tick数据：用于高频策略、订单簿重建、微观结构分析。数据量大，每秒可能几千条。
• K线数据：用于中低频策略、趋势判断、风险监控。数据量小，便于存储和分析。

我一般会同时维护两套数据流。Tick数据实时处理，K线数据每分钟或每小时聚合一次，存入数据库用于后续分析。

下面是一个从Tick数据生成1分钟K线的示例：

def ticks_to_kline(ticks, interval=60):
    if not ticks:
        return None
    
    # 按时间分组
    grouped = {}
    for tick in ticks:
        minute_key = tick['timestamp'] // (interval * 1000)
        if minute_key not in grouped:
            grouped[minute_key] = []
        grouped[minute_key].append(tick)
    
    klines = []
    for key, group in grouped.items():
        prices = [t['price'] for t in group]
        volumes = [t['volume'] for t in group]
        
        kline = {
            'timestamp': key * interval * 1000,
            'open': prices[0],
            'high': max(prices),
            'low': min(prices),
            'close': prices[-1],
            'volume': sum(volumes)
        }
        klines.append(kline)
    
    return klines

这里有个容易被忽略的点：K线的开盘价和收盘价。开盘价是这一分钟第一笔成交的价格，收盘价是最后一笔成交的价格。如果你用错了顺序，整个技术指标都会算错。

本章知识体系总览

说了这么多，我画了一张图来总结本章的核心逻辑。从数据源到清洗，再到存储和使用，每一步都有对应的技术方案和注意事项。

这张图把整个数据流程串起来了。从最底层的交易所API，到中间的清洗对齐，再到上层的存储和应用。每一步都环环相扣，任何一个环节出问题，都会影响最终的交易决策。

好了，这一章的内容就到这里。数据采集与清洗看起来是脏活累活，但恰恰是区分专业做市商和业余玩家的分水岭。把这些基础打扎实了，后面的策略开发才能事半功倍。