简介：这份《同花顺level-2全推行情Python API操作手册》面向从事高频交易、量化投资及涨停板策略的证券开发者，系统讲解如何通过Python API获取并处理沪深市场Level-2高频行情数据。手册以PDF形式交付，压缩包内仅1个文件，体积约850KB，内容紧凑、便于随查随用。目前已有2813人学习下载，在量化圈内具备一定参考热度。手册完整梳理了TDS系统的C/S架构与运行环境要求，并逐版本记录了从V2.0.0到V2.1.9的迭代细节：多连接模式改造、上证逐笔委托接入、级联功能、Level-1数据支持、Linux兼容、极速盘口与异步订阅接口等，还涉及字段类型由int改为float、回调参数由int变更为dict等易踩坑的变更点。读者可据此掌握五档与十档快照、逐笔成交与委托、委托队列等数据的订阅与回调写法，理解环境参数配置、断线重连与离线调试思路，从而搭建低延迟行情链路，为交易策略提供稳定数据支撑。
1. 同花顺 Level-2 全推行情 Python API 到底能拿到什么数据
很多人第一次接触 Level-2 行情，脑子里只有「比普通行情快」这一个印象，其实快只是表象，真正值钱的是逐笔委托和逐笔成交这两路全推数据。普通快照行情每 3 秒给你一张切片，你看到的是被压缩过的结果；而全推行情是把交易所每一笔委托、每一笔成交都推给你，你自己在本地重建订单簿。这两者的差别，就像看比赛集锦和看全场直播——集锦能告诉你谁赢了，直播才能告诉你赢的过程里资金是怎么一步步进场的。
这篇手册要解决的问题很具体：怎么用 Python 把同花顺 Level-2 的全推行情接进来，怎么把原始推送解析成能算的字段，怎么在本地维护一份可用的盘口，以及哪些参数设错了会让你在盘中才发现数据是错的。适合两类人：一类是做量化策略、需要更细粒度微观结构信号的开发者；另一类是做盘中监控、想自己算大单净流入和撤单率的工具党。前提是你已经有一个能正常登录的 Level-2 行情账号，本文不涉及账号获取，只讲拿到权限之后怎么把数据用起来。
2. 全推行情的数据结构：逐笔委托与逐笔成交怎么拆
2.1 快照、逐笔、全推三者的关系
先把概念理清楚，不然后面解析字段会一直糊。快照行情是定时切片，字段是「当前买一卖一、累计成交量」这类状态量；逐笔成交是每一笔撮合成交的记录，带成交价、成交量、买卖方向；逐笔委托是每一笔新进委托和撤单的记录，带委托价、委托量、委托类型。全推的意思是，交易所把逐笔委托和逐笔成交实时、不丢地推给行情源，而不是等你来轮询。
同花顺 Level-2 的 Python 接口，本质上是一个长连接客户端：你订阅标的，服务端持续推消息，你在回调里处理。所以整个程序的结构不是「请求-响应」，而是「连接-订阅-回调-本地重建」。这个认知不转过来，写出来的代码会一直想去找一个get_orderbook()这样的同步函数，然后发现根本没有。
一个常见的误区是把逐笔成交当成成交流水直接用。实际上逐笔成交里有一类叫「撤单成交」或者「盘后成交」的记录，方向字段和普通成交不一样，如果你不区分，算出来的主动买入量会偏高。这个坑后面第 5 章会专门讲。
2.2 逐笔委托的字段含义与方向判定
逐笔委托的核心字段一般包括：证券代码、委托时间（精确到毫秒甚至更细）、委托编号、委托价格、委托数量、委托类型（买/卖）、委托类别（限价/市价/本方最优等）。方向判定看起来简单，但有两个细节容易翻车。
第一，委托类型里的「市价」在深市和沪市语义不同。沪市的市价委托在撮合时会转成对手方最优价格，深市的市价委托则可能带五档即时成交剩余撤销等细分类型。如果你只按「买/卖」两分类，遇到市价单时价格字段可能是 0 或者一个占位值，直接拿去算加权均价就会把结果拉偏。
第二，撤单在逐笔委托里通常表现为一条新的委托记录，数量为负或者带一个「撤单」标志。不同行情源的表达方式不一样，有的用独立消息类型，有的用数量字段的正负号。你必须在接入的第一天就把这个规则确认清楚，否则本地订单簿会越跑越歪。
下面是一段典型的逐笔委托解析骨架，字段名按常见命名习惯写，实际以你拿到的接口文档为准：
# 逐笔委托解析：把原始推送转成结构化记录 def parse_order(msg): """ msg: 行情源推送的原始委托消息（dict 或 bytes 解包后） 返回统一结构的 dict，方便后续重建订单簿 """ record = { "code": msg.get("code"), # 证券代码 "ts": msg.get("time"), # 委托时间，毫秒级 "seq": msg.get("order_no"), # 委托编号，用于去重和排序 "price": msg.get("price", 0.0), # 委托价，市价单可能为 0 "volume": msg.get("volume", 0), # 委托量，撤单可能为负 "side": msg.get("side"), # 'B' 买 / 'S' 卖 "order_type": msg.get("order_type"), # 'LIMIT' / 'MARKET' / 'BEST' } # 市价单价格字段不可信，标记出来，后续单独处理 if record["order_type"] == "MARKET": record["price_valid"] = False else: record["price_valid"] = True # 撤单识别：数量为负或带撤单标志 record["is_cancel"] = record["volume"] < 0 or msg.get("cancel_flag", False) return record这段代码的关键不在语法，而在两个判断：price_valid和is_cancel。前者让你在算盘口均价时能跳过市价单的无效价格，后者让你在重建订单簿时知道这条记录是「加挂」还是「撤掉」。参数上，seq一定要保留，因为全推消息在极端行情下可能乱序到达，靠时间戳排序不够稳，委托编号才是可靠的排序依据。
2.3 逐笔成交的字段与主动方向还原
逐笔成交的字段一般有：证券代码、成交时间、成交编号、成交价格、成交数量、买卖方向、成交类型。这里的「买卖方向」指的是主动方，也就是这笔成交是买方主动打上去的还是卖方主动砸下来的。这个字段是算主动买入量的基础，但很多行情源给的方向是「成交方向」而非「主动方向」，需要你结合逐笔委托的委托编号去反推。
反推的逻辑是：一笔成交必然对应之前的一条委托，如果成交记录里带了买方委托号和卖方委托号，你就能知道这笔成交是吃掉了哪一方的挂单。吃掉买方挂单的是主动卖，吃掉卖方挂单的是主动买。如果接口只给了成交方向没给委托号，那就只能按「成交价接近卖一则为主动买」这种近似规则来判，精度会下降。
# 逐笔成交解析：还原主动买卖方向 def parse_trade(msg, order_cache): """ msg: 原始成交消息 order_cache: 本地维护的委托缓存 {order_no: record} """ trade = { "code": msg.get("code"), "ts": msg.get("time"), "trade_no": msg.get("trade_no"), "price": msg.get("price"), "volume": msg.get("volume"), "bid_order": msg.get("bid_order_no"), # 买方委托号 "ask_order": msg.get("ask_order_no"), # 卖方委托号 } # 优先用委托号反推主动方 if trade["ask_order"] and trade["ask_order"] in order_cache: trade["active_side"] = "B" # 吃掉卖方挂单，主动买 elif trade["bid_order"] and trade["bid_order"] in order_cache: trade["active_side"] = "S" # 吃掉买方挂单，主动卖 else: # 兜底：按成交方向字段，精度较低 trade["active_side"] = msg.get("side", "U") return tradeorder_cache是这段代码的核心依赖。你需要在收到逐笔委托时把委托编号存进去，收到对应成交后可以延迟清理。缓存不能无限增长，一般按时间窗口清理，比如保留最近 5 分钟的委托记录。这个窗口设太小，反推会失败；设太大，内存会涨。5 分钟是我在多数标的上试下来比较稳的值，冷门股可以缩到 2 分钟。
3. 用 Python 接全推行情：连接、订阅、回调的最小闭环
3.1 连接与登录的参数怎么设
全推行情的 Python 接入，第一步是建立长连接并完成登录鉴权。不同行情源的登录方式不一样，常见的是「账号 + 密码 + 服务器地址 + 端口」四件套，有的还需要一个客户端版本号或者协议版本号。这里最容易踩的坑是服务器地址选错：行情源通常有多个接入点，有的只推快照，有的才推 Level-2 全推。你如果连到了快照节点，会发现订阅成功但收到的数据里没有逐笔字段，然后怀疑是代码问题，其实是节点问题。
登录参数里还有一个容易被忽略的是心跳间隔。长连接如果长时间没有数据往来，中间的网络设备可能会把连接掐掉。心跳设太频繁浪费资源，设太稀疏又容易断。一般 15 到 30 秒发一次心跳比较合适，具体看行情源文档。我一般会设 20 秒，并且在心跳失败时触发重连，而不是等连接彻底断了才处理。
# 连接与登录骨架 import socket import json import threading import time class L2Client: def __init__(self, host, port, user, password): self.host = host self.port = port self.user = user self.password = password self.sock = None self.running = False self.heartbeat_interval = 20 # 心跳间隔，秒 self.last_recv = time.time() def connect(self): self.sock = socket.create_connection((self.host, self.port), timeout=10) login_msg = { "cmd": "login", "user": self.user, "password": self.password, "version": "1.0", # 协议版本，按行情源要求填 } self.sock.sendall((json.dumps(login_msg) + "\n").encode()) resp = self.sock.recv(4096) if b"ok" not in resp.lower(): raise RuntimeError("登录失败，检查账号或节点地址") self.running = True threading.Thread(target=self._heartbeat, daemon=True).start() threading.Thread(target=self._recv_loop, daemon=True).start() def _heartbeat(self): while self.running: time.sleep(self.heartbeat_interval) try: self.sock.sendall(b'{"cmd":"ping"}\n') except Exception: self._reconnect() break def _reconnect(self): # 重连逻辑：先关旧连接，再重连并重新订阅 self.running = False try: self.sock.close() except Exception: pass time.sleep(2) self.connect() self.resubscribe()这段代码里version字段是占位，实际填什么必须看行情源要求，填错会直接登录失败。_reconnect里重新订阅这一步很关键，很多人重连后忘了重新订阅，结果连接是活的但一条数据都没有，排查半天。resubscribe方法需要你自己维护一份已订阅列表，重连后遍历重订。
3.2 订阅标的与回调分发
登录成功后就是订阅。订阅消息一般包含证券代码列表和订阅类型（逐笔委托、逐笔成交、快照等）。这里有个实操细节：不要一次性订阅太多标的。全推行情的数据量很大，单只活跃股票在开盘时段每秒可能产生几十到上百条逐笔消息，你订阅几十只，回调处理跟不上就会积压，最后要么丢数据要么内存爆掉。
我的做法是分两级：核心关注的几只标的订阅全推（逐笔委托 + 逐笔成交），其余标的只订阅快照。这样既拿到关键标的的微观结构，又不会把连接压垮。订阅数量上限取决于你的机器和网络，一般单连接订阅 10 到 20 只全推标的比较稳，再多就要考虑多连接分片。
# 订阅与回调分发 def subscribe(self, codes, sub_type="L2"): """ codes: 证券代码列表 sub_type: 'L2' 全推 / 'SNAPSHOT' 快照 """ sub_msg = { "cmd": "subscribe", "codes": codes, "type": sub_type, } self.sock.sendall((json.dumps(sub_msg) + "\n").encode()) self.subscribed.extend(codes) def _recv_loop(self): buffer = b"" while self.running: try: data = self.sock.recv(65536) if not data: break buffer += data # 按换行拆包，处理粘包 while b"\n" in buffer: line, buffer = buffer.split(b"\n", 1) if not line.strip(): continue msg = json.loads(line.decode()) self._dispatch(msg) self.last_recv = time.time() except Exception as e: print("接收异常:", e) break def _dispatch(self, msg): mtype = msg.get("type") if mtype == "order": record = parse_order(msg) self.order_cache[record["seq"]] = record self.on_order(record) elif mtype == "trade": trade = parse_trade(msg, self.order_cache) self.on_trade(trade) elif mtype == "snapshot": self.on_snapshot(msg)_recv_loop里的粘包处理是必须的。TCP 是流式协议，一次recv可能拿到半条消息，也可能拿到三条半。用换行符做分隔是最简单的方案，前提是行情源每条消息以换行结尾。如果行情源用的是定长头 + 变长体，你就得按长度字段来拆。这个规则在接入第一天就要确认，否则会随机出现 JSON 解析失败。
order_cache的清理也要在这里做。可以在_dispatch里加一个计数器，每处理 N 条消息检查一次缓存大小，超过阈值就按时间戳清理最老的记录。不要等到内存报警才想起来清。
3.3 本地订单簿重建的最小实现
拿到逐笔委托和逐笔成交之后，本地订单簿的重建逻辑是：委托进来就加挂或撤单，成交进来就扣减对应价位的量。这里不需要真的维护一个完整的红黑树，用一个dict按价格聚合就够了，因为你的目的通常是算盘口不平衡、大单挂撤，而不是做撮合。
# 本地订单簿：按价格聚合买卖挂单 class OrderBook: def __init__(self): self.bids = {} # {price: volume} self.asks = {} def apply_order(self, rec): book = self.bids if rec["side"] == "B" else self.asks price = rec["price"] if not rec["price_valid"]: return # 市价单不进盘口 if rec["is_cancel"]: book[price] = book.get(price, 0) + rec["volume"] # volume 为负 if book[price] <= 0: book.pop(price, None) else: book[price] = book.get(price, 0) + rec["volume"] def apply_trade(self, trade): # 成交扣减对手方挂单 if trade["active_side"] == "B": book = self.asks elif trade["active_side"] == "S": book = self.bids else: return price = trade["price"] if price in book: book[price] -= trade["volume"] if book[price] <= 0: book.pop(price, None) def top_n(self, n=5): bid_prices = sorted(self.bids.keys(), reverse=True)[:n] ask_prices = sorted(self.asks.keys())[:n] return ( [(p, self.bids[p]) for p in bid_prices], [(p, self.asks[p]) for p in ask_prices], )apply_order里撤单的处理是把负数量加回去，等价于减掉挂单量。这里要求你的parse_order已经把撤单的volume转成负数，如果行情源给的是正数加标志位，你需要在解析时就取负。apply_trade只扣减对手方，因为成交吃掉的是挂单，主动方的委托在成交前已经作为挂单存在或者作为市价单不进入盘口。
top_n每次调用都排序，在盘中高频调用会有性能问题。实际使用中我会维护一个有序结构或者缓存排序结果，只在盘口变化时更新。如果只是做监控、每秒算几次，直接排序也能接受。
4. 盘中实战：用全推数据算三个可落地的指标
4.1 大单净流入：阈值怎么定才不拍脑袋
大单净流入是最常被提到的指标，但「大单」的阈值定多少，很多人是拍脑袋。固定金额阈值（比如 50 万以上算大单）在不同价位的股票上效果差异很大：高价股一笔普通成交就过线，低价股要很大的量才过线。更稳的做法是用成交量的分位数来定阈值，比如取过去 20 个交易日逐笔成交量的 90 分位作为大单门槛，每天开盘前更新一次。
# 大单净流入：按分位数阈值统计 import numpy as np def calc_big_order_flow(trades, vol_threshold): """ trades: 当日逐笔成交列表，每条含 price/volume/active_side vol_threshold: 大单成交量门槛（股数） """ buy_amt = 0.0 sell_amt = 0.0 for t in trades: if t["volume"] < vol_threshold: continue amt = t["price"] * t["volume"] if t["active_side"] == "B": buy_amt += amt elif t["active_side"] == "S": sell_amt += amt return { "big_buy": buy_amt, "big_sell": sell_amt, "net_inflow": buy_amt - sell_amt, }vol_threshold的更新逻辑建议单独写一个脚本，每天收盘后跑一次，把结果存成配置文件。不要盘中动态算，因为盘中样本不完整，分位数会漂。参数上，90 分位是个起点，你可以根据标的活跃度调到 85 或 95。活跃股用 95 分位能过滤掉更多噪音，冷门股用 85 分位否则一天也没几笔大单。
4.2 撤单率：识别虚假挂单的关键
撤单率是 Level-2 数据相对快照行情的最大优势之一。快照行情你只能看到挂单量的变化，但不知道是成交吃掉了还是撤掉了。全推数据里，撤单是一条独立的委托记录，你可以精确统计。
撤单率的算法是：某价位或某方向上的撤单总量除以挂单总量。但要注意时间窗口，开盘集合竞价阶段的撤单和连续竞价阶段的撤单含义完全不同。集合竞价的撤单很多是试盘，连续竞价的撤单才更能反映真实意图。我一般只统计连续竞价时段（9:30 到 11:30，13:00 到 15:00）的撤单。
# 撤单率统计：按方向和时间窗口 def calc_cancel_rate(orders, start_ts, end_ts): """ orders: 逐笔委托列表 start_ts/end_ts: 统计时间窗口，毫秒时间戳 """ add_vol = {"B": 0, "S": 0} cancel_vol = {"B": 0, "S": 0} for o in orders: if not (start_ts <= o["ts"] <= end_ts): continue if o["is_cancel"]: cancel_vol[o["side"]] += abs(o["volume"]) else: add_vol[o["side"]] += o["volume"] result = {} for side in ("B", "S"): total = add_vol[side] result[side] = cancel_vol[side] / total if total > 0 else 0.0 return result这个指标单独看意义有限，要结合价格位置。买一价附近的大量撤单，可能是护盘资金在试探；卖一价附近的大量撤单，可能是压盘资金在虚张声势。我一般会把撤单率和盘口不平衡度放在一起看，两个指标同向时信号更可靠。
4.3 盘口不平衡度：买卖压力比的实时计算
盘口不平衡度是买一到买五的总挂单量除以卖一到卖五的总挂单量，或者用差值除以总和。这个指标在快照行情上也能算，但全推数据下你能算得更细：可以只算最近 1 秒内新挂的单，过滤掉历史挂单的干扰。
# 盘口不平衡度：基于本地订单簿 def imbalance(book, n=5, recent_only=False, recent_window=1000): """ book: OrderBook 实例 n: 取前 n 档 recent_only: 是否只统计最近挂单 recent_window: 最近挂单的时间窗口，毫秒 """ bids, asks = book.top_n(n) bid_vol = sum(v for _, v in bids) ask_vol = sum(v for _, v in asks) total = bid_vol + ask_vol if total == 0: return 0.0 return (bid_vol - ask_vol) / total返回值在 -1 到 1 之间，越接近 1 说明买盘压力越大。recent_only模式需要你在 OrderBook 里额外记录每个价位的挂单时间，实现会复杂一些，但对捕捉瞬时压力变化更敏感。我一般在做短线监控时用 recent_only，做趋势判断时用全量。
5. 避坑与排查：全推行情最容易翻车的五个地方
5.1 现象：订阅成功但收不到逐笔数据
原因通常是连到了快照节点，或者订阅类型参数写错。有些行情源的订阅消息里，type字段区分L1和L2，你填了L2但账号权限只到L1，服务端不会报错，只是不推逐笔。解决方法是先确认账号权限，再确认节点地址，最后用一条已知会推逐笔的活跃标的做测试，看回调里有没有order和trade类型。
5.2 现象：盘口越跑越歪，买一卖一价格明显不对
原因是撤单没被正确识别，或者市价单被当成了限价单加进了盘口。排查时先打印最近 100 条委托记录，看撤单的volume符号和is_cancel标志是否符合预期。如果撤单是正数且没有标志位，你的parse_order就漏了。市价单的问题更隐蔽，表现为某个价位突然多出一大笔不合理的挂单，检查price_valid是否生效。
5.3 现象：回调处理越来越慢，最后程序卡死
原因是单线程回调里做了耗时操作，比如每条成交都去查数据库或者做复杂计算。全推数据是持续高速到达的，回调里任何阻塞都会导致积压。解决方法是回调里只做最轻量的解析和入队，把计算放到另一个线程或者用队列异步处理。队列要有上限，满了就丢弃最老的数据并记录，宁可丢数据也不能让内存无限涨。
5.4 现象：重连之后数据正常但指标突然跳变
原因是重连后本地订单簿没有重置，旧数据和新数据混在一起。重连意味着你丢失了断线期间的所有逐笔消息，本地订单簿已经不可信了。正确做法是重连成功后清空订单簿和委托缓存，重新订阅，并且给指标加一个「预热期」，比如重连后 30 秒内的指标不对外输出，等订单簿重新稳定。
5.5 现象：不同标的的逐笔时间戳精度不一致
原因是不同交易所或不同行情源对时间戳的精度定义不同，有的到秒，有的到毫秒，有的到微秒。如果你用时间戳做跨标的排序或者对齐，精度不一致会导致顺序错乱。解决方法是统一转换到你需要的精度，并且在排序时优先用委托编号而不是时间戳。时间戳只用来做展示和统计窗口划分，不用来做严格排序。
6. 把全推数据落成可回放的研究样本
盘中实时算指标只是第一步，真正让这套东西产生复利的是把全推数据落盘成可回放的样本。你不可能每次调策略都等到开盘，把逐笔委托和逐笔成交按标的、按日期存成文件，盘后就能反复回放、反复调参。这一步做得好不好，决定了你后面是每天只能试一次，还是一晚上能试一百次。
落盘格式我推荐用列式存储，比如 Parquet，按「日期 + 标的」分区。逐笔委托和逐笔成交分两个文件存，字段就是前面解析出来的那些。不要用 CSV，全推数据一天单只活跃标的就可能上百万条，CSV 读写都太慢。Parquet 压缩后体积能小一个数量级，读取也快。
# 落盘：按日期和标的写 Parquet import pandas as pd import os def save_ticks(records, trade_date, code, data_type): """ records: 解析后的记录列表 trade_date: '2025-01-01' 格式 code: 证券代码 data_type: 'order' 或 'trade' """ if not records: return df = pd.DataFrame(records) # 按日期和标的建目录，方便分区读取 out_dir = os.path.join("l2_data", trade_date) os.makedirs(out_dir, exist_ok=True) path = os.path.join(out_dir, f"{code}_{data_type}.parquet") df.to_parquet(path, index=False, compression="snappy")回放的时候，你按时间戳排序，逐条喂给订单簿重建逻辑，就能在盘后完整复现盘中任何一个时刻的盘口状态。这里有个细节：回放时不要用实时回调那套异步逻辑，直接用同步循环，一条一条处理，保证顺序严格。实时环境下消息可能乱序，回放环境下你要的是确定性。
回放验证的一个实用技巧是对账：用回放重建出来的收盘盘口，和当天最后一条快照行情的盘口做对比。如果买一卖一价格和量都对得上，说明你的重建逻辑没问题；如果对不上，大概率是撤单处理或者市价单处理有遗漏。这个对账我一般每天收盘后跑一次，当成一个自动化的健康检查。
最后说一个我自己的习惯：每次改完解析逻辑或者订单簿重建逻辑，不要直接上实盘，先拿最近三个交易日的历史数据回放一遍，看指标曲线有没有异常跳变。全推数据的坑大多不是逻辑写错，而是边界情况没覆盖到——某只股票某天出了一笔特殊的委托类型，你的代码没处理，盘后回放才能发现。实盘发现问题的代价，比盘后回放高太多了。希望帮到你。
本文还有配套的精品资源，点击获取
