4、数据获取与处理：交易所API对接、WebSocket实时行情、历史数据存储（CSV/数据库）、数据清洗与对齐

数据，是套利交易的血液。没有干净、及时的数据，再好的策略也是空中楼阁。这一章，我们聊聊怎么把数据从交易所「搬」到你的系统里，并且让它变得可用。

4.1 交易所API对接：你的第一道门槛

每个交易所都像个黑盒子，API就是打开它的钥匙。我个人习惯先把官方文档通读一遍，尤其是限频规则和错误码定义。这一步偷懒，后面会吃大亏。

常见的API类型有两种：

  • REST API：请求-响应模式，适合获取账户信息、下单、查询历史数据。说白了就是「你问一句，它答一句」。
  • WebSocket：全双工通信，适合订阅实时行情和成交数据。交易所会主动推数据给你，不用反复去问。

我在项目中遇到过最坑的事：某个交易所的REST接口，同一个请求在不同时间返回的字段名大小写不一致。嗯，后来我写了个自动校验层，每次请求都做一次字段映射，才彻底解决。

⚠️ 避坑指南：我曾经因为没处理好API限频，被交易所封了IP整整24小时。记住，每个接口都有调用频率限制，一定要在代码里加本地计数器，别指望交易所的返回码来提醒你。

4.2 WebSocket实时行情：别让数据迟到

套利交易对延迟极其敏感。REST轮询的方式，延迟通常在100ms以上，而WebSocket可以把延迟压到10ms以内。你想想看，价差可能就在这几十毫秒里消失。

一个典型的WebSocket订阅流程：

import websocket
import json

def on_message(ws, message):
    data = json.loads(message)
    # 这里处理行情数据
    process_ticker(data)

def on_error(ws, error):
    print(f"连接异常: {error}")
    # 自动重连逻辑
    reconnect()

ws = websocket.WebSocketApp(
    "wss://api.exchange.com/ws",
    on_message=on_message,
    on_error=on_error
)
ws.run_forever()

这里有个细节：心跳维持。很多交易所要求客户端定期发送ping帧，否则会断开连接。我建议在on_open回调里启动一个定时器，每30秒发一次ping。

💡 个人经验：WebSocket断线重连时，要重新订阅所有交易对。我习惯把订阅列表存在本地配置文件里，重连后自动读取并订阅，省得手动操作。

4.3 历史数据存储：CSV还是数据库？

这个问题我纠结了很久。后来想明白了：看用途。

存储方式 适用场景 优点 缺点
CSV 回测、小规模数据、快速验证 简单、可读性强、无需额外服务 查询慢、不支持并发、数据量大时卡顿
SQLite 单机应用、中等规模 轻量、支持SQL查询、事务安全 写入性能有限、不适合高并发
PostgreSQL/MySQL 生产环境、多客户端访问 高性能、支持复杂查询、数据安全 需要运维、资源消耗大
时序数据库（如InfluxDB） 高频行情、海量时间序列 写入极快、自动压缩、查询优化 学习成本高、生态相对小众

我个人习惯：回测用CSV，生产用PostgreSQL。CSV方便我随时用Excel打开看看数据长什么样，PostgreSQL则保证数据的一致性和查询效率。

4.4 数据清洗与对齐：脏数据比没数据更可怕

你从交易所拿到的原始数据，大概率是「脏」的。常见问题包括：

  • 缺失值：某根K线因为网络抖动没收到
  • 重复数据：WebSocket重连后重复推送了同一笔成交
  • 时间戳错乱：交易所服务器时钟不同步，导致数据顺序颠倒
  • 异常值：某个价格突然跳变100倍，明显是数据错误

我写过一个清洗流程，大致是这样：

def clean_ticker_data(df):
    # 1. 去重：按时间戳和交易对去重
    df = df.drop_duplicates(subset=['timestamp', 'symbol'])
    
    # 2. 排序：按时间戳升序排列
    df = df.sort_values('timestamp')
    
    # 3. 填充缺失值：用前一个有效值填充
    df = df.fillna(method='ffill')
    
    # 4. 过滤异常值：价格超过3倍标准差则剔除
    mean = df['price'].mean()
    std = df['price'].std()
    df = df[(df['price'] > mean - 3*std) & 
            (df['price'] < mean + 3*std)]
    
    return df
🔑 核心要点：数据对齐是套利交易的关键。不同交易所的行情数据，时间戳可能相差几毫秒。我建议把所有数据统一对齐到同一个时间基准（比如交易所的撮合时间），否则计算出的价差全是错的。

为什么会这样？举个例子：A交易所的行情延迟50ms，B交易所的行情延迟200ms。如果你直接用这两个数据计算价差，结果里包含了150ms的延迟误差，而不是真实的价差。我曾经因为这个bug，跑了一个月的回测全是假信号。

解决办法：时间戳对齐。把两个交易所的行情数据，按相同的时间窗口（比如1秒）做聚合，取窗口内的中间价或加权均价。这样虽然损失了一点精度，但保证了数据的一致性。

💡 一个小技巧：在存储历史数据时，额外保存一个「数据质量」字段。比如标记这条数据是原始数据、清洗后数据、还是插值数据。这样回测时可以根据质量等级做不同的处理，避免用脏数据污染策略评估。

嗯，数据这块就聊这么多。记住一句话：数据质量决定了策略的天花板。花80%的时间在数据处理上，一点都不夸张。

← 上一章 📖 返回目录 下一章 →