第3章数据获取与清洗

数据获取与清洗：交易所API对接、WebSocket实时流、数据去重与对齐、Tick级数据存储

做高频交易，第一关就是数据。没有干净、对齐的Tick数据，后面所有策略都是空中楼阁。我见过太多团队，策略在回测里跑得飞起，一上实盘就崩——十有八九是数据源出了问题。

今天咱们就聊聊，怎么从交易所把原始数据扒下来，怎么洗干净，怎么存起来。嗯，这里要注意，每一步都有坑。

交易所API对接：选对协议，少走弯路

交易所API主要分两类：REST和WebSocket。REST适合拿历史数据，WebSocket适合拿实时流。我个人习惯是，初始化时用REST拉一段快照，然后切到WebSocket做增量更新。

举个例子，币安的现货API：

import requests
import json

# 获取深度快照
url = "https://api.binance.com/api/v3/depth"
params = {"symbol": "BTCUSDT", "limit": 100}
resp = requests.get(url, params=params)
data = resp.json()
print(f"Bids: {data['bids'][:3]}, Asks: {data['asks'][:3]}")

这里有个坑——限频。交易所对REST请求有严格的频率限制，比如每秒最多10次。我曾经在项目里没注意这个，直接写了个死循环去拉数据，结果IP被交易所封了半小时。后来学乖了，加了个sleep(0.1)。

避坑指南： 我曾经在对接某二线交易所时，发现它的REST接口返回的深度数据居然有重复的订单ID。后来查文档才知道，它用的是「快照+增量」模式，但增量事件没处理好。所以，对接任何交易所，第一件事就是写个校验脚本，检查数据完整性。

WebSocket实时流：心跳与重连

WebSocket是高频交易的生命线。但网络不稳定，断连是家常便饭。你想想看，如果行情断了一秒，你的策略可能就错过了关键价位。

标准的做法是：订阅深度流，同时监听心跳。币安的WebSocket每3分钟发一次ping帧，客户端必须在10分钟内回复pong，否则连接会被断开。

import websocket
import json

def on_message(ws, message):
    data = json.loads(message)
    if 'bids' in data:
        # 处理深度更新
        print(f"Update: {data['bids'][:2]}")
    elif 'ping' in data:
        ws.send(json.dumps({"pong": data['ping']}))

def on_error(ws, error):
    print(f"WebSocket error: {error}")
    # 自动重连逻辑
    ws.close()
    ws.run_forever()

ws = websocket.WebSocketApp(
    "wss://stream.binance.com:9443/ws/btcusdt@depth20",
    on_message=on_message,
    on_error=on_error
)
ws.run_forever()

这里有个细节：重连时一定要带上上次的lastUpdateId。否则你会拿到一个不完整的深度快照，导致后续所有数据都错位。我有个同事就因为这个，回测时发现买卖价差经常为负，查了两天才找到原因。

个人经验： 我建议在WebSocket回调里加一个计数器，记录每秒收到的消息数。如果连续5秒低于阈值，就主动触发重连。这比单纯依赖心跳更可靠。

数据去重与对齐：Tick级数据的灵魂

原始数据拿到手，你会发现一堆问题：重复、乱序、缺失。为什么？因为交易所的撮合引擎是分布式的，不同节点的时钟有偏差。

去重很简单——用(exchange, symbol, trade_id)作为唯一键。但要注意，有些交易所的trade_id是全局递增的，有些是按交易对递增的。我遇到过一家，它的trade_id居然会重置，搞得我不得不加上时间戳做二次校验。

对齐就麻烦多了。不同交易所的Tick数据，时间戳精度不一样。币安是毫秒级，Coinbase是微秒级，OKX是纳秒级。你想想看，如果不对齐，回测时你会发现同一个时刻，两个交易所的买卖价差差了0.1秒，这在高频策略里是致命的。

我的做法是：统一转成纳秒级时间戳，然后按时间窗口做最近邻匹配。窗口大小一般设为1毫秒，如果超过这个范围，就标记为「未对齐」。

import pandas as pd

def align_ticks(binance_df, coinbase_df, window='1ms'):
    # 将时间戳转为纳秒
    binance_df['ts_ns'] = pd.to_datetime(binance_df['timestamp'], unit='ms')
    coinbase_df['ts_ns'] = pd.to_datetime(coinbase_df['timestamp'], unit='us')
    
    # 按时间窗口合并
    aligned = pd.merge_asof(
        binance_df.sort_values('ts_ns'),
        coinbase_df.sort_values('ts_ns'),
        on='ts_ns',
        tolerance=pd.Timedelta(window),
        direction='nearest'
    )
    return aligned

  核心要点： 数据对齐不是简单的「取交集」。你要决定哪个交易所作为主时钟。我一般选流动性最好的那个，因为它的Tick更新最快，时间戳最可靠。

Tick级数据存储：选对格式，省一半空间

高频数据量巨大。一个BTCUSDT交易对，一天就能产生几百万条Tick。如果存成CSV，一个月的原始数据就能吃掉几百GB。

我推荐用Parquet格式。它支持列式存储、压缩率高、读取速度快。同样的数据，Parquet比CSV小5-10倍，读取速度快10倍以上。

import pyarrow.parquet as pq
import pyarrow as pa

# 将DataFrame写入Parquet
table = pa.Table.from_pandas(df)
pq.write_table(table, 'btcusdt_ticks.parquet', compression='snappy')

# 读取时只加载需要的列
df = pq.read_table('btcusdt_ticks.parquet', columns=['ts_ns', 'price', 'volume']).to_pandas()

存储结构上，我习惯按交易对+日期分目录。比如：

data/
├── BTCUSDT/
│   ├── 2024-01-01.parquet
│   ├── 2024-01-02.parquet
│   └── ...
├── ETHUSDT/
│   ├── 2024-01-01.parquet
│   └── ...
└── ...

这样查询某一天的数据时，只需要加载一个文件，而不是遍历整个数据集。嗯，这里要注意，不要把所有数据塞到一个文件里。我见过有人这么干，结果一个Parquet文件几十GB，读一次要等半小时。

避坑指南： 我曾经用HDF5存Tick数据，结果发现写入速度越来越慢。后来排查发现，HDF5的索引机制在频繁追加写入时会产生大量碎片。换成Parquet后，这个问题就解决了。所以，高频场景下，Parquet比HDF5更靠谱。

知识体系总览

下面这张图，概括了本章的核心流程。从交易所到存储，每一步都有对应的技术和注意事项。

这张图把整个流程串起来了。从交易所API出发，经过WebSocket实时流，再到数据清洗（去重+对齐），最后落地到Parquet存储。每一步都有对应的技术选型和注意事项。

好了，数据获取与清洗就聊到这儿。记住一句话：数据质量决定策略上限。花80%的时间在数据上，剩下20%写策略，这才是高频交易的正确姿势。

📚 课程目录

数据获取与清洗：交易所API对接、WebSocket实时流、数据去重与对齐、Tick级数据存储

交易所API对接：选对协议，少走弯路

WebSocket实时流：心跳与重连

数据去重与对齐：Tick级数据的灵魂

Tick级数据存储：选对格式，省一半空间

知识体系总览