3、数据获取与清洗：WebSocket实时行情接入、REST API历史数据获取、数据对齐与缺失值处理

做交易所间套利，说白了就是跟数据打交道。你策略再牛，模型再漂亮，拿到的数据是脏的、慢的、对不齐的，那一切都是白搭。我个人习惯把数据这块分成三个层次：实时行情怎么接进来、历史数据怎么拉下来、以及最头疼的——怎么让不同交易所的数据在时间轴上对齐。

嗯，咱们一个一个说。

3.1 WebSocket 实时行情接入

实时行情，我建议你优先用 WebSocket。为什么？因为 REST API 轮询太慢了，而且容易被交易所限频。你想想看，套利机会往往就存在那么几十毫秒，轮询还没回来，价差已经没了。

WebSocket 接入的核心逻辑其实不复杂：建立连接、订阅频道、处理推送。但坑不少。

核心要点： 每个交易所的 WebSocket 协议都不一样，别指望一套代码通吃。我一般会为每个交易所单独封装一个连接器。

下面是一个简化版的 WebSocket 接入示例，以币安为例：

import asyncio
import websockets
import json

class BinanceWebSocket:
    def __init__(self, symbols):
        self.symbols = symbols
        self.url = "wss://stream.binance.com:9443/ws"
        self.conn = None
        
    async def connect(self):
        # 建立连接
        self.conn = await websockets.connect(self.url)
        # 订阅深度数据
        subscribe_msg = {
            "method": "SUBSCRIBE",
            "params": [f"{sym.lower()}@depth20@100ms" for sym in self.symbols],
            "id": 1
        }
        await self.conn.send(json.dumps(subscribe_msg))
        
    async def receive(self):
        while True:
            try:
                data = await self.conn.recv()
                # 解析并处理数据
                yield json.loads(data)
            except Exception as e:
                print(f"连接异常: {e}")
                # 重连逻辑
                await self.reconnect()
                break

这里有个细节要注意：心跳维持。很多交易所要求客户端定期发送 ping 帧，否则会断开连接。我曾经遇到过一整个晚上数据断流，就是因为忘了处理心跳，第二天复盘时发现策略根本没跑。

我的经验： 建议在 WebSocket 连接层加一个 watchdog 定时器，如果超过 30 秒没收到任何数据，主动触发重连。别指望交易所会通知你连接断了。

3.2 REST API 历史数据获取

历史数据主要用于回测和策略参数优化。REST API 获取历史数据，说白了就是发 HTTP 请求，但有几个坑你得提前知道。

第一个坑：限频。交易所对 REST API 的调用频率有严格限制。币安是 1200 次/分钟，火币是 100 次/秒。你如果一股脑并发请求，很快就会被封 IP。

第二个坑：数据分页。大部分交易所的历史 K 线接口一次只能返回 500-1000 条数据。你要拉一年的 1 分钟数据，得循环请求几百次。

我一般会这样处理：

import requests
import time
from datetime import datetime, timedelta

def fetch_historical_klines(symbol, interval, start_time, end_time):
    """
    分段获取历史K线数据
    """
    base_url = "https://api.binance.com/api/v3/klines"
    all_data = []
    current_start = start_time
    
    while current_start < end_time:
        params = {
            "symbol": symbol,
            "interval": interval,
            "startTime": int(current_start.timestamp() * 1000),
            "limit": 1000
        }
        
        try:
            resp = requests.get(base_url, params=params)
            resp.raise_for_status()
            data = resp.json()
            
            if not data:
                break
                
            all_data.extend(data)
            # 更新起始时间
            current_start = datetime.fromtimestamp(data[-1][0] / 1000)
            # 限频控制
            time.sleep(0.1)
            
        except Exception as e:
            print(f"请求失败: {e}")
            time.sleep(1)
            continue
    
    return all_data

注意： 不同交易所的 K 线数据时间戳格式不一样。币安是毫秒级时间戳，OKX 是秒级时间戳，火币又是毫秒级。你拉下来之后一定要统一转换成标准时间格式，不然后面对齐时会疯掉。

3.3 数据对齐与缺失值处理

这才是整个数据环节里最磨人的部分。不同交易所的数据到达时间不一样，深度更新频率不一样，甚至同一个交易所的不同交易对，数据推送也有延迟。

数据对齐，说白了就是让不同数据源在同一个时间点上都有值。我常用的方法有两种：

对齐方式 适用场景 优缺点
时间戳对齐 K线数据、Ticker数据 简单直接，但可能丢失精度
事件驱动对齐 深度数据、逐笔成交 精度高，但实现复杂

时间戳对齐的做法是：选定一个基准时间（比如每秒的整点），然后把落在同一秒内的数据归为一组。如果某个交易所这一秒没有数据，就用上一秒的数据填充。

事件驱动对齐更精细一些：每当收到一个交易所的更新，就检查其他交易所是否有对应的最新数据。如果都有，就生成一个对齐后的快照。

我个人的经验是：做套利策略，事件驱动对齐更靠谱。因为套利机会往往出现在数据更新的瞬间，时间戳对齐会把这些瞬间抹平。

缺失值处理三板斧：

  • 前向填充：用上一个有效值填充。适用于价格数据，因为价格不会突变。
  • 线性插值：用前后两个值的平均值填充。适用于成交量等连续数据。
  • 直接丢弃：如果缺失比例超过 5%，建议直接丢弃该时间段的数据。别硬填，会引入偏差。

我曾经在回测时发现一个诡异的现象：策略在回测里赚得盆满钵满，实盘却亏得一塌糊涂。查了两天才发现，原来是数据对齐时用了简单的时间戳对齐，导致回测里出现了很多「不存在的套利机会」。说白了，就是数据没对齐，让策略误以为两个交易所同时出现了价差。

避坑指南： 数据对齐后，一定要做一次「回放验证」。把对齐后的数据按时间顺序逐条播放，检查每个时间点上各交易所的数据是否合理。我一般会写一个可视化工具，把对齐结果画成折线图，一眼就能看出问题。

3.4 本章知识体系

下面这张图概括了数据获取与清洗的完整流程：

数据获取与清洗流程 WebSocket 实时行情 REST API 历史数据 数据清洗 格式统一 | 异常值过滤 | 去重 | 时间戳标准化 数据对齐 时间戳对齐 | 事件驱动对齐 | 缺失值处理 对齐后的标准数据集

从这张图可以看得很清楚：数据从两个源头进来，经过清洗、对齐，最终输出标准化的数据集。每一步都有坑，每一步都需要你亲自踩过才知道怎么处理。

嗯，数据这块就讲这么多。记住一句话：数据质量决定策略上限。你花再多时间在数据清洗上都不为过。

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

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