数据源与获取：沪深交易所Level-2行情接口、Tick级数据订阅、历史逐笔委托数据回放

做逐笔委托流分析，第一步就是搞定数据。没有数据，再牛的策略也是空中楼阁。

我刚开始接触Level-2行情时，其实挺懵的。市面上各种接口、各种协议，到底该选哪个？数据到了手里，又该怎么用？今天咱们就把这事彻底聊透。

Level-2行情接口：到底比Level-1强在哪？

普通投资者用的Level-1行情，是3秒快照。说白了，你看到的盘口是3秒前的样子。做短线交易，这3秒的延迟足够让你吃大亏。

Level-2行情就不一样了。它提供的是逐笔成交和逐笔委托的原始数据流。我个人的习惯是，只要做高频相关的策略，必须上Level-2。

对比项	Level-1行情	Level-2行情
刷新频率	3秒一次快照	实时逐笔推送
委托队列	仅显示前5档	显示前50档（深交所）或全量（上交所）
逐笔成交	无	有，包含每笔成交的买卖方向
逐笔委托	无	有，能看到每一笔挂单的撤单、成交
数据用途	趋势判断	主力资金追踪、订单簿重构

你想想看，如果你能看到每一笔委托的来龙去脉，是不是就能判断出主力是在吸筹还是在出货？这就是Level-2的价值所在。

Tick级数据订阅：实时流的接入方式

实时数据订阅，通常有两种方式：WebSocket和TCP直连。

WebSocket方式：适合个人开发者。门槛低，用Python的websocket库就能搞定。我早期做回测系统时，就是用WebSocket接的某券商的模拟盘。

TCP直连方式：适合机构。延迟更低，但需要自己维护心跳、重连、数据校验。嗯，这里要注意，TCP直连的协议通常是二进制格式，你得自己解析。

下面是一个简单的WebSocket订阅示例，订阅某只股票的逐笔委托流：

import websocket
import json

def on_message(ws, message):
    data = json.loads(message)
    # 解析逐笔委托数据
    if data['type'] == 'tick_order':
        print(f"时间: {data['time']}, 价格: {data['price']}, 数量: {data['volume']}, 方向: {data['direction']}")

def on_error(ws, error):
    print(f"连接异常: {error}")

def on_close(ws, close_status_code, close_msg):
    print("连接已关闭")

def on_open(ws):
    # 订阅股票代码
    subscribe_msg = {
        "action": "subscribe",
        "symbol": "600519",
        "channel": "tick_order"
    }
    ws.send(json.dumps(subscribe_msg))

if __name__ == "__main__":
    ws = websocket.WebSocketApp("wss://your-broker-api.com/level2",
                                on_open=on_open,
                                on_message=on_message,
                                on_error=on_error,
                                on_close=on_close)
    ws.run_forever()

这段代码看着简单，但实际跑起来坑不少。我曾经遇到过WebSocket莫名其妙断开，结果漏掉了关键数据，导致策略信号出错。后来我加了个心跳检测和自动重连机制，才算稳定下来。

历史逐笔委托数据回放：模拟真实战场

实时数据用来做交易，历史数据用来做回测。两者缺一不可。

历史逐笔委托数据，一般以CSV或Parquet格式存储。我推荐用Parquet，压缩率高，读取速度快。尤其是处理上亿条数据时，CSV能把你等哭。

数据回放的核心逻辑，就是按时间顺序逐笔推送数据，模拟真实的交易环境。下面是一个简单的回放框架：

import pandas as pd
import time

class TickReplayer:
    def __init__(self, data_path):
        self.data = pd.read_parquet(data_path)
        self.data = self.data.sort_values('time')  # 按时间排序
        self.index = 0
    
    def replay(self, speed=1.0):
        """回放逐笔委托数据"""
        while self.index < len(self.data):
            tick = self.data.iloc[self.index]
            # 模拟真实时间间隔
            if self.index > 0:
                prev_time = self.data.iloc[self.index - 1]['time']
                time_diff = (tick['time'] - prev_time) / speed
                time.sleep(time_diff)
            
            # 处理当前tick
            self.process_tick(tick)
            self.index += 1
    
    def process_tick(self, tick):
        """处理单笔委托数据"""
        print(f"回放: {tick['time']} | 价格: {tick['price']} | 数量: {tick['volume']} | 方向: {tick['direction']}")

# 使用示例
replayer = TickReplayer('data/2024_01_tick.parquet')
replayer.replay(speed=10.0)  # 10倍速回放

这个回放器虽然简陋，但核心思想都在了。实际项目中，你还需要考虑数据校验、异常处理、性能优化等问题。

数据获取的整体架构

说了这么多，咱们用一张图来总结一下整个数据获取的流程：

这张图把整个数据获取链路串起来了。从交易所源头，到接入方式，再到数据解析、存储，最后到实时交易和历史回放。每一步都有坑，每一步也都有优化空间。

好了，数据源和获取这块就聊到这。记住一句话：数据质量决定策略上限。花时间把数据链路搞通，后面的分析才能事半功倍。