Python接入Level-2全推行情：从连接解析到断线重连实战

简介面向量化与高频交易场景的《同花顺Level-2全推行情Python API操作手册》聚焦涨停板等策略开发者系统解决沪深市场Level-2行情接入与解析问题。手册从TDS系统C/S架构讲起说明其从交易所VDE接入行情、经解码与订阅处理后以API形式输出的低延迟机制并给出64位双核、8GB内存、Python3.5以上版本等运行环境要求在数据层覆盖五档/十档快照、逐笔成交、逐笔委托、委托队列、极速盘口等类型详解初始化实例、注册回调函数、设置环境参数、登录订阅等核心流程。PDF格式共1个文件压缩包约850KB附录保留V2.0.0至V2.1.9完整修订记录呈现多连接模式、级联、异步订阅、离线调试、字段精度优化等重要演进。已有2809人学习下载适合量化投研人员、交易系统开发者和Python初学者快速上手也可作为排查Level-2接口细节的案头参考。1. 为什么Level-2全推行情值得用Python自己接Level-2全推行情指的是服务端不按你订阅的股票列表过滤而是把全市场所有股票的快照、逐笔、委托队列一股脑推过来。不少人一听“全推”就下意识觉得“Python不适合做这种高吞吐的事”实际上只要把异步模型和数据解析分开Python完全能扛住几百路并发。这篇文章围绕我调通一套Level-2全推行情Python API的过程从连接、解析、落库到断线重连把文档里没写明白的细节和踩坑记录补齐。适合正在自建行情采集通道、想做本地L2库的开发者。读完你应该能自己动手实现一个能稳定跑一整天的全推采集脚本而不是一直对着文档翻车。2. 先把连接模型吃透全推不是“订阅一下就有数据”2.1 全推和订阅制的数据边界在全推模型下服务端不会等你一个个股票代码去订阅它会把整个市场的数据都推给你。所谓“全推”的边界是行情源的一次快照对应全市场股票而不是某个自选股池。我接触过的一款L2 API连接成功后只要发一个push_all指令随后每个数据包里的股票代码就是全部市场。订阅制则是另一种模型你向服务端上报关注的股票列表服务端只给你这些股票的行情。两种模型在带宽、内存和处理逻辑上有本质差异。全推的流量通常是订阅制的几十倍。以全市场为例光逐笔成交一天就可能上亿条快照每三秒全市场刷一次。如果你只做少数股票的因子订阅制更省但如果要做全市场选股或回放全推天然适合。全推需要设计一套按消息类型分发的解析器而不是维护几百个订阅回调。换句话说你需要从“事件驱动单标的”思维切到“流式处理全市场”思维。我常用的一个判断标准如果策略池只有几十只股票优先用订阅制如果准备每天落全市场流水直接上全推。全推多出来的带宽成本通常可以接受但Python程序必须从一开始就按吞吐量设计否则后面扩展很痛苦。这里说的吞吐量不仅仅是消息条数还包括单位时间内的字节数和对象创建数量这三者是三个不同的问题。拿到API文档时我第一件事不是看Python封装而是看它的通信协议和消息类型定义。很多服务商给的Python SDK反而把关键字段藏起来了文档里只有一句“请使用最新SDK”。真正能指导你写解析器的永远是那几张字段偏移表。如果表里没有每个字段的单位、scale和字节序这个文档还不够格让你直接写代码。2.2 连接生命周期鉴权、心跳、会话过期任何行情API都不希望你连上就永远占用一个连接所以会设计鉴权与会话有效期。常见流程先TCP连接再发送登录请求登录成功后服务端返回一个会话标识。后续所有包带这个标识或者靠连接本身维持会话。全推连接多数是长连接服务端每几秒检查一次客户端心跳如果超过阈值没有心跳就会把连接断开。心跳包的内容因厂商而异有的只需要发一个空包有的要求带上本地时间或会话ID。参数上需要关心的有心跳间隔、心跳超时阈值、重连次数。我一般把心跳间隔设为文档默认值的一半留出网络抖动余量。比如文档说30秒必须发送一次我就在10到15秒发一次。注意不要过于频繁否则会被网关当成异常流量限流。会话过期是最容易忽视的坑。登录后账号在其他地方登录会让当前会话失效也可能服务端每天凌晨重新要求鉴权。如果程序只是无限重连但没重新鉴权就会一直连上后被断开。所以重连状态机里必须有一个step先重新登录再恢复数据流。很多线上采集器出问题都出在这个闭环没走通。全推还有一个特殊点重连后是否需要重新发送“启动全推”指令。大多数服务端会保留前一个会话的上下文但有些服务端在连接断开后会自动清理所有会话状态你必须重走一遍完整流程。建议在代码里把连接流程写成状态机而不是线性执行的函数。状态至少包含已连接未登录、已登录未启动、已启动接收中、重连等待。这样任何一步异常都能定位。2.3 最小Python连接代码先用socket把流接住import socket import struct import json # 这些参数在服务商的接入文档里都能找到 HOST l2.example.com # 真实环境换成行情网关IP PORT 7702 # 典型行情端口 USER your_account PASSWORD your_password # 建立TCP连接超时设短一点避免长时间假死 sock socket.create_connection((HOST, PORT), timeout10) sock.settimeout(5) # 构造登录包这里用JSON演示真实协议可能是二进制 login json.dumps({ cmd: login, user: USER, password: PASSWORD, }).encode(utf-8) # 很多网关会先收4字节大端长度所以发包前先带上长度 sock.sendall(struct.pack(I, len(login)) login) # 读取一个完整包前4字节是payload长度后面是payload def recv_packet(s): head s.recv(4) if not head: return None length struct.unpack(I, head)[0] body b while len(body) length: chunk s.recv(length - len(body)) if not chunk: break body chunk return body resp recv_packet(sock) print(resp)逻辑说明使用struct.pack(I)是网络字节序大端这是行情协议里最常用的长度表示recv返回的字节数可能小于请求的长度所以要用循环收满设置5秒超时是为了让心跳逻辑能及时感知死连接。参数说明timeout10是建立连接阶段的最大等待s.settimeout(5)是每次读包超时。如果读超时抛出socket.timeout你可以在外层捕获后触发重连。HOST/PORT需要从服务商控制台申请USER/PASSWORD由业务账号创建。登录成功后一般还要发送“启动全推”的指令之后才能真正收到行情。我曾经在这个最小代码里只做到登录就停住了后来抓包对比才发现少了一条start_push控制包。下一章讲数据解析时你会看到消息类型这件事有多重要。3. 行情数据解析从字节流到干净的行情对象3.1 常见字段布局快照、逐笔、委托队列全推行情的每一条消息一般先带一个消息头头里至少包含消息类型和序号。不同类型的数据字段布局差别很大。快照是最易理解的一个股票当前时刻的完整状态。L2快照通常有十档买卖盘口比L1的五六档更多字段包括时间戳、最新价、成交量、成交额、买卖十档的价格和手数。注意有些服务商把盘口放在快照里有些把委托队列单独推送。逐笔成交是最有研究价值的数据它记录每一笔实际成交的发生顺序。常见字段有成交时间、成交价格、成交数量、成交编号、买号、卖号、执行类型。这里的成交编号用于和逐笔委托关联判断主动买卖方向。委托队列则是在某个价位上的挂单明细能看到每笔委托的量对盘口分析很有用。从程序角度你需要先把三种消息类型分流。我常用的方式是读取消息头里的type字段后用一个字典dispatch到不同解析函数。不要在一个函数里处理所有类型否则后面维护成本很高。下面是一张我整理过的字段对照表具体字段名以各服务商文档为准。消息类型消息头中的type典型字段推送频率快照1时间戳、十档价量、总量每3秒一次逐笔成交2时间、价格、量、买卖方向、成交编号随机可能密集逐笔委托3时间、价格、挂单量、委托号随机可能密集这样设计的好处是后续增加新的消息类型只需要新增一个解析函数并注册到字典里原来的流程不用动。我一般还会在dispatch函数里加一个统计计数器看看每秒收到多少条逐笔、多少条快照这能帮你尽早发现流量异常。3.2 scale因子与价格精度陷阱把行情价格传成整数然后附带一个scale是所有L2 API的通用做法。比如最新价显示为42.30元协议里可能传4230scale2。我们解析时必须把scale落到对应的字段上。最容易翻车的点字段并不共用同一个scale可能价格scale2成交量scale0盘口价量又各有一个scale。如果一个结构里混用不同scale要分别处理。另一个精度坑是时间戳。有的接口用秒有的用毫秒有的从开盘日零点开始计数。如果直接用会导致分钟K线分错位置。我的解决方式是先用一个已知的盘面数据比如现价、当前时间做一次对拍确认时间戳的单位后再写进正式解析器。这个步骤看似简单但我见过不少人在这一步省事最后对账时花了两天时间。建议写一个transform函数而不是在解析函数里到处乘除这样字段改scale时只改一处。这个函数要能处理价格对象和整数两种输入避免回调里出错。还要注意负数有些字段如涨跌额在协议里用补码表示直接当成无符号数转出来会变成十几亿的异常数字。签名扩展问题在金融协议里非常常见遇到int字段先看文档里有没有注明“有符号”。3.3 用struct和dataclass解析一帧数据import struct from dataclasses import dataclass # 假设快照payload布局如下使用大端字节序 # 6字节ASCII股票代码8字节时间戳毫秒4字节最新价int8字节成交量int # 实际协议一般会把字段偏移表写在文档里这里仅为演示解析思路 SNAPSHOT_FORMAT 6sQ i Q SNAPSHOT_SIZE struct.calcsize(SNAPSHOT_FORMAT) dataclass class Snapshot: code: str timestamp: int last_price: int volume: int def parse_snapshot(payload: bytes) - Snapshot: code_b, ts, price, volume struct.unpack_from(SNAPSHOT_FORMAT, payload, 0) return Snapshot( codecode_b.decode(ascii).strip(\x00), timestampts, last_priceprice, volumevolume, )逻辑说明格式字符串6sQ i Q表示大端6字节字符串、无符号64位整数、32位整数、无符号64位整数。struct.unpack_from可以让你从payload的任意偏移开始解包方便一个包里塞多帧数据。calcsize用来校验字段长度如果文档里给出了字节数你应该先用calcsize核对避免错位。参数说明如果服务商文档里时间戳是32位秒就把Q改成I如果价格字段是2字节就把i改成h。不要盲目套用这个格式。逐笔成交的解析类似只是字段不同。建议先抓一组真实数据包用hexdump对比字段偏移再固化解析类。解析后最好返回dataclass或namedtuple不要返回字典字典键的字符串开销很大在高频逐笔下会拖慢速度。我通常会用__slots__优化内存但可读性会牺牲一点。另外不要在解析函数里做数据库写入哪怕是一条sqlite insert都会有巨大延迟。解析和落库必须解耦这正是下一章要解决的问题。4. 把全推流落库增量处理与断点续传4.1 为什么全推不能只靠内存全推一天的数据量在我看来是看着不大但收集起来会撑爆内存的那种。单只股票逐笔一天可能几千笔但全市场就是千万到亿级。每个Python对象如果以字典形式保存一条记录占用几百字节一个亿就是几十GB。别指望8G内存机器能扛住。因此采集器必须从一开始就区分实时计算和离线归档。如果你只是算分钟K线那内存保留增量即可不需要落全量如果你要保存历史逐笔就必须落盘或写数据库。落在哪个存储要看查询方式做数据回放用Parquet列存很合适做实时因子查询用ClickHouse这类列式数据库如果是小型项目SQLite加批量事务也能应付。我建议把采集器拆成两个阶段网络解析阶段只处理字节流落库阶段专门消费解析后的对象。中间用有界队列衔接不要让网络解析线程直接操作数据库。还要预先估算存储空间一条逐笔记录如果保留完整字段约80字节一亿条就是8GB加上索引和文件系统开销至少要预留一点五倍空间。这个估算在项目启动前就应该做否则跑几天磁盘满了很被动。4.2 双缓冲队列与批量写入双缓冲其实就是两块区域交替使用避免生产者和消费者互相锁等待。在Python里可以用asyncio.Queue实现类似效果。生产协程把解析后的对象放进队列消费协程攒够一批再写库。这样写库的IO开销被摊平数据库事务数量也大幅减少。import asyncio async def batch_write(items): # 这里省略连接数据库的细节把items一次性写入 # 例如 INSERT INTO ticks (code, ts, price, volume) VALUES ... pass async def consume(queue: asyncio.Queue): pending [] while True: item await queue.get() pending.append(item) if len(pending) 500: await batch_write(pending) pending.clear()逻辑说明queue.get()会等待生产者放入数据避免空转批量阈值设为500条是为了在延迟和写库开销之间平衡。如果写一条要1ms500条一次写可能只要2ms效率提升很明显。参数说明阈值可以根据每条记录大小调整如果每条记录很大就调小阈值比如100条如果数据库写入慢可以调大到2000条但要小心pending积压。还要给队列设置最大长度。asyncio.Queue(maxsize10000)会在满时让生产者put等待这样起到背压作用。如果生产者因为阻塞而等待网络层的内核缓冲区会慢慢变大最终导致TCP接收窗口变小行情源会感知到压力。这不是坏事至少不会内存爆炸。如果对实时性要求极高可以丢弃旧数据而不是阻塞网络解析。我实际跑下来批量写入还带来一个额外好处数据库的WAL日志和事务开销大幅降低回放时查询效率也更高。但要注意批量写入不能由网络解析协程直接执行否则某次大事务会阻塞整个连接的事件循环。正确做法是把batch_write丢给asyncio.to_thread执行。4.3 断线重连后的数据补齐方案断线重连在长连接里是常态关键是重连后如何把缺口补上。很多服务端支持按序号续推你只需要在重连时把本地已处理的最大消息序号带给服务端。但需要注意的是不同消息类型的序号可能是分开的所以要用字典分别保存每个类型的last_seq。# 记录每种消息类型最近的序号 last_seq {} def handle_packet(msg_type, seq, payload): if seq last_seq.get(msg_type, 0): # 重复或乱序直接丢弃 return # 正常处理 parse_and_store(msg_type, payload) last_seq[msg_type] seq逻辑说明这里假设同一消息类型的序号严格递增。如果服务端返回的序号是乱序的就不能简单丢弃需要做pending排序。参数说明last_seq.get(msg_type, 0)让第一个包能够通过判断。断线后重连时发送resume请求携带各类型最后一次序号。如果服务端不支持就需要用快照校正。我有一次重连后以为序号续上就行结果半小时后发现分钟线全是缺口。后来才明白支持续推的服务端也只能补行情网关内存里的数据内存外的历史缺口还得靠历史数据接口拉。所以建议每天收盘后做一次对账发现缺口就用数据商的历史数据接口补齐。这个对账任务最好做成独立脚本不占采集器带宽。还有一点容易被忽略断线期间全推快照不再推送但时间戳还在走。重连后拿到第一帧快照时要判断它和本地最后一条记录之间的时间差如果超过阈值就触发补量流程。有些服务商在全推之外还提供一个“离线快照”接口可以在重连后先拉一份全市场快照作为基准再开始接收增量。这个方案比单纯靠序号续传更可靠。5. Level-2全推实盘接口的避坑手册这一章列出我在生产环境里遇到过的五类问题每类都按现象、原因、解决顺序写。这些问题不是从文档里读出来的而是实际跑了几天后一条条撞上的。能避一个是一个。5.1 连接正常却收不到新行情现象TCP连接建立成功心跳也正常就是收不到任何快照或逐笔。原因很多全推行情API在鉴权后默认是静默状态还要再发一个启动推送的指令。我第一版就漏了这一步。解决登录成功后查一下接口文档里和推送相关的指令比如subscribe、start_push、enter_market。发送后观察第一个数据包。另外确认你连的是行情端口而不是交易登录端口这两个端口往往共用文档却走不同协议。排查这个问题的过程也很有代表性。我先用tcpdump看了网卡流量发现连接上有来有回但都是心跳应答没有业务数据。然后我试着直接发送一个最常见的subscribe all结果收到了一大堆数据包。这让我确定了问题不在网络而在应用层缺少启动指令。如果你也遇到类似情况建议先看一下服务端有没有一个“数据开关”的概念。5.2 价格突然变成小数错位现象某个股票最新价显示4230实际应为42.30。原因scale因子没有解析。协议字段里有一个price_scale或decimal快照、逐笔、盘口可能各自带一个。解决写一个字段转换层先读scale再乘10**-scale。不要直接在解析函数里除100因为不同字段scale不同。我建议在dataclass里保留原始整数和scale在输出时再转换这样调试方便得多。我在开发时用了一个笨办法打开行情软件找一个价格和小数位都明确的股票比如最新价12.34元然后看协议返回的整数字段是多少。如果返回1234scale就是2如果返回12340scale就是3。这个方法虽然不是官方文档但能让你快速确认字段定义。写单元测试时把类似“价格4230scale2期望得到42.30”的用例固化下来以后改代码不怕改错。5.3 程序跑一段时间后内存只涨不降现象内存曲线像楼梯一样10分钟涨几百MB最后整个进程被操作系统杀掉。原因最常见的是asyncio.Queue无界增长消费者写库速度跟不上另一个常见原因是把解析后的数据加入了全局list或日志队列没有清理。解决给队列加maxsize设置背压排查代码里是否有全局append操作。还有一个容易被忽略的点日志文件如果每帧打一条磁盘和内存都会被拖垮。生产环境只保留告警日志不要打印每一条行情。这类问题用tracemalloc可以快速定位。启动时加上python -X tracemalloc5 collector.py跑一段时间后调用tracemalloc.get_traced_memory()和snapshot.statistics就能看到内存被哪些对象占用。我记得第一次定位时发现是asyncio.Queue里积压了十几万条逐笔对象每条是一个dataclass光字段就占了很大空间。后来改成maxsize1000内存立刻平稳。5.4 重连后收到一堆旧数据现象重连后日志里出现大量时间戳早于本地最新时间的数据导致重复计数。原因服务端从某个内存缓冲区回放数据但没有按本地断点截断。解决在解析层维护last_seq和last_timestamp双条件过滤。如果seq小于等于本地记录直接丢弃如果seq相同但时间戳更早也要丢弃。注意不要只用时间戳因为集合竞价阶段可能出现无成交但时间戳推进的情况。我的习惯是seq为主时间为辅。重复数据如果不处理会导致成交量被重复计算因子信号完全失真。所以我不仅要在解析时过滤还要在落库时建一个唯一索引以消息类型和seq作为联合唯一键。这样即使过滤逻辑漏了一条数据库也会拒绝重复插入。这个双保险让我后面做数据对账时省了很多事。5.5 不同行情源返回的“全推”语义不一样现象同一个股票从两个行情源收到的累计成交量对不上。原因所谓全推往往只推普通交易时段有些源还带盘前和收盘集合竞价有些源把集合竞价单独放一个字段还有一些源在瞬时停牌时跳过快照。解决对接前先做一份样例数据的字段对照表把两个源的快照字段意义逐一比对。不要假设所有L2服务商都是同一个标准。如果做全市场数据对比务必先统一时间切片和成交口径。这个问题的坑在于两个源单看都正确完全无法判断谁有问题。我当时的做法是取同一段分钟区间分别计算两边的累计成交量再和行情软件里显示的“今开、总量”做三角验证。最终发现是一个源把集合竞价成交量并入了第一笔快照而另一个源没有。统一口径后两边就一致了。所以设计存储时最好给每条记录加一个source字段方便以后追溯差异。6. 从能跑到可用一个最小全推采集器的进阶打磨6.1 把采集器变成命令行工具每次改连接地址都要改代码很蠢我把采集器封装成命令行工具参数包括host、port、账号、输出目录。配置文件用简单的TOML或YAML程序启动时加载。python collector.py --config config.toml --log logs/这样部署到新环境五秒就能跑。配置文件里还会带上股票过滤规则虽然全推是全市场数据但落库时可以只保留需要的板块减少存储压力。这个过滤动作要在解析之后、入队之前做不要在拿到全推字节流之前做因为那样会干扰序号连续性。6.2 验证数据完整性的两个偏方第一找一个本地L1行情软件把全推采集到的快照和它显示的现价、总量做对拍误差应该在合理范围内。第二统计每只股票每分钟的逐笔条数如果一分钟为零很大概率是断线缺口。用pandas做一次groupby就能发现。这两个方法都很土但真的能救命。我第一次跑长测就是用逐笔条数统计发现下午开盘前有一个两分钟的缺口当时日志里没有任何报错。后来查原因才知道是网络切换导致连接静默断开而操作系统要很久才能感知TCP断线。从那以后我不仅关心内存曲线还关心每分钟数据条数这个业务指标。6.3 上线前检查清单连接参数和账号信息走环境变量或配置文件不写死在代码里。心跳间隔设置为默认值一半。队列有界写库阈值有上限。断线重连逻辑覆盖重新鉴权和启动推送。磁盘空间预留一天数据量的一点五倍以上。挂机运行24小时后看内存曲线是否平稳。我个人习惯是每次改完配置先跑两分钟看第一条快照的字段再跑一小时看有没有断线最后才会挂长期。这一步能避免很多“睡一觉起来发现采集器挂了”的尴尬。你也最好把这几步固定成自己的检查习惯而不是每次凭感觉。希望帮到你。本文还有配套的精品资源点击获取