同花顺Level-2行情SDK实战：逐笔委托与涨停板策略落地

简介：《同花顺level-2全推行情Python API操作手册》面向从事高频交易与量化分析的证券开发者，尤其适合需要涨停板策略、逐笔委托与盘口深度数据的实战场景。手册以PDF形式呈现，压缩包内共1个文件，约850KB，内容围绕TDS系统的C/S架构、运行环境要求与Python接口调用展开。读者可系统了解从V2.0.0到V2.1.9的版本演进，包括多连接模式、级联功能、逐笔业务序号、通道号字段、极速盘口最新价、异步订阅接口async_sub及离线调试等关键更新，并掌握五档与十档行情快照、逐笔成交、逐笔委托、委托队列等数据的订阅与回调处理方式。手册还给出环境参数设置、回调函数定义与消息通知机制的具体说明，便于快速搭建低延迟行情接收链路。目前已有2825人学习下载，适合希望深入Level-2数据接口、优化交易策略的量化开发者参考。

1. 拆开一份 Level-2 行情 SDK：从涨停板策略到逐笔委托的落地路径

做涨停板策略的人大概都经历过这种场景：盘中盯着某只票的封单量，五档行情里买一挂着几十万手，但封单到底是真金白银还是随时会撤的虚单，五档数据根本看不出来。这时候就需要 Level-2 的逐笔委托和委托队列——每一笔挂单、撤单都能看到，封单的真实性才有判断依据。这份《同花顺 level-2 全推行情 Python API 操作手册》围绕的就是这套数据的接入方式，核心是一个叫 TDS 的行情服务系统，通过 Python SDK 把沪深两市的 Level-1 和 Level-2 行情推送到本地。它适合两类人：一是做量化交易、需要高频行情做因子计算或策略回测的开发者；二是想把逐笔成交、委托队列接进自己系统做实时监控的工程团队。手册覆盖了从环境搭建、回调注册、订阅管理到级联分发和离线调试的完整链路，不是一份简单的接口列表，更像是一个行情接入项目的工程笔记。

2. TDS 系统架构与 Python SDK 环境搭建

2.1 C/S 架构下的数据链路

TDS 整个系统是 C/S 结构，数据从交易所的 VDE 接入，经过解码、订阅分发，最终以 API 形式对外提供服务。整条链路可以拆成三层：接入层负责从行情源拿原始数据，处理层做解码和订阅匹配，API 层通过 SDK 把数据推给应用。SDK 本身分 Level-1 和 Level-2 两个版本，Level-1 提供五档行情快照、逐笔成交和逐笔委托；Level-2 在此基础上多了十档行情快照、委托队列和极速盘口。这里有个容易忽略的点：极速盘口需要额外产品权限，不是装了 SDK 就能拿到的。

从数据内容来看，沪深两个市场的支持范围基本对齐，但字段细节有差异。比如上交所的证券快照有买入撤单笔数、撤单数量、撤单金额这些字段，深交所没有；深交所的匹配成交量、匹配成交金额、成交方式这些字段，上交所也不提供。写策略的时候如果要做跨市场统一处理，这些字段差异得在数据层先抹平。

2.2 安装与环境依赖

运行环境的要求不算低。硬件方面需要 64 位双核以上处理器、8G 以上内存，互联网带宽下行 100Mb 以上——行情数据是持续推送的，带宽不够会直接导致堵包。操作系统支持 Windows 7 及以上和 CentOS 7 等常用 Linux 发行版，Python 版本要求 3.5 及以上，必须是 64 位。

安装方式有两种，取决于拿到的是 tar.gz 包还是源码目录：

# 方式一：直接 pip 安装 tar.gz 包 pip install ./L2-tdsapi-python-X.X.X-XXXXXXXX.tar.gz # 方式二：进入源码目录用 setup.py 安装 python setup.py install

Linux 系统下安装需要 root 权限，这一点在部署到服务器时要注意。安装完成后验证是否成功：

# 进入 Python 环境后执行 import datafeed # 如果没有报异常，说明安装成功

注意：如果 import 报错，先检查 Python 是否为 64 位版本，32 位 Python 无法加载 SDK 的动态库。

2.3 初始化与回调注册

SDK 的使用流程是：创建 TDS 实例 → 定义回调函数 → 设置环境参数并注册回调 → 登录 → 订阅或查询。回调函数是整个 SDK 的核心机制，行情数据和事件通知都通过回调推送到应用层。

from datafeed.tds_api import TDS # 创建实例 tds = TDS() # 定义行情推送回调函数 def callback_stock(data, count, stats): """ @param data: 行情数据列表，可能包含多条 @param count: data 中数据的条数 @param stats: 传输链路时间戳字典 """ print("receive count:", count) print("receive data:", data) # 注册回调函数和设定环境参数 tds.set_env(fnctickstock=callback_stock)

set_env是必须第一个调用的函数，它负责注册所有回调并设置运行参数。除了fnctickstock（证券快照回调），还可以注册fnctickindex（指数快照）、fncorderqueue（委托队列）、fncorderdata（逐笔委托）、fnctransaction（逐笔成交）、fncsuperstock（极速盘口）这五个数据回调，以及fncnotify（断线通知）和fucsubscribe（异步订阅回调）。

set_env还接受几个关键参数：m_heartbeat_interval控制心跳间隔（秒），m_reconn_count设置断线重试次数，m_reconn_interval设置重连间隔，m_timeout是同步请求超时时间（毫秒，默认 5000），m_connectmarket指定连接市场（默认 "USH;USZ"，可以只填 "USZ" 或 "USH" 连单市场），m_connectsubscribe设为 1 时断线重连成功后会自动重发订阅请求。m_logpath和m_loglevel分别控制日志路径和级别。

这里有个血泪经验：回调函数里千万不要做耗时操作。SDK 的数据推送是持续不断的，回调里如果做数据库写入、网络请求这类阻塞操作，很容易导致接收行情堵包，后面的数据全卡在缓冲区里。

3. 登录、查询与订阅的完整操作链路

3.1 登录与错误码排查

登录接口很直接，传入账号密码，成功返回 0，失败返回负数：

# 登录 ret = tds.login('test1', '123456') if ret != 0: print("login failed, code:", ret)

返回值里有一批错误码需要记住，排查问题时能省很多时间。-1004是用户名密码错误，-1005是无法连接至服务器，-1003是 IP 被禁止访问，-1016是重复登录，-1018是版本太低需要更新程序。特别说一下-1018，这个错误码是 V2.1.3 版本才加进去的，之前版本过低时只会返回一个笼统的错误，根本不知道是版本问题。如果遇到登录一直失败但账号密码确认没问题，先检查 SDK 版本是不是太旧了。

3.2 同步查询接口

查询类接口是同步的，调用后直接返回结果，适合在启动阶段拉取基础数据：

# 获取 USHI 市场的所有代码列表 code_list = tds.get_code_list('USHI') # 获取 USHI 市场的所有指数快照 index_data = tds.get_tick_index('USHI', []) # 获取 USHA 市场 600000 的证券快照 stock_data = tds.get_tick_stock('USHA', code_list=['600000']) # 获取 USHA 市场 600000 的极速盘口 super_data = tds.get_super_stock('USHA', code_list=['600000'])

get_code_list传入市场代码返回该市场的代码列表，get_tick_index和get_tick_stock分别获取指数和证券快照，code_list传空列表[]表示全市场。get_super_stock获取极速盘口数据，同样需要产品权限。

市场代码的命名规则需要记一下：上交所是 USH 开头，USHI 是上海指数、USHA 是上海 A 股、USHB 是上海 B 股、USHD 是上海债券、USHJ 是上海基金、USHP 是上海退市整理、USHT 是上海 ST 风险警示板。深交所对应 USZ 开头，USZI 深圳指数、USZA 深圳 A 股、USZB 深圳 B 股、USZD 深圳债券、USZJ 深圳基金、USZP 深圳退市整理。注意上证 A 股还包括 USHT 这个风险警示板市场，做全市场扫描的时候别漏了。

3.3 订阅管理与异步订阅

订阅接口控制实时行情的推送范围，有四个同步接口和一个异步接口：

# 订阅 USHA 全市场证券快照（取消原有订阅并重新订阅） tds.sub('stock', 'USHA', []) # 取消 USHA 快照的所有订阅 tds.unsub('stock', 'USHA') # 增加 USHA 全市场逐笔成交的订阅 tds.add_sub('trans', 'USHA', []) # 取消 USHA 全市场快照的订阅 tds.del_sub('stock', 'USHA')

sub是取消指定市场指定数据类型的所有订阅后重新订阅，unsub是取消所有订阅，add_sub是在已有订阅基础上增加代码，del_sub是取消指定代码的订阅。data_type支持 index、stock、queue、order、trans、superstock 六种。

异步订阅接口async_sub是 V2.1.4 版本加入的，调用后不阻塞，结果通过fucsubscribe回调通知：

def on_subscribe(data_type, op_type, market, code_list, statuscode, callback_type): """ @param data_type: 订阅的数据类型 @param op_type: 操作类型，1-全新订阅 2-取消订阅 3-增加订阅代码 4-取消订阅代码 @param market: 市场代码 @param code_list: 订阅的代码列表 @param statuscode: 订阅状态，0 为成功 @param callback_type: 1-异步订阅请求 2-断线重连后自动重发 """ print("subscribe status:", statuscode) tds.set_env(fucsubscribe=on_subscribe) tds.async_sub('stock', 1, 'USHA', [])

异步订阅的好处是不用等同步返回，在需要批量订阅多个市场时效率更高。callback_type参数区分了是主动调用异步订阅触发的回调，还是断线重连后自动重发订阅触发的回调，排查订阅问题时这个参数很有用。

4. 级联模式与离线调试：多实例部署和本地回放

4.1 级联模式的主从配置

级联模式解决的是一个很实际的问题：当需要运行多个 SDK 实例时，如果每个实例都去连 datafeed 服务器，不仅浪费连接资源，还可能触发服务器的连接数限制。级联模式让主 SDK 负责连接服务器和接收数据，从 SDK 连主 SDK 获取数据，在局域网内做数据分发。

配置分两步。第一步是新建tool.ini配置文件，Python 版本放在程序文件同级目录：

; 主 SDK 配置 master=true socket_port=9999 ; 从 SDK 配置 master=false socket_port=9998

master字段区分主从属性，socket_port是监听端口。如果主从在同一台机器上，端口必须不同。这里有个坑：端口号后面必须接换行符，否则配置文件读取会失败，表现为主 SDK 启动后从 SDK 连不上，但日志里看不出什么明确错误。

第二步是从 SDK 调用register_addr注册主 SDK 的地址：

from datafeed.tds_api import TDS tds = TDS() def callback_stock(data, count, stats): print("receive count:", count) tds.set_env(fnctickstock=callback_stock) # 从 SDK 注册连接的 IP 和端口 tds.register_addr('master', 'xxx.xxx.xxx.xxx', 9999) # 登录，从 SDK 的账号密码需要与主 SDK 保持一致 tds.login('test1', '123456')

register_addr必须在登录前调用，而且必须在主 SDK 连接登录成功之后才能进行从 SDK 的连接。从 SDK 的账号密码要和主 SDK 一致，否则登录会失败。级联模式下主从 SDK 必须在同一局域网内，跨网段是不行的。

4.2 离线调试模式的数据回放

离线调试模式是 V2.1.0 版本加入的功能，对于策略回测和本地开发非常实用。它的机制是读取本地 CSV 文件来驱动 SDK 运行，调用后登录不连接线上网关，直接返回成功，完全保持离线状态。

from datafeed.tds_api import TDS tds = TDS() def callback_stock(data, count, stats): print("receive count:", count) print("receive data:", data) tds.set_env(fnctickstock=callback_stock) # 进入离线调试模式 tds.start_offline_debug( path='D:\\datafeed', subtype=['stock', 'index', 'order', 'queue', 'trans', 'superstock'], push_space=3000, # 回放推送间隔，单位毫秒 push_count=5000, # 单次推送条数 playback_times=1 # 数据文件遍历次数 ) # 登录，但不会连接 server tds.login('test1', '123456')

数据文件有命名规范，必须是stock.csv、index.csv、order.csv、queue.csv、trans.csv、superstock.csv、codelist.ini其中之一，SDK 根据文件名确定数据类型，命名不对的文件不会被读取。CSV 表头字段名称要参照模板，顺序不强制。

离线调试的规则值得细看：调用查询股票快照的函数时，SDK 读取对应数据文件的前 3000 条（指数是前 300 条），剔除重复代码后生成一份供查询的数据；调用订阅接口返回订阅成功，SDK 按文件顺序读取，每次读指定条数，间隔指定时间再读，文件遍历完切换到下一个文件，所有文件完成指定循环次数后调试结束。每次读取生成的数据包会根据类型和订阅代码触发对应回调，如果当次数据不包含订阅代码，可能不触发回调——这个机制和实时行情是一样的。

注意：级联模式下父节点支持离线调试，启动后子节点可以获取数据，子节点不需要开启离线调试。子节点本身不支持离线调试，调用该函数无效。

5. 避坑与常见问题排查

5.1 回调堵包导致行情延迟

现象：行情推送越来越慢，最终完全停止，日志里没有报错。

原因：回调函数里做了耗时操作，比如写数据库、发 HTTP 请求、做复杂计算。SDK 的数据推送是持续不断的，回调处理速度跟不上推送速度时，数据在缓冲区堆积，最终堵死。

解决：回调函数里只做数据入队操作，把数据丢到内存队列后立即返回，另起线程或进程从队列消费做后续处理。队列满了就丢弃旧数据，保证实时性优先。

5.2 登录返回 -1018 版本过低

现象：账号密码确认无误，但登录一直失败，返回 -1018。

原因：SDK 版本太旧，服务端已经不支持。这个错误码是 V2.1.3 版本才加入的，之前版本过低时只返回笼统错误。

解决：更新到最新版本的 SDK 包，重新安装。如果更新后仍然报错，检查 Python 是否为 64 位版本。

5.3 级联模式从 SDK 连不上主 SDK

现象：主 SDK 登录成功，从 SDK 调用register_addr后登录失败。

原因：常见的有三种——tool.ini里socket_port后面没有换行符导致配置读取失败；主从 SDK 不在同一局域网；从 SDK 的账号密码和主 SDK 不一致。

解决：检查tool.ini文件格式，确保端口号后有换行符；确认主从机器在同一网段；核对从 SDK 登录用的账号密码是否与主 SDK 完全相同。

5.4 订阅成功但收不到数据

现象：sub返回 0 表示成功，但回调函数一直不触发。

原因：可能是市场代码填错了，比如把 USHA 写成了 USH；也可能是订阅的代码不在该市场下，比如用 USHA 订阅了一个深交所的代码；还有一种情况是极速盘口没有产品权限，订阅返回成功但实际不推送数据。

解决：先用get_code_list确认市场代码和代码列表是否正确；极速盘口需要确认是否已开通对应权限；检查set_env里m_connectmarket是否限制了连接市场，如果只填了 "USZ" 但订阅的是上交所数据，自然收不到。

5.5 断线重连后订阅丢失

现象：网络波动导致断线，重连成功后行情不再推送。

原因：m_connectsubscribe参数没有设置为 1，断线重连后不会自动重发订阅请求。

解决：在set_env中设置m_connectsubscribe=1，这样断线重新登录成功后会自动重发订阅信息。同时注册fncnotify回调，当收到断线通知时（code 为 1 表示网络断开，2 表示网卡不正常，3 表示重复登录被踢下线），根据情况决定是否重新调用 Login 和订阅。

6. 二次开发：从数据接入到策略落地的改造思路

拿到 SDK 之后，大多数人的第一反应是先把数据接进来看看长什么样，但真正要落地到策略里，中间还有不少工程工作要做。我一般会按这样的顺序来改造。

第一步是封装数据层。SDK 回调推过来的data_list是原始字典列表，字段多且沪深有差异。我会先写一个适配层，把沪深字段统一成内部标准格式，缺失的字段补默认值。比如上交所快照没有matching_volume，深交所没有withdraw_buynum，统一处理后在策略层就不用关心数据来源了。

# 数据适配层示例 FIELD_MAP = { 'market': str, 'code': str, 'new_price': float, 'volume': int, 'turnover': float, 'bidorder_price': list, 'askorder_price': list, 'bid_numorders': list, 'ask_numorders': list, } def normalize_snapshot(raw: dict, market: str) -> dict: """将沪深快照统一为标准格式""" result = {} for field, dtype in FIELD_MAP.items(): val = raw.get(field) if val is None: # 深交所没有的字段补默认值 result[field] = dtype() if dtype not in (list,) else [] else: result[field] = dtype(val) result['_source'] = market return result

第二步是设计数据管道。回调里只做入队，消费端做解析、适配、分发。我一般用queue.Queue做缓冲，消费线程按数据类型分发给不同的处理器。快照数据用于计算盘口指标，逐笔成交用于计算主动买卖量，逐笔委托和委托队列用于分析封单真实性。

第三步是涨停板策略的关键数据点。判断封单质量需要几个维度的数据：委托队列里的买一挂单量、逐笔委托里的买入委托量和撤单量、逐笔成交里的主动买入量。如果买一挂单量很大但逐笔委托里频繁出现大额撤单，说明封单不牢；如果主动买入量持续大于卖出量，说明买盘积极。这些判断都需要 Level-2 的逐笔数据支撑，五档行情是做不了的。

第四步是离线调试做回测。把历史行情数据按 SDK 要求的 CSV 格式准备好，用start_offline_debug回放，策略逻辑完全不用改，回调触发机制和实时行情一致。这样可以在不连行情服务器的情况下验证策略逻辑，调试成本低很多。

有一个习惯我后来一直保持：每次改完回调逻辑，先在离线调试模式下跑一遍完整回放，确认数据解析和策略信号都正常，再切到实时行情。因为实时行情出问题的时候，排查成本比离线高太多了——你没法暂停行情等你看日志。从那以后我每次动回调相关的代码，都强制走一遍离线回放验证，这个习惯帮我省了不少线上排查的时间。希望帮到你。

本文还有配套的精品资源，点击获取