第四章：数据获取与清洗——如何获取现货与期货行情数据

做跨市场套利，说白了就是吃两边的价差。但有个前提——你得先拿到干净的数据。

我见过太多人，策略逻辑写得漂亮，回测曲线也好看，一上实盘就崩。为什么？数据源就有问题。现货和期货的时间戳对不上，或者某天缺了几个 tick，回测时没发现，实盘直接亏钱。

这一章，我就把数据获取和清洗的坑，一个一个给你讲清楚。

4.1 数据源的选择——别在第一步就踩坑

我个人习惯把数据源分成三类：官方API、第三方数据商、爬虫。每类都有它的适用场景。

官方交易所API，比如上期所的CTP、中金所的飞马。优点是数据最权威，延迟最低。缺点嘛——接入门槛高，而且有流量限制。我记得有一次做股指期货套利，CTP接口突然断了，幸好我本地有备份，不然那天就白干了。

第三方数据商，像Wind、Choice、Tushare。适合做回测研究，数据已经清洗过一遍。但要注意，他们的数据有时会有几秒延迟，做高频套利的话不太够用。

爬虫，比如从东方财富、新浪财经抓数据。免费，但维护成本高。网站改个版，你的爬虫就废了。我建议只用来做辅助验证，别当主力。

4.2 用API获取行情数据

先看一个最简单的例子。用ccxt库获取比特币现货和期货数据。ccxt支持上百个交易所，接口统一，写起来很舒服。

import ccxt
import pandas as pd
from datetime import datetime

# 初始化交易所
exchange = ccxt.binance({
    'rateLimit': 1200,
    'enableRateLimit': True,
})

# 获取现货数据
spot_ohlcv = exchange.fetch_ohlcv(
    symbol='BTC/USDT',
    timeframe='1m',
    limit=100
)

# 获取期货数据
future_ohlcv = exchange.fetch_ohlcv(
    symbol='BTC/USDT:USDT',
    timeframe='1m',
    limit=100
)

# 转成DataFrame
spot_df = pd.DataFrame(
    spot_ohlcv,
    columns=['timestamp', 'open', 'high', 'low', 'close', 'volume']
)
spot_df['timestamp'] = pd.to_datetime(spot_df['timestamp'], unit='ms')

future_df = pd.DataFrame(
    future_ohlcv,
    columns=['timestamp', 'open', 'high', 'low', 'close', 'volume']
)
future_df['timestamp'] = pd.to_datetime(future_df['timestamp'], unit='ms')

print("现货数据：")
print(spot_df.head())
print("\n期货数据：")
print(future_df.head())

这里有个细节——现货和期货的symbol写法不一样。现货是BTC/USDT，期货是BTC/USDT:USDT。我刚开始做的时候，这个冒号漏写了，查了半天bug。

4.3 爬虫抓取——应急方案

有时候API拿不到历史数据，或者想补一些特殊品种的数据，就得靠爬虫。我一般用requests+BeautifulSoup，简单粗暴。

import requests
from bs4 import BeautifulSoup
import json

def fetch_spot_from_sina(symbol):
    """从新浪财经抓取现货行情"""
    url = f'http://hq.sinajs.cn/list={symbol}'
    headers = {
        'Referer': 'http://finance.sina.com.cn',
        'User-Agent': 'Mozilla/5.0'
    }
    resp = requests.get(url, headers=headers)
    # 解析返回的数据
    data = resp.text.split('=')[1].strip('"')
    fields = data.split(',')
    return {
        'name': fields[0],
        'open': float(fields[1]),
        'close': float(fields[3]),
        'high': float(fields[4]),
        'low': float(fields[5]),
        'volume': float(fields[8])
    }

# 抓取螺纹钢现货
spot_data = fetch_spot_from_sina('rb9999')
print(spot_data)

4.4 数据对齐——套利的核心难点

数据拿到手了，但现货和期货的时间戳对不上。为什么？因为现货是24小时交易，期货有夜盘，而且不同交易所的撮合频率不一样。

我常用的对齐方法有两种：

• 时间戳对齐：把两个数据都按整分钟/整小时对齐。比如把09:30:15的数据归到09:30:00。
• 频率对齐：如果现货是1分钟K线，期货是5分钟K线，那就把现货重采样成5分钟。

# 时间戳对齐示例
def align_timestamps(spot_df, future_df, freq='1min'):
    """
    将现货和期货数据按指定频率对齐
    """
    # 设置时间戳为索引
    spot_df.set_index('timestamp', inplace=True)
    future_df.set_index('timestamp', inplace=True)
    
    # 重采样到统一频率
    spot_resampled = spot_df.resample(freq).last()
    future_resampled = future_df.resample(freq).last()
    
    # 合并两个DataFrame
    aligned = pd.DataFrame({
        'spot_close': spot_resampled['close'],
        'future_close': future_resampled['close']
    })
    
    # 删除两边都为NaN的行
    aligned.dropna(how='all', inplace=True)
    
    return aligned

# 使用示例
aligned_data = align_timestamps(spot_df, future_df, freq='1min')
print(aligned_data.head())

嗯，这里要注意。重采样的时候，用last()还是mean()？我个人习惯用last()，因为套利策略看的是收盘价。但如果你做的是高频策略，可能要用ohlc()保留更多信息。

4.5 缺失值处理——别让NaN毁了你的策略

数据对齐之后，肯定会有缺失值。比如期货夜盘交易到凌晨1点，现货可能只到下午3点。那晚上8点到凌晨1点的现货数据就是空的。

怎么处理？我一般按优先级排序：

1. 前向填充：用上一个非空值填充。适合处理非交易时段。
2. 线性插值：用前后两个值的平均值。适合处理短暂的数据缺失。
3. 直接删除：如果缺失太多，干脆删掉那段时间。

# 缺失值处理示例
def handle_missing_values(df, method='ffill'):
    """
    处理缺失值
    method: 'ffill' 前向填充
            'linear' 线性插值
            'drop' 删除缺失行
    """
    df_clean = df.copy()
    
    if method == 'ffill':
        df_clean.fillna(method='ffill', inplace=True)
        # 如果开头还有NaN，用后面的值回填
        df_clean.fillna(method='bfill', inplace=True)
    elif method == 'linear':
        df_clean.interpolate(method='linear', inplace=True)
    elif method == 'drop':
        df_clean.dropna(inplace=True)
    
    return df_clean

# 看看不同方法的效果
print("原始数据缺失情况：")
print(aligned_data.isnull().sum())

print("\n前向填充后：")
print(handle_missing_values(aligned_data, 'ffill').head())

print("\n线性插值后：")
print(handle_missing_values(aligned_data, 'linear').head())

4.6 数据存储——别每次都重新爬

数据清洗完了，存起来。下次直接用，省得再跑一遍。

我推荐用HDF5格式。读写快，支持压缩，还能存多个表。当然，小数据量用CSV也行。

# 存储清洗后的数据
import h5py

# 保存到HDF5
aligned_data.to_hdf(
    'btc_arbitrage_data.h5',
    key='aligned_1min',
    mode='w',
    complevel=9  # 压缩级别
)

# 读取
loaded_data = pd.read_hdf('btc_arbitrage_data.h5', key='aligned_1min')
print(f"加载数据量：{len(loaded_data)} 行")

我个人习惯每天收盘后跑一次数据更新脚本，把当天的数据追加到HDF5文件里。这样回测时随时有最新的数据可用。

4.7 实战中的几个小技巧

• 检查数据完整性：每天跑完数据后，对比一下现货和期货的行数。差太多肯定有问题。
• 记录数据来源：在DataFrame里加一列source，标记数据是从API还是爬虫来的。方便排查问题。
• 保留原始数据：清洗前先备份。万一洗错了，还能恢复。
• 注意节假日：国内期货有春节、国庆长假，现货市场可能不休市。这段时间的价差数据要特别处理。

---

公众号:蓝海数据掘金营，微信deep3321