数据源与API：多交易所API接入、行情数据清洗与对齐

做套利的人，最怕什么？

不是策略亏钱，而是数据错了。

我见过太多团队，策略逻辑写得漂漂亮亮，结果一上线就亏。查到最后，发现是某个交易所的行情延迟了200毫秒，或者两个交易所的K线时间戳没对齐。嗯，这种坑，我自己也踩过。

今天我们就来聊聊，怎么把多交易所的数据源管好。

一、多交易所API接入的通用架构

说白了，每个交易所的API都长得不一样。有的用REST，有的用WebSocket，有的鉴权方式还特别奇葩。但不管怎么变，核心流程就三步：

1. 建立连接——拿到数据流
2. 数据解析——把JSON/Protobuf转成统一格式
3. 数据分发——喂给策略模块

我个人习惯，会先封装一个抽象基类。所有交易所的适配器都继承它。这样换交易所时，只需要改一个文件。

二、REST与WebSocket的取舍

做套利，行情延迟就是钱。我建议这样选：

场景	推荐方式	原因
获取历史K线	REST	一次性拉取，不需要实时
实时Ticker	WebSocket	延迟低，毫秒级推送
深度订单簿	WebSocket + 增量更新	全量快照太费带宽
交易/账户	REST	安全性高，出错可重试

我记得有一次，团队用REST去拉深度数据，每秒请求一次。结果交易所限流了，直接封了IP。后来改成WebSocket增量更新，带宽省了90%，数据还更实时。

三、行情数据清洗——那些不得不做的脏活

你以为交易所推过来的数据就是干净的？太天真了。

我遇到过的情况：

• 某交易所突然推送一个价格为0的Ticker
• 某合约的成交量在某个时刻变成负数
• 某K线的最高价比最低价还低

这些数据如果直接喂给策略，后果不堪设想。

所以，清洗是必须的。我一般会做这几步：

1. 空值检查——字段为None或NaN的直接丢弃
2. 范围校验——价格、成交量必须在合理区间内
3. 逻辑校验——最高价 ≥ 开盘价 ≥ 最低价，成交量 ≥ 0
4. 时间戳校验——不能是未来的时间，不能是太旧的数据

四、时间对齐——套利的核心难点

做跨交易所套利，最头疼的就是时间对齐。

为什么？因为每个交易所的服务器时间不一样。有的快几毫秒，有的慢几十毫秒。你看到A交易所价格是100，B交易所价格是101，觉得有套利空间。但很可能，这两个价格根本就不是同一时刻的。

我的做法是这样的：

1. 统一时间基准——所有数据到达后，打上本地时间戳（用NTP同步过的）
2. 数据对齐——把不同交易所的行情，按本地时间戳对齐到同一个时间窗口
3. 延迟补偿——如果某个交易所的WebSocket延迟稳定在50ms，就手动减去这个偏移

五、代码示例：一个简单的多交易所行情聚合器

下面是我写的一个简化版。它从两个交易所拉取Ticker，清洗后对齐输出。

import time
import threading
from dataclasses import dataclass
from typing import Dict, Optional

@dataclass
class Ticker:
    exchange: str
    symbol: str
    bid: float
    ask: float
    local_ts: float  # 本地时间戳

class ExchangeAdapter:
    """交易所适配器基类"""
    def fetch_ticker(self, symbol: str) -> Optional[Ticker]:
        raise NotImplementedError

class ExchangeA(ExchangeAdapter):
    def fetch_ticker(self, symbol: str) -> Optional[Ticker]:
        # 模拟API调用
        time.sleep(0.01)
        return Ticker(
            exchange="A",
            symbol=symbol,
            bid=100.0,
            ask=100.1,
            local_ts=time.time()
        )

class ExchangeB(ExchangeAdapter):
    def fetch_ticker(self, symbol: str) -> Optional[Ticker]:
        time.sleep(0.02)
        return Ticker(
            exchange="B",
            symbol=symbol,
            bid=99.9,
            ask=100.0,
            local_ts=time.time()
        )

class DataCleaner:
    """数据清洗层"""
    @staticmethod
    def clean(ticker: Optional[Ticker]) -> bool:
        if ticker is None:
            return False
        if ticker.bid <= 0 or ticker.ask <= 0:
            return False
        if ticker.bid > ticker.ask:
            return False
        # 检查时间戳是否在合理范围内（最近5秒）
        if abs(time.time() - ticker.local_ts) > 5:
            return False
        return True

class TickerAggregator:
    """行情聚合器"""
    def __init__(self):
        self.adapters = {
            "A": ExchangeA(),
            "B": ExchangeB()
        }
        self.cleaner = DataCleaner()
        self.latest: Dict[str, Ticker] = {}

    def poll_all(self, symbol: str):
        """轮询所有交易所"""
        for name, adapter in self.adapters.items():
            ticker = adapter.fetch_ticker(symbol)
            if self.cleaner.clean(ticker):
                self.latest[name] = ticker
            else:
                print(f"丢弃来自 {name} 的脏数据")

    def get_aligned(self) -> Dict[str, Ticker]:
        """获取对齐后的行情"""
        return self.latest.copy()

# 使用示例
agg = TickerAggregator()
agg.poll_all("BTC/USDT")
aligned = agg.get_aligned()
for ex, t in aligned.items():
    print(f"{ex}: bid={t.bid}, ask={t.ask}, ts={t.local_ts}")

这段代码虽然简单，但骨架是完整的。实际生产环境里，我会加上线程池、重试机制、监控告警。嗯，这些后面章节再细聊。

六、知识体系总览

下面这张图，是我自己总结的多市场数据接入的核心流程。你可以把它贴在工位旁边。

你看，整个流程其实不复杂。但每个环节都有坑。数据源层要注意API限流和断线重连，清洗层要小心脏数据漏网，对齐层要处理好延迟补偿。

我个人建议，刚开始做的时候，先把清洗和对齐这两块写稳。策略可以简单，但数据必须可靠。数据不对，再好的策略也是白搭。

---

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