第四章：数据获取与清洗——套利分析的基石

做跨期套利，数据就是你的眼睛。眼睛要是花了，再好的策略也白搭。

我见过太多人，策略逻辑写得漂漂亮亮，回测曲线也完美，结果一上实盘就崩。为什么？数据源没对齐，时间戳差了那么几毫秒，或者某个异常值没处理干净。嗯，今天我们就来聊聊这个看似基础、实则要命的话题。

4.1 交易所API对接——第一道坎

说实话，对接交易所API这事儿，没什么技术含量，但坑特别多。我个人习惯是，先看文档，再看文档，最后还是看文档。

不同交易所的API风格差异很大。有的用RESTful，有的用WebSocket。做套利的话，我建议你优先搞定WebSocket。为什么？因为套利机会转瞬即逝，REST轮询的延迟你扛不住。

举个实际例子。我曾经对接过某商品交易所的API，他们的行情数据是分帧推送的。每帧包含一个时间戳、一个合约代码、以及买卖五档数据。看起来很简单对吧？但问题来了——他们的时间戳精度只到秒级。

你想想看，秒级精度对于高频套利意味着什么？同一秒内的两笔交易，你根本分不清谁先谁后。这要是用来做价差计算，结果可想而知。

代码层面，我一般这样处理WebSocket连接：

import websocket
import json

def on_message(ws, message):
    data = json.loads(message)
    # 这里要特别注意时间戳的处理
    # 我习惯把交易所时间戳和本地时间戳都记录下来
    tick = {
        'exchange_ts': data['timestamp'],
        'local_ts': time.time_ns(),
        'symbol': data['symbol'],
        'bid': data['bid'],
        'ask': data['ask']
    }
    process_tick(tick)

ws = websocket.WebSocketApp(
    "wss://api.exchange.com/ws",
    on_message=on_message
)
ws.run_forever()

4.2 Tick数据与Bar数据——两种武器

做套利，你手里得有两把刷子：Tick数据和Bar数据。

Tick数据，说白了就是每一笔成交的快照。它记录了每一瞬间的买卖盘口。做高频套利，Tick数据是必需品。但它的缺点也很明显——数据量巨大，一天下来几个GB很正常。

Bar数据，就是按时间窗口聚合后的数据。比如1分钟Bar、5分钟Bar。它牺牲了细节，换来了效率。做中低频套利，Bar数据够用了。

我个人在实战中的经验是：

• 日内套利（持仓几分钟到几小时）：用Tick数据，精度越高越好
• 隔夜套利（持仓一天以上）：用1分钟Bar就够了
• 统计套利（基于日频数据）：用日线Bar

4.3 数据对齐与重采样——最容易被忽视的环节

做跨期套利，你得同时盯着两个合约。但这两个合约的数据，时间戳往往对不上。

为什么会这样？因为不同合约的成交频率不一样。近月合约可能每秒成交几十笔，远月合约可能几秒才成交一笔。如果你直接拿原始数据算价差，结果就是——近月合约已经变了十次价，远月合约还是老价格。这算出来的价差，能信吗？

所以，数据对齐是必须的。我常用的方法有两种：

1. 时间戳对齐：以某个固定时间点（比如每秒）为基准，取该时刻前后最近的有效数据
2. 事件驱动对齐：只有当两个合约都有新数据时，才计算价差

我个人更倾向于第二种。为什么？因为时间戳对齐会引入「未来数据」的风险。你取的是「前后最近」的数据，但那个「后」的数据，在实盘中你是看不到的。

重采样也是个技术活。把Tick数据转成Bar数据，看起来简单，但细节决定成败：

import pandas as pd

def resample_ticks_to_bars(ticks, freq='1min'):
    # 注意：这里要用 'right' 闭区间
    # 我踩过坑，用 'left' 会导致最后一笔数据丢失
    bars = ticks.resample(freq, label='right', closed='right').agg({
        'bid': 'ohlc',
        'ask': 'ohlc',
        'volume': 'sum'
    })
    return bars

4.4 异常值处理——别让一颗老鼠屎坏了一锅粥

数据里总会有一些「脏东西」。比如：

• 交易所的测试数据混进来了
• 网络抖动导致的数据重复或缺失
• 某个合约突然出现一个离谱的价格（比如比正常价格高10倍）

这些异常值，如果不处理，你的策略就会做出错误的判断。

我处理异常值的流程是这样的：

1. 价格合理性检查：价格必须在合理范围内。比如螺纹钢，价格不可能低于2000，也不可能高于8000
2. 价差合理性检查：两个合约的价差，通常不会超过某个阈值。如果突然出现一个巨大的价差，多半是数据有问题
3. 时间连续性检查：相邻两个Tick的时间差，不能太大也不能太小。如果时间差为0，说明数据重复了

代码实现起来也不复杂：

def clean_ticks(ticks):
    # 1. 价格范围检查
    ticks = ticks[(ticks['bid'] > 2000) & (ticks['bid'] < 8000)]
    ticks = ticks[(ticks['ask'] > 2000) & (ticks['ask'] < 8000)]
    
    # 2. 价差检查
    ticks['spread'] = ticks['ask'] - ticks['bid']
    ticks = ticks[ticks['spread'] < 100]  # 价差不超过100
    
    # 3. 去重
    ticks = ticks.drop_duplicates(subset=['exchange_ts', 'symbol'])
    
    return ticks

我的做法是：先标记异常值，而不是直接删除。然后人工复核，确认是数据问题还是真实行情。嗯，虽然麻烦了点，但安全第一。

知识体系总览

说了这么多，我们来梳理一下本章的核心逻辑。下面这张图，是我自己画的数据处理流程图，你一看就明白了：

这张图把整个数据处理流程串起来了。从交易所API获取原始数据，经过格式转换、对齐重采样、异常值清洗，最终得到可以用于策略开发的高质量数据。每一步都环环相扣，缺一不可。

好了，数据获取与清洗这部分，我就讲这么多。记住一句话：数据质量决定了你的策略能走多远。别嫌麻烦，把基础打牢了，后面的路才好走。

---