第三章：数据获取与清洗——交易所API对接、Level2行情数据解析、数据清洗与对齐、数据存储方案

做市商回测，说白了就是拿历史数据模拟真实交易。但有个问题——你拿到的数据，跟交易所里实际发生的数据，是不是一回事？

我刚开始做回测框架时，就踩过这个坑。从某数据商买来的1分钟K线，回测跑得漂漂亮亮，实盘一上线直接亏麻。后来才发现，人家给的K线是「复权」过的，而我做市商策略需要的是原始逐笔数据。嗯，从那以后，我对数据获取和清洗这件事，再也不敢马虎。

核心观点：做市商回测的数据质量，直接决定策略的生死。Level2行情是标配，逐笔成交和十档盘口缺一不可。

3.1 交易所API对接：选对路子，少走弯路

国内交易所的API，大体分两类：

  • REST API：请求-响应模式，适合拉取历史数据、查询账户信息。延迟高，但稳定。
  • WebSocket API：全双工通信，适合接收实时行情。延迟低，但容易断线重连。

我个人习惯的做法是：历史数据用REST，实时数据用WebSocket。别混着用，否则代码会变得很难维护。

举个例子，对接上交所的L1行情（其实现在L2也开放了），代码大概长这样：

import requests
import json

# 上交所行情API示例（伪代码）
def fetch_sse_level2(symbol):
    url = f"https://api.sse.com.cn/level2/{symbol}"
    headers = {
        "User-Agent": "Mozilla/5.0",
        "Authorization": "Bearer YOUR_TOKEN"
    }
    resp = requests.get(url, headers=headers)
    if resp.status_code == 200:
        return resp.json()
    else:
        raise Exception(f"API请求失败: {resp.status_code}")

小技巧：很多交易所的API有频率限制。我一般会在请求之间加个time.sleep(0.1)，或者用异步请求库aiohttp来并发控制。别一股脑全发出去，容易被封IP。

3.2 Level2行情数据解析：逐笔数据才是宝藏

Level2行情跟Level1最大的区别在哪？

Level1给你的是「切片」——每3秒一张快照。Level2给你的是「流水」——每一笔成交、每一笔委托都清清楚楚。

做市商策略，需要的就是这种颗粒度。你想想看，如果你只知道每分钟的开高低收，你怎么判断中间有没有人砸盘？怎么捕捉到那0.1秒的流动性缺口？

Level2的核心数据包括：

  • 逐笔成交：每一笔交易的成交价、成交量、成交方向（主动买/主动卖）
  • 十档行情：买卖各10档的委托价格和委托量
  • 逐笔委托：每一笔挂单的详细信息（这个数据量最大，也最有价值）

解析的时候，要注意字节对齐。我记得有一次，深交所的Level2数据格式调整，把某个字段从4字节改成了8字节，我这边没及时更新解析代码，结果所有数据都错位了。排查了整整两天……

# Level2逐笔成交解析示例（深交所格式）
import struct

def parse_szse_trade(raw_bytes):
    # 假设每条逐笔成交记录固定长度48字节
    trade = {}
    trade['timestamp'] = struct.unpack('>Q', raw_bytes[0:8])[0]  # 8字节时间戳
    trade['price'] = struct.unpack('>I', raw_bytes[8:12])[0] / 10000.0  # 4字节价格
    trade['volume'] = struct.unpack('>I', raw_bytes[12:16])[0]  # 4字节数量
    trade['direction'] = 'B' if raw_bytes[16] == 0x01 else 'S'  # 1字节方向
    # ... 其他字段
    return trade

注意：不同交易所的Level2数据格式完全不同。上交所用二进制，深交所也是二进制但字段顺序不同。中金所的期货数据又是另一套。千万别想一套代码通吃所有市场。

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

数据拿到手，直接扔进回测引擎？别急，先洗一洗。

我见过最离谱的情况：某天的数据里，某只股票在14:59:59突然出现一笔成交价是0.01元，然后下一秒恢复正常。这明显是数据错误。如果不处理，回测里你的策略会以为捡到了大便宜，疯狂买入……然后实盘就悲剧了。

清洗的常规步骤：

  1. 去重：同一个时间戳、同一笔成交，只保留一条。用pandas的drop_duplicates就行。
  2. 去异常值：价格超出涨跌停板、数量为0或负数、时间戳不在交易时段内——统统干掉。
  3. 补缺失：某些时间段可能没有成交（比如开盘前集合竞价阶段），需要根据业务逻辑填充或插值。
  4. 对齐时间戳：不同数据源的时间戳精度可能不同（有的精确到毫秒，有的到微秒），要统一成同一个精度。

对齐这一步，我吃过亏。当时把逐笔成交数据和十档行情数据合并，发现时间戳差了3毫秒。别小看这3毫秒，在高频场景下，足以让策略做出完全相反的判断。后来我强制把所有时间戳都对齐到最近的100微秒，才算解决。

import pandas as pd

def align_timestamps(df, freq='100us'):
    """
    将时间戳对齐到指定频率
    freq: '100us' 表示100微秒
    """
    df['timestamp'] = df['timestamp'].dt.floor(freq)
    return df

# 清洗示例
def clean_trade_data(df):
    # 去重
    df = df.drop_duplicates(subset=['timestamp', 'price', 'volume'])
    # 去异常价格（假设股票价格在1-1000元之间）
    df = df[(df['price'] >= 1) & (df['price'] <= 1000)]
    # 去异常数量
    df = df[df['volume'] > 0]
    return df

3.4 数据存储方案：选对工具，事半功倍

数据清洗完了，存哪？

这个问题，我纠结了很久。最开始用CSV，简单粗暴。但数据量一上来，一个月的逐笔数据就几十个G，CSV根本扛不住。后来试过MySQL，查询慢得让人抓狂。最后锁定了两个方案：

存储方案 适用场景 优点 缺点
Parquet + 分区 历史数据存储、批量分析 压缩率高、查询快、支持列式存储 不适合频繁更新
InfluxDB / ClickHouse 时序数据、实时查询 写入快、聚合查询强 部署复杂、学习成本高

我个人习惯的做法是：历史数据用Parquet按日期分区存储，实时数据用InfluxDB。回测时从Parquet读，监控时从InfluxDB读，各司其职。

# 使用Parquet存储逐笔数据
import pandas as pd

def save_trade_data(df, date_str):
    # 按日期分区存储
    path = f"/data/level2/trades/{date_str}/"
    df.to_parquet(path, partition_cols=['symbol'], compression='snappy')

def load_trade_data(date_str, symbol):
    path = f"/data/level2/trades/{date_str}/"
    df = pd.read_parquet(path, filters=[('symbol', '=', symbol)])
    return df

避坑指南：我曾经把数据一股脑全塞进一个Parquet文件，结果查询某只股票的数据时，要扫描整个文件，慢得要命。后来加了分区和过滤条件，查询速度提升了100倍。记住：分区键选对，性能翻倍。

3.5 本章知识体系总览

说了这么多，我画了一张图，帮你把整个数据获取与清洗的流程串起来：

数据获取与清洗流程 交易所API 数据供应商 历史数据文件 Level2行情解析（逐笔成交、十档盘口、逐笔委托） 数据清洗（去重 → 去异常 → 补缺失 → 对齐时间戳） Parquet（历史数据） InfluxDB（实时数据） 最终输出：干净、对齐、可回测的做市商数据

这张图里，从数据源到最终存储，每一步都有坑。但只要你按照这个流程走下来，数据质量基本就有保障了。

最后说一句：数据是回测的基石。你花80%的时间在数据上，都不为过。策略模型再漂亮，数据是脏的，一切都是白搭。

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

← 上一章 📖 返回目录 下一章 →