第四节：多合约数据获取

做跨期套利，说白了就是吃价差波动的饭。但有个前提——你得先拿到干净、对齐的数据。我见过太多人策略写得漂亮，结果数据源没处理好，一上线就亏钱。

今天我们就聊聊怎么用CCXT和Websocket，把多个期货合约的实时行情抓下来，再清洗对齐。嗯，这部分我踩过的坑不少，咱们一个个说。

4.1 为什么不用REST API？

你可能会问：直接用CCXT的fetch_ticker不行吗？当然行，但有个问题——速度。

REST API是轮询模式，你每隔1秒请求一次，实际拿到的是0.5秒前的数据。做高频套利时，这0.5秒的延迟可能让你错过最佳入场点。我在实盘中遇到过，用REST API抓两个合约的价差，结果因为时间戳不对齐，价差信号频繁误报。

Websocket是推送模式，交易所主动把数据推给你。延迟低，数据实时性高。做跨期套利，我建议优先用Websocket。

4.2 CCXT + Websocket 实战

CCXT本身不直接支持Websocket，但我们可以用ccxt.pro——它是CCXT的Websocket版本。安装很简单：

pip install ccxt.pro

然后我们写个示例，同时订阅两个合约的ticker：

import ccxt.pro as ccxt
import asyncio

async def watch_tickers(exchange, symbols):
    """同时监听多个合约的实时行情"""
    while True:
        try:
            # 批量订阅ticker
            tickers = await exchange.watch_tickers(symbols)
            for symbol, ticker in tickers.items():
                print(f"{symbol}: 最新价 {ticker['last']}, 时间 {ticker['datetime']}")
        except Exception as e:
            print(f"连接异常: {e}")
            # 断线重连
            await asyncio.sleep(1)

async def main():
    exchange = ccxt.binance({
        'enableRateLimit': True,
        'options': {
            'defaultType': 'future',  # 期货合约
        }
    })
    
    # 以BTC永续和当季合约为例
    symbols = ['BTC/USDT:USDT', 'BTC-240628']
    await watch_tickers(exchange, symbols)

if __name__ == '__main__':
    asyncio.run(main())

这里有个细节：watch_tickers返回的是一个字典，key是合约名，value是ticker对象。我习惯把时间戳统一转成毫秒，方便后续对齐。

4.3 数据清洗：别让脏数据毁了你的策略

拿到原始数据后，不能直接用。为什么？因为交易所的数据偶尔会有异常值。比如某次网络抖动，推送了一个价格为0的ticker。如果你没过滤，价差瞬间变成几千美金，策略直接开仓——然后爆仓。

我一般做这几步清洗：

1. 空值检查： 检查ticker的last、bid、ask是否为None
2. 价格合理性： 价格是否在合理范围内（比如BTC不可能为0或1亿）
3. 时间戳连续性： 检查时间戳是否单调递增，防止数据回退
4. 去重： 同一时间戳的重复数据只保留一条

代码实现大概这样：

def clean_ticker(ticker, symbol):
    """清洗单个ticker数据"""
    # 1. 空值检查
    if ticker['last'] is None or ticker['bid'] is None or ticker['ask'] is None:
        return None
    
    # 2. 价格合理性（以BTC为例）
    price = ticker['last']
    if price <= 0 or price > 1000000:
        return None
    
    # 3. 时间戳转毫秒
    timestamp = ticker['timestamp']
    if timestamp < 1e12:  # 如果是秒级时间戳
        timestamp *= 1000
    
    return {
        'symbol': symbol,
        'price': price,
        'bid': ticker['bid'],
        'ask': ticker['ask'],
        'timestamp': timestamp,
        'volume': ticker['baseVolume']
    }

4.4 数据对齐：跨期套利的核心难点

清洗完数据后，最头疼的问题来了——对齐。两个合约的ticker推送频率不一样，时间戳也不一样。比如BTC永续合约每秒推送10次，当季合约每秒只推送2次。你怎么计算价差？

我常用的方法有两种：

方法	原理	适用场景
时间戳对齐	以最新到达的数据为准，丢弃旧数据	高频做市，延迟敏感
插值对齐	用线性插值估算中间时刻的价格	中低频策略，需要平滑数据

我个人更推荐时间戳对齐。原因很简单——做市策略要的是实时性，不是平滑性。你想想看，如果插值出来的价格和实际价格差了0.1%，你的套利空间可能就没了。

时间戳对齐的代码：

class DataAligner:
    """多合约数据对齐器"""
    def __init__(self):
        self.latest_data = {}  # 存储每个合约的最新数据
    
    def update(self, symbol, ticker):
        """更新某个合约的数据，并尝试对齐"""
        self.latest_data[symbol] = ticker
        
        # 检查是否所有合约都有数据
        if len(self.latest_data) < 2:
            return None
        
        # 取所有合约中时间戳最小的（最旧的）作为基准
        timestamps = [v['timestamp'] for v in self.latest_data.values()]
        min_ts = min(timestamps)
        
        # 如果某个合约的数据太旧（超过100ms），丢弃
        aligned = {}
        for sym, data in self.latest_data.items():
            if data['timestamp'] - min_ts < 100:  # 100ms阈值
                aligned[sym] = data
        
        if len(aligned) == 2:
            return aligned
        return None

这里有个阈值100ms，是我根据经验调的。如果两个合约的数据时间差超过100ms，我宁愿等下一次推送，也不强行对齐。因为时间差越大，价差计算越不准。

4.5 整体架构图

说了这么多，咱们画个图总结一下整个数据流程：

这个流程看起来简单，但每个环节都有坑。比如Websocket断线重连、数据清洗的阈值设置、对齐的时间窗口——这些都需要根据实际交易对和交易所特性来调。

4.6 实战建议

最后给几个实战建议：

• 先做回测，再上实盘： 用历史数据模拟Websocket推送，验证你的清洗和对齐逻辑
• 加日志记录： 每条推送都记录原始数据和处理后的数据，方便排查问题
• 设置数据健康检查： 如果连续5秒没有收到某个合约的数据，触发告警
• 别迷信交易所数据： 我遇到过交易所推送的ticker和实际成交价差0.5%的情况，建议用多个数据源交叉验证

好了，数据获取这块就聊到这儿。记住一句话：数据质量决定策略上限。你策略再牛，数据是脏的，结果就是亏钱。下一节我们聊聊怎么用这些对齐后的数据计算价差和做市信号。

---

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