数据接口与获取：主流Level2数据源对比

做短线交易，数据就是你的眼睛。Level2行情比普通行情快那么零点几秒，但就是这零点几秒，决定了你是吃肉还是喝汤。今天咱们聊聊数据源怎么选、怎么接、怎么存。

主流Level2数据源：券商 vs 第三方

我个人习惯把数据源分成两类：券商自带的，和第三方提供的。各有各的脾气。

对比维度	券商Level2	第三方Level2
数据延迟	约200-500ms	约50-200ms
接口稳定性	高（券商自营）	中（依赖第三方服务器）
接入成本	低（开户即可）	高（年费几千到几万）
数据完整性	完整（交易所直连）	可能缺字段
编程友好度	差（多为C++/C#）	好（Python SDK）

嗯，这里要注意。券商的数据虽然稳定，但接口通常很「原始」。我最早用某券商的C++接口，光编译环境就折腾了两天。第三方数据源虽然贵，但人家把脏活累活都干了，你直接pip install就能用。

Python接口对接实战

咱们直接上代码。以某第三方数据源为例，它的Python接口大概是这样的：

import level2_api as l2

# 初始化连接
client = l2.Client(
    api_key="your_api_key",
    secret_key="your_secret",
    data_type="level2"
)

# 订阅个股行情
client.subscribe(["600519.SH", "000001.SZ"])

# 回调函数处理实时数据
def on_tick(data):
    print(f"时间: {data['time']}")
    print(f"最新价: {data['last_price']}")
    print(f"买一档: {data['bid_prices'][0]} x {data['bid_volumes'][0]}")
    print(f"卖一档: {data['ask_prices'][0]} x {data['ask_volumes'][0]}")
    # 这里可以加你的策略逻辑

client.on_tick = on_tick
client.start()

这段代码看着简单，但坑不少。我曾经在项目中遇到过一个问题：回调函数里处理数据太慢，导致数据堆积，内存爆了。后来加了队列缓冲才解决。

再说说券商接口。券商的Level2接口通常走TCP长连接，数据是二进制流。你需要自己解析协议。举个例子：

import socket
import struct

# 连接券商行情服务器
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect(("192.168.1.100", 8888))

# 发送订阅请求
subscribe_packet = build_subscribe_packet(["600519"])
sock.send(subscribe_packet)

# 循环接收数据
while True:
    header = sock.recv(16)  # 固定头部16字节
    body_len = struct.unpack("I", header[12:16])[0]
    body = sock.recv(body_len)
    tick = parse_level2_tick(body)
    print(tick)

你看，券商接口就是这种风格。你得自己拼包、拆包、处理粘包。说白了，就是回到了C语言时代。但好处是延迟确实低，毕竟少了一层转发。

数据缓存与存储

数据拿到手了，怎么存？这是个大学问。短线交易的数据量有多大？你想想看，一只股票一天产生约4000笔Tick数据，全市场4000多只股票，一天就是1600万条记录。这还不算盘口快照。

我建议分三层存储：

1. 内存缓存：存最近1分钟的Tick数据，用于实时计算
2. 本地文件：存当天的完整数据，用于盘后复盘
3. 数据库：存历史数据，用于回测和策略优化

内存缓存我用Redis，速度快，还支持过期自动清理。代码大概这样：

import redis
import json

r = redis.Redis(host='localhost', port=6379, db=0)

def cache_tick(stock_code, tick_data):
    key = f"tick:{stock_code}:{tick_data['time']}"
    r.setex(key, 60, json.dumps(tick_data))  # 60秒过期

def get_latest_tick(stock_code):
    keys = r.keys(f"tick:{stock_code}:*")
    if not keys:
        return None
    latest_key = max(keys)
    return json.loads(r.get(latest_key))

本地文件存储，我推荐用Parquet格式。比CSV快10倍，压缩率也高。每天收盘后，把当天的Tick数据转成Parquet存起来。

import pandas as pd

# 假设tick_list是当天的所有Tick数据
df = pd.DataFrame(tick_list)
df.to_parquet(f"data/2024-01-15_tick.parquet", compression="snappy")

数据库层面，我踩过不少坑。最开始用MySQL，写入速度跟不上。后来换了ClickHouse，专门为时序数据优化的，写入速度能达到每秒百万行。

最后，画一张图总结一下整个数据流转：

这张图把整个数据流转讲清楚了。从交易所出来，经过券商或第三方接口，到Python对接层，然后分三路存储。最后，策略程序从不同存储层取数据做计算。

我个人习惯是：实时策略直接从Redis取数据，延迟最低。盘后复盘用Parquet文件，查询灵活。历史回测用ClickHouse，数据量大也不怕。