3. 数据获取与清洗：交易所API对接、WebSocket实时数据流、数据对齐与清洗、缺失值处理、数据存储方案

做跨交易所套利，说白了就是跟数据打交道。我见过太多人策略写得漂亮，结果死在数据上——不是延迟太高，就是两个交易所的数据对不上。今天咱们就把这块硬骨头啃下来。

3.1 交易所API对接：选对工具，事半功倍

每个交易所都有自己的API，但核心逻辑差不多。我个人习惯用 ccxt 这个库，它封装了上百个交易所的接口，省去了重复造轮子的麻烦。

举个例子，对接币安和OKX的现货行情：

import ccxt
import time

# 初始化交易所对象
binance = ccxt.binance({
    'rateLimit': 1200,  # 限频控制
    'enableRateLimit': True,
})
okx = ccxt.okx({
    'rateLimit': 600,
    'enableRateLimit': True,
})

# 获取ticker数据
def fetch_tickers(exchange, symbols):
    try:
        tickers = exchange.fetch_tickers(symbols)
        return tickers
    except ccxt.RateLimitExceeded as e:
        print(f"触发限频，等待{exchange.rateLimit}ms")
        time.sleep(exchange.rateLimit / 1000)
        return fetch_tickers(exchange, symbols)
    except Exception as e:
        print(f"请求失败: {e}")
        return None

symbols = ['BTC/USDT', 'ETH/USDT']
btc_ticker_binance = fetch_tickers(binance, symbols)
btc_ticker_okx = fetch_tickers(okx, symbols)

这里有个坑——限频。我曾经在实盘时没处理好，直接触发了交易所的IP封禁，整整半小时没法交易。所以一定要加重试机制和退避策略。

3.2 WebSocket实时数据流：速度就是生命

套利交易拼的就是速度。REST API轮询太慢了，毫秒级的延迟都可能让利润溜走。所以必须上WebSocket。

WebSocket的好处是：建立一次连接，服务器主动推送数据。延迟通常在10-50ms，比REST快一个数量级。

来看一个简单的WebSocket订阅示例：

import websocket
import json

def on_message(ws, message):
    data = json.loads(message)
    # 处理深度数据
    if 'bids' in data:
        print(f"收到深度更新: {len(data['bids'])} 档买单")

def on_error(ws, error):
    print(f"WebSocket错误: {error}")

def on_close(ws, close_status_code, close_msg):
    print("连接关闭，尝试重连...")
    # 这里可以加自动重连逻辑

def on_open(ws):
    # 订阅BTC/USDT深度
    subscribe_msg = {
        "op": "subscribe",
        "args": ["depth.BTC-USDT"]
    }
    ws.send(json.dumps(subscribe_msg))

# 以OKX为例
ws_url = "wss://ws.okx.com:8443/ws/v5/public"
ws = websocket.WebSocketApp(ws_url,
                            on_open=on_open,
                            on_message=on_message,
                            on_error=on_error,
                            on_close=on_close)
ws.run_forever()

嗯，这里要注意：WebSocket连接不稳定是常态。网络抖动、服务器重启都可能导致断连。我建议至少实现以下机制：

• 心跳检测：每隔30秒发一次ping，确保连接存活
• 自动重连：断连后指数退避重试，最多重试5次
• 数据缓存：断连期间用REST API补数据

3.3 数据对齐与清洗：让两个交易所说同一种语言

数据拿到手了，但问题来了——不同交易所的数据格式、时间戳、精度都不一样。比如币安的时间戳是毫秒级，OKX是微秒级。直接拿来用？会出大问题。

数据对齐的核心是时间对齐和价格对齐。

时间对齐

我通常的做法是：

1. 统一将时间戳转为UTC毫秒
2. 按1秒或100ms的粒度切片
3. 取每个切片内的最新数据

import pandas as pd
import numpy as np

def align_data(binance_data, okx_data, freq='1s'):
    """
    对齐两个交易所的数据
    :param binance_data: DataFrame, 包含'timestamp'和'price'列
    :param okx_data: DataFrame, 包含'timestamp'和'price'列
    :param freq: 对齐频率，如'1s'、'100ms'
    """
    # 统一时间戳为毫秒
    binance_data['timestamp'] = pd.to_datetime(binance_data['timestamp'], unit='ms')
    okx_data['timestamp'] = pd.to_datetime(okx_data['timestamp'], unit='us')  # 微秒转毫秒

    # 设置索引并重采样
    binance_aligned = binance_data.set_index('timestamp').resample(freq).last()
    okx_aligned = okx_data.set_index('timestamp').resample(freq).last()

    # 合并
    aligned = pd.concat({
        'binance': binance_aligned['price'],
        'okx': okx_aligned['price']
    }, axis=1)

    return aligned.dropna()

你想想看，如果时间没对齐，价差计算全是错的。我见过有人用毫秒级数据直接减，结果价差波动大得离谱，其实是时间戳差了50毫秒。

价格对齐

不同交易所的价格精度不同。比如币安BTC价格是小数点后2位，OKX是小数点后4位。直接比较？得先统一精度。

我的做法是：以精度低的交易所为准，对精度高的做四舍五入。

def align_price_precision(price_binance, price_okx, precision=2):
    """统一价格精度"""
    price_binance = round(price_binance, precision)
    price_okx = round(price_okx, precision)
    return price_binance, price_okx

3.4 缺失值处理：别让空数据毁了你的策略

实盘中数据缺失是常态。网络波动、交易所维护、数据推送延迟，都可能导致某条数据为空。

常见的处理方法有：

方法	适用场景	缺点
向前填充	短期缺失（< 5个周期）	可能引入滞后
线性插值	价格缓慢变化时	剧烈波动时不准
删除缺失行	缺失比例低（< 1%）	可能丢失信息
用另一交易所数据补	两个交易所高度相关	引入外部偏差

我个人最常用的是向前填充 + 阈值判断。如果缺失超过3个周期，就标记为无效数据，不参与策略计算。

def handle_missing(data, max_gap=3):
    """
    处理缺失值
    :param data: Series
    :param max_gap: 最大允许连续缺失数
    """
    # 向前填充
    filled = data.fillna(method='ffill', limit=max_gap)

    # 标记超过阈值的缺失
    missing_mask = data.isna().rolling(window=max_gap+1).sum() > max_gap
    filled[missing_mask] = np.nan

    return filled

3.5 数据存储方案：既要快，又要稳

数据存哪里？这取决于你的使用场景。

• 实时计算：用Redis，内存数据库，读写都在微秒级
• 历史回测：用InfluxDB或ClickHouse，时序数据库，查询效率高
• 长期归档：用Parquet格式存到对象存储（如S3），压缩率高，查询快

我自己的架构是这样的：

# 实时数据 -> Redis
import redis
r = redis.Redis(host='localhost', port=6379, db=0)

def store_ticker(symbol, price, timestamp):
    key = f"ticker:{symbol}:{timestamp}"
    r.set(key, price, ex=60)  # 60秒过期

# 历史数据 -> InfluxDB
from influxdb import InfluxDBClient
client = InfluxDBClient(host='localhost', port=8086)
client.switch_database('crypto_data')

def store_history(symbol, price, timestamp):
    json_body = [
        {
            "measurement": "ticker",
            "tags": {"symbol": symbol},
            "time": timestamp,
            "fields": {"price": price}
        }
    ]
    client.write_points(json_body)

3.6 本章知识体系

下面这张图总结了数据获取与清洗的核心流程：

整个流程环环相扣。数据源是基础，对齐清洗是关键，存储方案决定效率。任何一个环节出问题，策略就跑偏。

好了，这一章的内容就到这里。数据获取与清洗是套利交易的基石，花再多时间打磨都不为过。下一章咱们聊聊价差计算与信号生成，那才是真正赚钱的地方。

---