3. 数据获取与清洗：交易所API对接、WebSocket实时数据流、数据清洗与对齐、缺失值处理与时间戳同步

各位同学，欢迎来到价差模型构建的第一道硬门槛——数据环节。

说实话，我见过太多策略在回测里跑得飞起，一上实盘就崩。十有八九，问题出在数据上。不是API断了，就是两个交易所的时间戳没对齐，价差算出来全是错的。所以这一章，咱们把数据获取和清洗的底子打牢。

3.1 交易所API对接：REST vs WebSocket

交易所API主要分两种：REST和WebSocket。我个人习惯把REST叫做“拉模式”，WebSocket叫“推模式”。

• REST API：你主动去问交易所要数据。适合获取历史K线、账户信息。缺点是慢，有频率限制。
• WebSocket：交易所主动把数据推给你。适合实时行情，延迟低。做价差交易，必须用WebSocket。

我在项目中遇到过一个问题：某交易所的REST接口每秒只能请求10次，我一开始没注意，结果程序跑着跑着就被封IP了。嗯，这里要注意，每个交易所的限频规则都不一样，一定要先读文档。

3.2 WebSocket实时数据流实战

WebSocket连接建立后，交易所会持续推送tick数据。一个典型的tick长这样：

{
  "channel": "btcusdt.trade",
  "ts": 1695000000123,
  "price": 27500.50,
  "size": 0.01,
  "side": "buy"
}

这里有个坑：时间戳的单位。有的交易所用毫秒，有的用微秒，甚至还有用秒的。我曾经因为没注意单位，把微秒当毫秒处理，结果价差序列全乱了。排查了整整一个下午。

另外，WebSocket断连是家常便饭。我建议写一个自动重连机制，心跳超时3秒就重连。别问我为什么是3秒，经验值。

3.3 数据清洗与对齐：两个交易所的“时差”问题

做价差交易，最核心的一步是时间对齐。假设交易所A在T1时刻报了一个买一价，交易所B在T2时刻报了一个卖一价。如果T1和T2差了500毫秒，你算出来的价差就是错的。

怎么对齐？我常用的方法是时间戳同步 + 线性插值。

1. 先统一时间戳基准。比如都转成UTC毫秒。
2. 设定一个固定的时间窗口（比如100ms一个切片）。
3. 每个切片内，取最新的一条数据。如果某个切片没有数据，用前后切片做线性插值。

3.4 缺失值处理：别让NaN毁了你的策略

数据缺失的原因很多：网络抖动、交易所维护、行情不活跃。处理方式就几种：

• 删除：如果缺失比例小于1%，直接扔掉。
• 向前填充：用上一个有效值填充。适合高频场景，因为价格变化连续。
• 线性插值：用前后两个有效值做线性拟合。适合中低频。
• 不处理：有些策略模型能容忍NaN，但价差模型不行。

我个人习惯：先向前填充，再线性插值。为什么？因为高频场景下，价格跳跃不会太大，向前填充最接近真实情况。如果连续缺失超过5个tick，我建议直接标记为无效数据，不要强行填充。

3.5 时间戳同步：NTP与本地时钟漂移

你的服务器和交易所服务器之间，存在时钟漂移。哪怕只差10毫秒，对高频价差交易来说都是致命的。

解决方案：

• 服务器配置NTP服务，每5分钟同步一次。
• 程序启动时，记录本地时间与交易所时间的偏差。
• 运行过程中，持续监控偏差变化。如果偏差超过50毫秒，发出告警。

3.6 本章知识体系总览

下面这张图，是我自己总结的数据处理流程。你照着这个框架搭，基本不会出大问题。

3.7 代码骨架示例

下面是一个简化的数据清洗流程代码。你拿回去改改就能用。

import pandas as pd
import numpy as np

def align_tick_data(df_a, df_b, freq='100ms'):
    """
    对齐两个交易所的tick数据
    df_a, df_b: 必须包含 'ts' (毫秒时间戳) 和 'price' 列
    """
    # 统一时间戳索引
    df_a.set_index('ts', inplace=True)
    df_b.set_index('ts', inplace=True)
    
    # 重采样到固定频率
    resampled_a = df_a['price'].resample(freq).last()
    resampled_b = df_b['price'].resample(freq).last()
    
    # 合并
    aligned = pd.concat([resampled_a, resampled_b], axis=1)
    aligned.columns = ['price_a', 'price_b']
    
    # 缺失值处理：先向前填充，再线性插值
    aligned.fillna(method='ffill', inplace=True)
    aligned.interpolate(method='linear', inplace=True)
    
    # 丢弃仍然为NaN的行
    aligned.dropna(inplace=True)
    
    # 计算价差
    aligned['spread'] = aligned['price_a'] - aligned['price_b']
    
    return aligned

好了，数据获取与清洗这部分就讲到这里。记住一句话：数据干净，策略才敢上仓位。下一章咱们聊聊价差模型的核心——协整关系与统计套利。不过那是后话了，先把今天的内容消化掉。

---

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