第三章：委托单流数据获取——从Level-2行情到可用数据
做量化交易的朋友都知道，数据是命根子。但说实话，很多人第一步就栽了——拿到的委托单数据根本不能用。我刚开始做高频策略时，就吃过这个亏。今天咱们聊聊怎么用Python把Level-2行情数据拿到手，洗干净，存起来。
3.1 Level-2行情到底多了什么？
普通行情（Level-1）每3秒推送一次，只有五档买卖盘。Level-2呢？
- 逐笔成交：每一笔真实成交，精确到毫秒
- 十档行情：买卖各10档，不是5档
- 委托队列：前50笔挂单的明细
- 逐笔委托：每一笔撤单、挂单的记录
嗯，这里要注意：不同券商的Level-2数据格式不一样。我遇到过最坑的一次，某券商把委托队列的买卖方向搞反了，害我回测亏了三天才发现。
核心结论：Level-2的核心价值在于「逐笔」数据。没有逐笔，你看到的都是聚合后的结果，很多微观信号就丢了。
3.2 数据获取：两种主流方案
我个人习惯用两种方式获取Level-2数据。你想想看，选哪种取决于你的策略频率和预算。
方案一：通过券商API直连
适合做高频的团队。延迟低，但门槛高。
# 以某券商API为例（伪代码）
from broker_api import Level2Client
client = Level2Client(api_key='your_key', secret='your_secret')
# 订阅逐笔成交
def on_tick(data):
print(f"时间:{data['time']}, 价格:{data['price']}, 量:{data['volume']}")
client.subscribe('600519.SH', 'tick', on_tick)
client.start()
我的经验：API直连的坑在于网络抖动。我曾经因为机房网络波动，丢了3秒的逐笔数据，回测结果直接偏了5%。建议做本地缓存+断线重连机制。
方案二：通过数据服务商购买
适合大多数个人交易者。比如万得、聚宽、Tushare等。延迟稍高，但省心。
# 以Tushare为例
import tushare as ts
pro = ts.pro_api('your_token')
# 获取某只股票的逐笔成交
df = pro.tick(
ts_code='600519.SH',
trade_date='20240115',
start_time='093000',
end_time='150000'
)
print(df.head())
避坑指南：我曾经用某免费数据源做回测，发现它的逐笔数据是「模拟」的——把3秒快照拆成了逐笔。这种数据做策略，结果就是纸上富贵。一定要确认数据是真实逐笔还是合成逐笔。
3.3 数据清洗：脏数据比没数据更可怕
拿到原始数据后，别急着跑策略。先洗一洗。我总结了三步走：
第一步：去重
Level-2数据经常有重复推送。尤其是网络重连后，会补发历史数据。
# 去重逻辑
df = df.drop_duplicates(subset=['time', 'price', 'volume'])
print(f"去重后剩余 {len(df)} 条")
第二步：异常值过滤
价格突变、成交量异常大，这些很可能是数据错误。
# 价格异常过滤
price_mean = df['price'].mean()
price_std = df['price'].std()
df = df[(df['price'] > price_mean - 3*price_std) &
(df['price'] < price_mean + 3*price_std)]
注意：3倍标准差过滤对高频数据可能太激进。我一般用中位数绝对偏差（MAD），对极端值更鲁棒。
第三步：时间对齐
不同数据源的时间戳精度不一样。有的精确到毫秒，有的只到秒。必须统一。
# 统一时间格式
df['time'] = pd.to_datetime(df['time'], unit='ms')
df = df.set_index('time').resample('1S').agg({
'price': 'last',
'volume': 'sum'
})
3.4 数据存储：CSV还是数据库？
这个问题我纠结了很久。直接说结论：
| 存储方案 | 优点 | 缺点 | 适用场景 |
|---|---|---|---|
| CSV | 简单、可读性强、方便分享 | 查询慢、不支持并发、占用空间大 | 小规模回测、个人研究 |
| SQLite | 轻量、支持SQL查询、单文件 | 并发写入差、不适合大规模 | 中等规模、单机策略 |
| ClickHouse | 列式存储、查询极快、支持高并发 | 部署复杂、需要运维 | 生产环境、高频策略 |
我个人习惯：研究阶段用CSV，上线后用ClickHouse。为什么？CSV方便我肉眼检查数据质量，ClickHouse能支撑每秒几万条的写入。
# CSV存储示例
df.to_csv('level2_600519_20240115.csv', index=False)
# SQLite存储示例
import sqlite3
conn = sqlite3.connect('level2.db')
df.to_sql('tick_data', conn, if_exists='append', index=False)
conn.close()
小技巧：CSV文件建议按日期分片存储。比如每天一个文件，文件名带日期。这样查询某一天的数据时，不用加载全部。
3.5 知识体系总览
下面这张图是我自己画的，把整个数据获取流程串起来了。你保存下来，以后做数据管道时对照着看。
3.6 避坑总结
最后，把我踩过的坑列出来，你少走弯路：
- 数据源校验：拿到数据后，先随机抽几条，跟行情软件对比。我吃过亏，某数据商把买一卖一搞反了。
- 时间戳精度：不同交易所精度不同。上交所精确到毫秒，深交所精确到微秒。合并数据时一定要对齐。
- 存储性能：CSV写入速度慢，一天的数据可能几百万行。建议用二进制格式（如Parquet）替代CSV。
- 数据完整性：网络波动会导致数据缺失。我习惯在采集端做心跳检测，超过1秒没数据就报警。
一句话总结：数据获取不是简单的「调API、存文件」。从源头校验到清洗再到存储，每一步都可能埋坑。你想想看，如果数据本身是错的，再牛逼的策略也是白搭。
公众号:蓝海数据掘金营，微信deep3321