第一章 数据获取与清洗：交易所API对接与数据对齐

做期现套利，说白了就是吃价差。但价差这东西，转瞬即逝。你想想看，如果数据都拿不准，策略再牛也是白搭。我个人习惯，把数据获取和清洗看作整个套利系统的地基——地基不稳，楼盖得再高也得塌。

这一章，咱们就聊聊怎么从交易所拿到干净、对齐的数据。嗯，这里要注意，我说的“干净”不是指数据本身有没有病毒，而是指它能不能直接用。

1.1 交易所API对接：Binance和OKX

目前主流交易所，Binance和OKX的API文档写得还算清楚。我刚开始对接时，踩过不少坑。比如，Binance的REST接口有频率限制，你请求太快会被封IP。OKX的WebSocket订阅格式又跟Binance不太一样。

先看一个最简单的REST请求示例，获取现货最新价格：

import requests
import json

# Binance 现货价格
def get_binance_spot_price(symbol='BTCUSDT'):
    url = f'https://api.binance.com/api/v3/ticker/price?symbol={symbol}'
    resp = requests.get(url)
    data = resp.json()
    return float(data['price'])

# OKX 现货价格
def get_okx_spot_price(symbol='BTC-USDT'):
    url = f'https://www.okx.com/api/v5/market/ticker?instId={symbol}'
    resp = requests.get(url)
    data = resp.json()
    return float(data['data'][0]['last'])

print(f'Binance BTC: {get_binance_spot_price()}')
print(f'OKX BTC: {get_okx_spot_price()}')

期货数据稍微复杂点。因为期货有不同交割月份，比如当季、次季。你需要指定合约代码。Binance的永续合约代码是 BTCUSDT 后面加个 PERP，OKX则是 BTC-USDT-SWAP。

# Binance 永续合约价格
def get_binance_futures_price(symbol='BTCUSDT'):
    url = f'https://fapi.binance.com/fapi/v1/ticker/price?symbol={symbol}'
    resp = requests.get(url)
    data = resp.json()
    return float(data['price'])

# OKX 永续合约价格
def get_okx_futures_price(symbol='BTC-USDT-SWAP'):
    url = f'https://www.okx.com/api/v5/market/ticker?instId={symbol}'
    resp = requests.get(url)
    data = resp.json()
    return float(data['data'][0]['last'])

1.2 期货与现货数据对齐

数据拿回来了，但有个问题：现货和期货的报价时间戳可能不一致。比如，现货数据是100ms更新一次，期货可能是200ms。你直接拿两个时间点的价格算价差，结果肯定不准。

怎么做对齐？我个人习惯用“最近邻匹配”。说白了，就是找到时间上最接近的那一对数据。

import pandas as pd

# 假设有两个DataFrame：spot_df 和 futures_df
# 都包含 'timestamp' 和 'price' 两列

def align_data(spot_df, futures_df):
    # 将时间戳设为索引
    spot_df = spot_df.set_index('timestamp')
    futures_df = futures_df.set_index('timestamp')
    
    # 合并，使用最近邻方法
    merged = pd.merge_asof(
        spot_df.sort_index(),
        futures_df.sort_index(),
        left_index=True,
        right_index=True,
        direction='nearest',
        tolerance=pd.Timedelta('100ms')  # 最多容忍100ms偏差
    )
    return merged

aligned = align_data(spot_df, futures_df)
print(aligned.head())

1.3 缺失值处理

交易所偶尔会抽风，比如网络抖动导致某条数据没返回。这时候数据里就会出现NaN。你不能直接跳过，因为策略需要连续的数据流。

处理缺失值，我常用的方法有三种：

• 前向填充（ffill）： 用上一个有效值填充。适合价格这种变化不大的数据。
• 线性插值： 用前后两个值的平均值。适合成交量这类数据。
• 直接删除： 如果缺失太多，干脆扔掉这一行。但要注意，别把连续数据搞断了。

# 前向填充
df['price'].fillna(method='ffill', inplace=True)

# 线性插值
df['volume'].interpolate(method='linear', inplace=True)

# 删除缺失超过50%的行
df.dropna(thresh=len(df.columns) * 0.5, inplace=True)

1.4 时间戳统一

不同交易所的时间戳格式可能不同。Binance用的是毫秒级Unix时间戳，OKX用的是毫秒级，但有些接口返回的是字符串格式。你必须统一成一种格式，否则对齐时会出问题。

我建议统一用 datetime 对象，或者毫秒级整数。看个人习惯。我自己喜欢用毫秒级整数，因为做数学运算快。

import pandas as pd

# 假设从OKX拿到的是字符串时间戳 '2025-01-15T10:30:00.000Z'
def parse_okx_timestamp(ts_str):
    # 去掉末尾的Z，然后转成datetime
    dt = pd.to_datetime(ts_str.replace('Z', ''))
    # 转成毫秒级时间戳
    return int(dt.timestamp() * 1000)

# 假设从Binance拿到的是毫秒级整数 1736922600000
# 直接使用即可

# 统一成毫秒级整数后，再对齐
spot_df['timestamp_ms'] = spot_df['timestamp'].apply(parse_okx_timestamp)
futures_df['timestamp_ms'] = futures_df['timestamp']  # 假设已经是毫秒级

知识体系总览

下面这张图，是我自己总结的数据获取与清洗流程。你可以把它当作一个检查清单：

嗯，到这里，数据获取与清洗的核心内容就讲完了。你可能会问，为什么花这么大篇幅讲这些基础工作？因为我在实盘中吃过太多亏了。数据不对，策略再漂亮也是纸上谈兵。记住一句话：垃圾进，垃圾出。

下一章，咱们会聊价差计算与统计特征。但先把今天的内容消化掉，把代码跑通。有问题随时翻回来看这张流程图。

---