第3章数据获取与清洗

3. 数据获取与清洗：多源数据API调用、时间对齐、缺失值处理

做跨市场套利，说白了就是跟数据打交道。我见过太多人策略写得漂亮，结果死在数据上——时间戳对不上、交易所断流、某根K线突然跳空……这些坑我全踩过。今天咱们就把数据获取和清洗这层窗户纸捅破。

3.1 多源数据API调用：别让接口成为瓶颈

跨市场套利至少涉及两个交易所。比如你做BTC期现套利，得同时拉现货和期货数据。每个交易所的API风格都不一样，有的用REST，有的用WebSocket，有的限频严格得像防贼。

我个人习惯用 ccxt 这个库，它把主流交易所的API统一了。你传个交易所名字，它自动帮你处理签名、限频、重连。省心。

import ccxt
import pandas as pd
from datetime import datetime

# 初始化交易所
binance = ccxt.binance({
    'apiKey': 'YOUR_API_KEY',
    'secret': 'YOUR_SECRET',
    'enableRateLimit': True,  # 自动限频，别关
    'options': {'defaultType': 'spot'}  # 现货
})

okex = ccxt.okex({
    'apiKey': 'YOUR_API_KEY',
    'secret': 'YOUR_SECRET',
    'enableRateLimit': True,
    'options': {'defaultType': 'swap'}  # 永续合约
})

# 拉取1分钟K线
def fetch_ohlcv(exchange, symbol, timeframe='1m', limit=1000):
    try:
        data = exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
        df = pd.DataFrame(data, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
        df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
        df.set_index('timestamp', inplace=True)
        return df
    except Exception as e:
        print(f"拉取失败: {e}")
        return None

# 同时拉取现货和合约数据
spot_df = fetch_ohlcv(binance, 'BTC/USDT')
swap_df = fetch_ohlcv(okex, 'BTC/USD-SWAP')

print(spot_df.head())
print(swap_df.head())

⚠️ 注意： 限频设置 enableRateLimit=True 必须开。我曾经在实盘时忘了开，结果API被交易所封了5分钟，那5分钟价差波动剧烈，直接亏了2个点。别学我。

这里有个细节：不同交易所的K线时间戳可能差几毫秒。比如Binance的K线是整点对齐，OKEx的可能是整点+1秒。你直接合并会出问题。怎么办？往下看。

3.2 时间对齐：把不同时钟拧到一根轴上

时间对齐是跨市场套利最恶心的一步。没有之一。

为什么？因为交易所的服务器时间不同，网络延迟不同，K线生成逻辑也不同。你拿到的两个时间序列，就像两个不同步的钟。直接算价差，全是噪音。

我常用的方法是 重采样 + 向前填充。说白了，就是选一个基准时间轴，把其他数据都对齐过来。

# 假设 spot_df 和 swap_df 已经拉取完毕
# 合并两个DataFrame
combined = pd.DataFrame({
    'spot_close': spot_df['close'],
    'swap_close': swap_df['close']
})

# 重采样到1分钟，向前填充
combined_resampled = combined.resample('1T').ffill()

# 删除重采样后产生的NaN（开头几行可能没有数据）
combined_resampled.dropna(inplace=True)

# 计算价差
combined_resampled['spread'] = combined_resampled['swap_close'] - combined_resampled['spot_close']

print(combined_resampled.head())

💡 小技巧： 如果你做高频套利（比如秒级），别用重采样。直接用原始时间戳做 pd.merge_asof，它能按最近时间匹配。我做过一个项目，用 merge_asof 把延迟从50ms降到了5ms。

你想想看，如果两个交易所的K线时间戳差了2秒，你算出来的价差可能完全是假的。所以时间对齐不是可选项，是必选项。

3.3 缺失值处理：数据断流了怎么办？

数据缺失在实盘中太常见了。交易所维护、网络抖动、API限频……随便一个原因就能让你的DataFrame出现NaN。

处理缺失值，我分三种情况：

零星缺失（比如连续1-2根K线）：用前向填充 ffill()，或者线性插值 interpolate()。我个人偏爱线性插值，因为它更平滑。
连续缺失（比如5分钟以上）：别填充了，直接删掉。填充出来的数据是假的，套利信号会失真。
开盘/收盘附近缺失：这是最危险的。比如某交易所9:30开盘，但数据9:32才到。这时候的价差计算会爆炸。我建议用 pd.between_time 把开盘前5分钟的数据全部剔除。

# 示例：处理缺失值
def clean_missing(df, max_gap=3):
    """
    max_gap: 允许连续缺失的最大K线数
    """
    # 先标记连续缺失
    df['gap'] = df['close'].isna().astype(int).groupby(df['close'].notna().cumsum()).cumsum()
    
    # 连续缺失超过max_gap的行，直接删除
    df_clean = df[df['gap'] <= max_gap].copy()
    
    # 对剩余缺失值做线性插值
    df_clean['close'] = df_clean['close'].interpolate(method='linear')
    
    # 删除辅助列
    df_clean.drop('gap', axis=1, inplace=True)
    
    return df_clean

# 应用清洗
spot_clean = clean_missing(spot_df, max_gap=3)
swap_clean = clean_missing(swap_df, max_gap=3)

  🔑 核心原则： 宁可少数据，不要假数据。缺失值填充得再漂亮，也是你猜的。套利交易赚的是确定性，不是猜谜。

3.4 知识体系总览

下面这张图是我自己总结的数据获取与清洗流程。每次做新项目，我都先对着这张图检查一遍，省得漏步骤。

3.5 实战中的避坑指南

讲几个我踩过的坑，你遇到了直接绕开：

API返回的数据顺序可能乱。有些交易所的K线接口不保证按时间升序返回。我建议每次拉完数据都 df.sort_index() 一下，别偷懒。
注意交易对命名差异。Binance上叫 BTC/USDT，OKEx上可能叫 BTC-USD-SWAP。你写代码时得做个映射表，不然拉错数据都不知道。
WebSocket断线重连。如果用实时数据，一定要写重连逻辑。我见过有人WebSocket断了3小时没发现，策略还在跑，结果亏得底朝天。
数据存储别用CSV。CSV读写慢，而且不支持时间索引。用Parquet或者HDF5，速度能快10倍。

📌 我的习惯： 每次拉完数据，先画个图看看。用 plt.plot(spot_df.index, spot_df['close']) 扫一眼，有没有跳空、有没有断点，一目了然。别信数据，信眼睛。

好了，数据获取和清洗就这些。记住：数据干净了，策略才有意义。下一章咱们聊价差特征工程，到时候你会感谢今天认真清洗数据的自己。

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

📚 课程目录

3. 数据获取与清洗：多源数据API调用、时间对齐、缺失值处理

3.1 多源数据API调用：别让接口成为瓶颈

3.2 时间对齐：把不同时钟拧到一根轴上

3.3 缺失值处理：数据断流了怎么办？

3.4 知识体系总览

3.5 实战中的避坑指南