3、数据获取与清洗：多交易所API对接、数据对齐、缺失值处理

做价差策略，最头疼的其实不是策略逻辑本身。

而是数据。我见过太多人，策略写得很漂亮，一上实盘就崩。为什么？数据没对齐。时间戳差了一秒，交易对名字写错一个字母，或者某个交易所突然断流了五分钟。这些细节，才是真正决定策略生死的东西。

3.1 多交易所API对接：别小看这步

我个人习惯，先把交易所的API文档通读一遍。别偷懒。每个交易所的限频规则、返回格式、WebSocket订阅方式，都不一样。

举个例子，Binance的REST API返回的是毫秒时间戳，而OKX返回的是Unix秒级时间戳。你如果不做转换，直接拿来用，数据就全乱了。

我建议你封装一个统一的接口层。像这样：

class ExchangeAPI:
    def __init__(self, exchange_name):
        self.name = exchange_name
        if exchange_name == 'binance':
            self.client = BinanceClient()
        elif exchange_name == 'okx':
            self.client = OKXClient()
        # 其他交易所同理

    def get_ticker(self, symbol):
        raw = self.client.fetch_ticker(symbol)
        # 统一返回格式
        return {
            'exchange': self.name,
            'symbol': symbol,
            'bid': raw['bid'],
            'ask': raw['ask'],
            'timestamp': self._normalize_ts(raw['timestamp'])
        }

    def _normalize_ts(self, ts):
        # 统一转为毫秒
        if ts < 1e12:  # 秒级时间戳
            return ts * 1000
        return ts

这样做的好处是，后面所有策略代码都不用关心底层是哪个交易所。换交易所，改一行配置就行。

3.2 数据对齐：时间戳和交易对

数据对齐，说白了就是让不同交易所的数据站在同一条起跑线上。

时间戳对齐是最容易出问题的。两个交易所的行情到达你的服务器，可能有几十到几百毫秒的延迟。你不能直接拿它们的时间戳去比较。

我常用的方法是：

1. 所有数据统一使用服务器本地时间作为基准
2. 收到数据时，记录服务器时间，而不是交易所时间戳
3. 然后按固定的时间窗口（比如1秒）做切片对齐

代码实现大概是这样：

import pandas as pd
import numpy as np

def align_data(df_binance, df_okx, freq='1s'):
    """
    将两个交易所的tick数据对齐到固定频率
    """
    # 用服务器时间作为索引
    df_binance.set_index('server_time', inplace=True)
    df_okx.set_index('server_time', inplace=True)

    # 重采样到1秒
    df_binance_resampled = df_binance.resample(freq).last()
    df_okx_resampled = df_okx.resample(freq).last()

    # 合并
    aligned = pd.merge(
        df_binance_resampled[['bid', 'ask']],
        df_okx_resampled[['bid', 'ask']],
        left_index=True, right_index=True,
        suffixes=('_binance', '_okx')
    )
    return aligned

交易对对齐也是个坑。同一个币种，不同交易所的命名可能不一样。比如BTC/USDT，在Binance叫BTCUSDT，在OKX叫BTC-USDT，在Coinbase叫BTC-USD。

我建议维护一个统一的交易对映射表：

SYMBOL_MAP = {
    'BTC/USDT': {
        'binance': 'BTCUSDT',
        'okx': 'BTC-USDT',
        'coinbase': 'BTC-USD'
    },
    'ETH/USDT': {
        'binance': 'ETHUSDT',
        'okx': 'ETH-USDT',
        'coinbase': 'ETH-USD'
    }
}

这样，策略里只用统一交易对名称，底层自动映射。

3.3 缺失值处理：别让数据坑了你

数据缺失是常态。交易所断流、网络抖动、行情暂停，都会导致数据缺失。

我遇到过最离谱的一次，某个交易所因为系统升级，停了整整两分钟的数据。那两分钟里，我的价差信号直接变成NaN。如果不处理，策略会以为价差归零，然后疯狂开仓。

处理缺失值，我一般分三步：

1. 检测：用pandas的isnull()检查哪些位置缺失
2. 标记：连续缺失超过N个点，直接标记为无效区间
3. 填充：短时间缺失，用前向填充或插值

代码示例：

def clean_missing_data(df, max_gap=5):
    """
    清洗缺失值
    max_gap: 连续缺失超过这个数，直接丢弃
    """
    # 标记连续缺失
    df['gap_flag'] = df['spread'].isnull().astype(int)
    df['gap_group'] = df['gap_flag'].ne(df['gap_flag'].shift()).cumsum()

    # 计算每组缺失的长度
    gap_lengths = df[df['gap_flag']==1].groupby('gap_group').size()

    # 长缺失直接设为NaN
    long_gaps = gap_lengths[gap_lengths > max_gap].index
    df.loc[df['gap_group'].isin(long_gaps), 'spread'] = np.nan

    # 短缺失用前向填充
    df['spread'].fillna(method='ffill', inplace=True)

    # 如果开头还有缺失，用后向填充
    df['spread'].fillna(method='bfill', inplace=True)

    return df[['spread']]

3.4 整体数据流架构

下面这张图，是我在实际项目中用的数据流架构。你可以参考一下：

这个架构的核心思想是：把数据获取和策略逻辑彻底解耦。你想想看，如果策略代码里到处是交易所API调用，那维护起来得多痛苦？

嗯，这里要注意一点。数据清洗不是一次性的工作。市场环境在变，交易所的API也在变。我建议你每周检查一次数据质量，看看有没有新的异常模式出现。

我曾经就因为没及时更新交易对映射表，导致某个新上线的币种一直没被策略捕捉到。后来加了自动检测机制，才解决这个问题。

好了，数据获取与清洗这部分，核心就是这些。记住一句话：数据质量决定策略上限。再好的策略，喂进去的是垃圾数据，出来的也只能是垃圾结果。

---

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