4. 数据获取与清洗：交易所API对接、历史数据下载、数据清洗与对齐、缺失值处理

做市策略回测，说白了就是拿历史数据模拟交易。数据不对，回测就是白做。我见过太多人花几个月写策略，最后发现数据里有坑，全白费了。今天咱们就把数据这块彻底讲透。

4.1 交易所API对接：从零开始拿数据

做市策略需要的数据，跟普通交易不太一样。普通交易看K线就够了，做市策略需要的是逐笔成交数据和深度快照。为什么？因为做市赚的是买卖价差，你得知道每一笔订单是怎么成交的。

我个人习惯用 ccxt 这个库，它封装了上百家交易所的API。你想想看，如果每家交易所都单独写一套对接代码，那得累死。

来看一段实际代码。我用的是币安的API，因为它的数据质量最好，而且免费额度够用。

import ccxt
import pandas as pd
from datetime import datetime

# 初始化交易所
exchange = ccxt.binance({
    'rateLimit': 1200,  # 币安的限制是1200次/分钟
    'enableRateLimit': True,  # 自动限速，别把自己封了
})

# 获取深度数据
def fetch_orderbook(symbol='BTC/USDT', limit=100):
    """获取订单簿深度"""
    try:
        ob = exchange.fetch_order_book(symbol, limit=limit)
        return {
            'bids': ob['bids'][:10],  # 取前10档买单
            'asks': ob['asks'][:10],  # 取前10档卖单
            'timestamp': ob['timestamp']
        }
    except Exception as e:
        print(f'获取深度失败: {e}')
        return None

# 获取历史逐笔成交
def fetch_trades(symbol='BTC/USDT', since=None, limit=1000):
    """获取历史成交记录"""
    trades = exchange.fetch_trades(symbol, since=since, limit=limit)
    df = pd.DataFrame(trades)
    df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
    return df

4.2 历史数据下载：别只盯着最近一个月

做市策略回测，至少需要3个月以上的数据。为什么？因为市场环境会变。牛市的做市策略在熊市可能亏得裤衩都不剩。

我建议你下载以下时间段的数据：

数据用途	推荐时间跨度	数据频率
策略开发	最近3个月	逐笔成交 + 深度快照
参数优化	最近6个月	逐笔成交
稳健性测试	最近12个月	K线 + 深度快照
极端行情测试	包含312、519等事件	全部数据

下载数据时，要注意时间对齐。不同交易所的时间戳格式不一样，有的用毫秒，有的用微秒。我习惯统一转成毫秒时间戳，方便后续处理。

def download_historical_data(symbol, start_date, end_date):
    """批量下载历史数据"""
    all_trades = []
    since = exchange.parse8601(start_date)
    end = exchange.parse8601(end_date)
    
    while since < end:
        trades = exchange.fetch_trades(symbol, since=since, limit=1000)
        if len(trades) == 0:
            break
        all_trades.extend(trades)
        since = trades[-1]['timestamp'] + 1  # 避免重复
        time.sleep(exchange.rateLimit / 1000)  # 限速
    
    df = pd.DataFrame(all_trades)
    df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
    return df

4.3 数据清洗与对齐：脏数据比没数据更可怕

数据下载下来，你以为就能直接用了吗？太天真了。交易所的数据经常有各种问题：

• 重复数据：同一笔成交被推送了两次
• 乱序数据：时间戳不是严格递增的
• 异常价格：比如BTC突然变成0.01美元
• 缺失字段：某些成交记录没有买卖方向

我在项目中遇到过最离谱的一次，币安的某只山寨币深度数据里，卖一价竟然是负数。你想想看，这要是直接拿去回测，策略会疯狂买入，因为觉得价格太便宜了。

清洗流程我总结为三步：

1. 去重：按交易ID和时间戳去重
2. 排序：按时间戳升序排列
3. 过滤异常：剔除价格、数量明显异常的数据

def clean_trades(df):
    """清洗逐笔成交数据"""
    # 1. 去重
    df = df.drop_duplicates(subset=['id', 'timestamp'])
    
    # 2. 排序
    df = df.sort_values('timestamp').reset_index(drop=True)
    
    # 3. 过滤异常价格（比如价格偏离中位数太多）
    median_price = df['price'].median()
    std_price = df['price'].std()
    df = df[(df['price'] > median_price - 5*std_price) & 
            (df['price'] < median_price + 5*std_price)]
    
    # 4. 过滤异常数量（比如数量为0或负数）
    df = df[df['amount'] > 0]
    
    return df

4.4 缺失值处理：别让数据缺口毁了你的回测

数据缺失是做市策略回测中最头疼的问题。为什么？因为做市策略是连续报价的，数据一断，你的策略就相当于在真空中交易，结果完全不可信。

常见的缺失情况有两种：

• 时间戳缺失：某段时间内完全没有数据
• 字段缺失：有数据但某些字段是NaN

处理方式取决于缺失的原因：

缺失类型	可能原因	处理方法
短时间缺失（<1分钟）	网络抖动、交易所限流	线性插值或前向填充
长时间缺失（>1小时）	交易所维护、数据源问题	标记为不可交易时段
字段缺失	数据格式不完整	从其他数据源补全

def handle_missing_data(df, freq='1s'):
    """处理缺失值"""
    # 设置时间索引
    df = df.set_index('timestamp')
    
    # 生成完整的时间序列
    full_index = pd.date_range(start=df.index.min(), 
                               end=df.index.max(), 
                               freq=freq)
    
    # 重采样并前向填充（短时间缺失）
    df = df.reindex(full_index)
    df = df.fillna(method='ffill', limit=60)  # 最多填充60个缺失值
    
    # 标记长时间缺失
    df['is_valid'] = ~df['price'].isna()
    
    return df

4.5 数据对齐：把不同频率的数据整合到一起

做市策略回测需要同时用到多种数据：深度数据（毫秒级）、逐笔成交（随机时间）、K线（分钟级）。怎么把它们对齐到同一个时间轴上？

我的做法是：

1. 以深度数据的时间戳为主轴
2. 将逐笔成交按时间戳归并到最近的深度快照上
3. K线数据作为辅助，用于计算波动率等指标

def align_data(orderbook_df, trades_df, kline_df):
    """对齐三种数据"""
    # 1. 将逐笔成交对齐到深度快照
    trades_df['aligned_ts'] = pd.cut(
        trades_df['timestamp'],
        bins=orderbook_df['timestamp'],
        labels=orderbook_df['timestamp'][:-1]
    )
    
    # 2. 合并数据
    merged = orderbook_df.merge(
        trades_df.groupby('aligned_ts').agg({
            'price': ['mean', 'std'],
            'amount': 'sum'
        }),
        left_on='timestamp',
        right_index=True,
        how='left'
    )
    
    # 3. 添加K线数据
    merged = merged.merge(
        kline_df[['close', 'volume']],
        left_on=pd.to_datetime(merged['timestamp'], unit='ms').dt.floor('1min'),
        right_index=True,
        how='left'
    )
    
    return merged

4.6 本章知识体系

说了这么多，咱们用一张图把整个数据获取与清洗的流程串起来：

这张图把整个流程串起来了。从API对接开始，到数据下载、清洗、缺失值处理、对齐，最后得到可用的回测数据。每一步都有坑，每一步都得小心。

数据这块，说白了就是「垃圾进，垃圾出」。你花再多时间优化策略，数据不对全是白搭。我见过太多人把精力花在策略上，结果数据一塌糊涂。记住：数据质量决定了回测的天花板。