我本人做A股量化也有五年多了从最早靠手工在通达信里翻了半天数据到后来被迫自己写脚本拉数据、跑策略、盯盘算是把行情数据获取这条路上的坑基本踩了个遍。今天想认真聊聊一个很实用的组合Python配合pytdx库直接对接通达信的行情协议高效拉取分笔成交数据。这玩意儿在量化交易里是硬需求不管是做高频策略、盘口分析还是盯主力资金的动向没有tick级数据基本就是盲人摸象。这篇文章既适合刚接触Python量化、还在为数据源发愁的新手也适合已经在用日线级别数据、但想进一步做精细分析的老手。我尽量把我在实际项目中用到的完整方案、踩过的坑、以及常见的排查思路都整理出来你可以把它当成一份可以直接参考的实操笔记。1. 分笔成交数据到底能做什么1.1 先搞清楚你拿到的是什么在通达信里按F1或者切到分时成交明细你能看到一条条逐笔成交记录包括成交时间、成交价格、成交手数、买卖性质等这就是分笔成交数据业内一般叫Tick数据。很多人一开始做量化习惯直接用日K线或者分钟K线无论从哪个量化平台拿本质上都是已经聚合好的数据。K线当然能做很多事情但它把一整天里面几十万笔交易的细节给压扁了。你只看到开盘价、收盘价、最高最低价和成交量至于这些量是主动买出来的还是主动卖出来的盘口发生过什么变化完全看不出来。分笔数据就不一样了。每一笔都是最原始的成交记录拿它做回测和策略分析能还原市场的真实博弈过程。举个例子一根大阳线收盘你看日K只知道今天涨了但如果你看分笔数据发现绝大部分成交都是尾盘最后五分钟突然打出来的这可能意味着有资金刻意拉抬也可能只是一笔乌龙指两者的后续含义完全相反。1.2 哪些策略依赖tick级别的信息我把量化的常见需求按数据粒度分个层这样你就能直观理解分笔数据在其中的定位日线/周线级适合长线选股、基本面因子分析数据量小免费源一大堆。分钟级适合日内波段、短线择时普通行情接口就能拿到。分笔级Tick级适合高频统计、盘口异动捕捉、主力行为分析、精准订单流数据研究。具体来说分笔成交数据的典型应用场景包括计算主动买盘与主动卖盘的比例资金流向的微观基础。捕捉大单成交异动比如突然出现一笔或多笔巨量主动买往往是行情启动的前兆。用分笔数据自行聚合出任意时间周期的K线比如你自己定义的2分钟、5分钟、特殊时段K线平台给不了这么灵活。研究开盘前和收盘后集合竞价阶段的成交特征。建立盘口冲击成本模型模拟自己下单后的市场影响。这些玩法用日线数据是完全做不了的。如果你对市场的理解还想往深处走一层分笔数据这个坎迟早要过。2. 为什么是pytdx而不是其他数据源2.1 pytdx的来历与定位pytdx是一个纯Python实现的通达信行情接口库底层直接解析通达信客户端使用的通信协议通过TCP连接到通达信行情服务器获取实时行情数据。它不是通过爬虫抓网页也不是通过模拟鼠标点击操作通达信界面而是直接走协议层通信。这意味着它获取数据的速度快、格式稳定、获取字段丰富而且不依赖操作系统里面的任何桌面软件。这三年我在好几个行情数据源之间来回切换过pytdx给我最大的感觉是——轻巧、稳定、免费。没有复杂的认证流程装好库、填上服务器地址、建立一个连接就能拉数据对个人量化玩家来说极其友好。2.2 为什么不直接读通达信本地文件有的朋友可能知道通达信客户端会在本地电脑存一份完整的历史分笔数据文件后缀是.lc1或.lc5理论上可以用Python直接解析这些二进制文件。我早期也走过这条路但很快就放弃了原因很现实通达信版本的更新可能导致本地数据文件结构变化解析代码随时作废。必须先手动打开通达信客户端还得确保数据完整下载一旦漏了某几天的数据你根本不知道。本地文件读取不适合做实时监控你没法在盘中快速拿到刚刚那一秒的最新成交。pytdx直接请求服务器数据从根本上绕开了这些问题。它给你的是纯数据接口拿到就能用不用关心底层文件长什么样。2.3 和其他免费数据源对比市面上做量化数据服务的厂商很多有免费的但也有限制有的需要注册Token有的一天只能拉几次有的数据有延迟。pytdx最核心的竞争力在于数据是实时直接从通达信行情服务器取的不走第三方中转。你不需要购买任何付费服务也不会有明显的延迟特别适合做盘中实时分析。当然它也有自己的限制比如行情服务器偶尔会断连、请求频率太高会被暂时拒绝、历史分笔数据一般只能回补到最近几天这受限于通达信服务器的数据存储策略。但对于绝大多数个人量化需求来说它足够用了。整体选型思路很简单如果核心需求是拿真实、及时的A股逐笔成交数据且想省掉付费数据源的成本pytdx几乎是当下最优解。3. 环境准备安装与连接测试3.1 安装pytdxpytdx库已经发布在PyPI上安装非常方便直接用pip命令pip install pytdx如果你的网络环境访问默认PyPI源速度较慢可以换用国内镜像源pip install pytdx -i https://pypi.tuna.tsinghua.edu.cn/simple安装完成后你可以在Python交互环境里验证一下是否导入成功from pytdx.hq import TdxHq_API api TdxHq_API() print(pytdx导入成功)如果这段代码不报错说明库已经装好可以继续玩连接了。3.2 选择行情服务器并建立连接pytdx获取行情数据需要先连接一台通达信行情服务器。通达信官方客户端里有大量的行情服务器地址网上也能搜到很多公共服务器列表。我一般会把多个服务器地址维护在一个列表里连接时按顺序尝试优先连速度最快的from pytdx.hq import TdxHq_API # 收集了一些常用的通达信行情服务器地址 servers [ (119.147.212.81, 7709), (101.227.73.20, 7709), (101.227.77.254, 7709), (114.80.63.12, 7709), (123.125.108.14, 7709), ] def create_connection(): api TdxHq_API() for ip, port in servers: try: api.connect(ip, port) print(f连接成功: {ip}:{port}) return api except Exception as e: print(f连接失败: {ip}:{port}, 错误: {e}) raise ConnectionError(所有服务器均连接失败) api create_connection()这里有几个细节说明一下通达信服务器默认端口基本都是7709。行情服务器有负载和稳定性差异。有时候不是你的代码有问题而是你连的这台服务器刚好在维护换个地址就好了。建议把服务器列表写成一个配置文件方便随时增删而不是硬编码在代码里。3.3 用get_security_quotes快速验证连通状态连接成功后可以先用一个最简单的接口测试一下能否正常取数比如查询一只股票的实时快照# 获取平安银行的最新行情快照 # 参数格式里第一个元素是市场代码0代表深圳1代表上海 quotes api.get_security_quotes([(0, 000001)]) if quotes: print(quotes) else: print(未获取到数据请检查连接或参数)返回结果里会包含最新价、涨跌幅、成交量、买卖五档等字段。如果这一步顺利说明连接没问题可以进入正式的取数环节了。4. 核心环节获取分笔成交数据的完整实现4.1 认识通达信的基础市场代码在pytdx里几乎所有的行情接口都要求传入一个市场代码加股票代码的组合。这个设计经常让新手懵圈所以我把常见的市场代码整理了一下市场代码对应市场常见品种0深圳证券交易所000、001、002、003开头股票、深市可转债1上海证券交易所600、601、603、605、688开头股票、沪市可转债这里有个非常容易踩的坑股票代码需要传字符串并且要补足6位比如平安银行是“000001”而不是“1”。同时市场代码不能搞混同一个代码在不同市场可能存在但实际含义完全不同。4.2 核心接口get_transaction_data详解获取分笔成交数据pytdx里最常用的接口是get_transaction_data。它的完整签名如下api.get_transaction_data(market, code, start, count)四个参数的意思分别是market市场代码0代表深圳1代表上海。code6位数字股票代码字符串形式。start起始偏移量从0开始。0表示从最新的那一笔开始往后取。count本次要取多少笔最大不能超过800。这里有一个关键逻辑必须理解start不是时间而是笔数偏移量。每次请求最多返回800笔如果当天的成交笔数超过800笔你想获取更早的数据就要利用上一次返回结果里的最后一条记录继续往前翻。4.3 单次获取并解析返回字段先演示一下最基础的调用方法# 获取平安银行当天最新的50笔分笔成交 data api.get_transaction_data(0, 000001, 0, 50) for item in data: print(item)返回的结果是一个列表每个元素对应一笔成交包含的字段通常有time成交时间格式是HHMMSS比如90532代表09:05:32。price成交价格单位是元。vol成交手数注意不是股数A股一手等于100股。buyorsell成交性质0代表买入1代表卖出2代表中性盘集合竞价等。num这笔成交内部包含的明细笔数大单合并时这个值大于1。我在做交易行为分析的时候最依赖的就是buyorsell这个字段。它区分了每笔成交是由买方主动发起的还是卖方主动发起的这构成了资金净流入流出的微观基础。4.4 突破800笔上限循环翻页获取全量数据A股一天下来活跃股票的成交笔数动辄几万笔活跃度高的甚至超过十万笔。单次接口调用拿不到完整数据必须通过循环逐页拉取。这里我直接贴一段自用的翻页函数from pytdx.hq import TdxHq_API # 建立连接服务器列表部分省略参考上文 api create_connection() def get_all_transactions(market, code): all_data [] start 0 while True: batch [] try: batch api.get_transaction_data(market, code, start, 800) except Exception as e: print(f拉取数据出错: {e}) break if not batch: break all_data.extend(batch) print(f已获取 {len(all_data)} 笔) # 如果本次返回不足800笔说明已经到最早的边界了 if len(batch) 800: break # 每页800笔下一次从start800继续往后取 start 800 return all_data # 获取今天平安银行的全部分笔成交 df get_all_transactions(0, 000001) print(f总共获取 {len(df)} 笔成交)这段代码有几个细节值得展开说判断结束的条件是len(batch) 800。因为接口最多返回800笔只有返回数量不足800笔时才说明已经拉到了最早那批数据。翻页的过程中不需要任何延时吗原则上短时间内的连续请求一般不会被服务器拒绝但如果你的循环很激进比如每次都立刻请求下一页且长期高频运行服务器可能临时断掉你的连接。稳妥起见我通常在每页之间加一个极短的sleep(0.05)。如果你只需要获取最近一分钟的实时成交那直接调用一次接口即可非常快。4.5 一个隐蔽的大坑成交量字段是累计值这个坑我当年踩得刻骨铭心必须单独拿出来讲。在get_transaction_data返回的数据里vol字段并不是单笔成交的手数而是从开盘到该笔成交时刻的累计成交量单位是手。如果你直接拿它当单笔成交量用后续统计一定会出错。为什么接口要这么设计因为通达信客户端的分笔成交显示有的是逐笔推送有的是某个时间点合并后的状态用累计值可以在客户端断线重连后自动恢复状态。正确的处理方法很简单自己做一个差分# 将累计成交量转换为单笔成交量 for i in range(len(data) - 1, 0, -1): data[i][vol_single] data[i][vol] - data[i - 1][vol] data[0][vol_single] data[0][vol]差分逻辑是从后往前减因为前面的数据是累计值后面的是之后的累计值两者相减就是中间新增的那部分成交。如果不做差分你统计出来的成交量会随着时间递增而且会严重偏大策略信号会被干扰得面目全非。5. 实战用分笔数据重构高级分析指标5.1 从分笔数据聚合出任意的分钟K线有了全天的分笔成交明细你就可以自己做主把数据聚合成任意周期的K线。平台通常只提供1分钟、5分钟、15分钟、30分钟、60分钟你想看3分钟K线平台可能没有但用分笔数据自己聚合几分钟就能搞定。聚合逻辑是先按时间把分笔数据分组再计算每组内的高开低收和总成交量。import pandas as pd def aggregate_kline(transactions, period_minutes3): df pd.DataFrame(transactions) df[datetime] pd.to_datetime(df[time].astype(str).str.zfill(6), format%H%M%S) # 设置时间锚点以当天9:30为基准 base pd.Timestamp(2024-01-01 09:30:00) df[bin] ((df[datetime] - base).dt.total_seconds() // (period_minutes * 60)) # 这里省略日期部分实际使用时需要拼接当天日期 grouped df.groupby(bin).agg( open(price, first), high(price, max), low(price, min), close(price, last), vol(vol, last) # 由于vol是累计值这里取组内最后一笔的累计值 ) return grouped # 示例获取分笔数据后自定义聚合 transactions get_all_transactions(0, 000001) kline_3min aggregate_kline(transactions, 3) print(kline_3min.head())有一点要注意在聚合时我直接用组内最后一笔的vol作为该K线的总量因为它本身是累计值。如果你之前做了差分操作那聚合时就要改为sum。5.2 计算主动买卖力量比平时我们看到的资金流向指标绝大多数平台都是基于分笔数据中的买卖性质算出来的。自己也完全可以实现一个而且可以做到完全透明、可控。基本原理是把每笔成交按buyorsell分成主动买入和主动卖出两类分别统计成交手数然后计算占比。def calc_buy_sell_ratio(transactions): total_buy 0 total_sell 0 for item in transactions: # 差分得到单笔成交量 vol_single item[vol_single] if item[buyorsell] 0: total_buy vol_single elif item[buyorsell] 1: total_sell vol_single if total_buy total_sell 0: return 0.5 return total_buy / (total_buy total_sell) # 假设 transactions 已经做好差分 ratio calc_buy_sell_ratio(transactions) print(f主动买入占比: {ratio:.2%})主动买入占比超过50%说明当天的市场成交更多由买方主导反之则卖方更激进。这个指标本身并不复杂但它可以帮你验证很多技术面信号的真伪。比如某只股票放量上涨但主动买入占比却很低这就说明上涨可能是卖压减轻而非买盘增强后劲存疑。5.3 监控大单异动的实时告警逻辑如果只是做盘后分析拿到数据就够了。但在盘中做监控才是pytdx真正的用武之地。我的一套简化实时监控逻辑是这样的每3秒拉取一次最近的分笔数据与上次记录的累计成交量做增量比较一旦增量里出现超大单就触发一次告警。import time last_vol 0 def monitor_big_order(market, code, threshold5000): global last_vol while True: try: data api.get_transaction_data(market, code, 0, 10) if data: # 最新一笔的累计成交量 current_vol data[0][vol] # 新增成交量 diff current_vol - last_vol if last_vol 0 and diff threshold: print(f检测到大单: 新增{diff}手, 最新价{data[0][price]}) last_vol current_vol except Exception as e: print(f监控出错: {e}) time.sleep(3)这个策略的缺点是如果单笔大单在两次轮询之间被合并了单看增量可能不够灵敏。更精细的做法是每次多取几十笔然后逐笔检查差分后的单笔成交量是否超过阈值。不管用哪种方式本质上都是先用pytdx快速获取最新分笔数据再做增量计算和规则判断。6. 常见问题与排查技巧实录6.1 返回空数据的几个原因get_transaction_data返回空列表的情况多半是以下几个原因非交易时段。周末、节假日、午间休市时段服务器不会返回分笔数据这是正常的。股票代码或市场代码传错。比如深市股票传了market1肯定查不到。选择的服务器没有该股票的数据权限少见但存在换个服务器试试。新股或停牌股当天没有成交也会返回空列表。排查思路就是先拿一只极其活跃的股票比如贵州茅台、宁德时代做测试如果活跃股票能返回数据、你的目标股票不能那基本可以确定是股票自身或参数的问题而不是库或服务器的问题。6.2 连接经常断开怎么办长时间运行pytdx的进程尤其是那种每隔几秒请求一次的监控程序经常会出现连接被服务器断开的情况。这其实不是因为pytdx的问题而是通达信服务器资源有限会主动回收空闲或不活跃的连接。我的解决办法是封装一个带有自动重连机制的客户端类import time from pytdx.hq import TdxHq_API class TdxClient: def __init__(self, servers): self.servers servers self.api None def connect(self): for ip, port in self.servers: api TdxHq_API() try: api.connect(ip, port) self.api api return True except Exception: continue return False def request(self, func, *args, **kwargs): # 带重试机制的请求封装 for attempt in range(3): try: if self.api is None: self.connect() result getattr(self.api, func)(*args, **kwargs) return result except Exception as e: print(f请求异常尝试重连 ({attempt 1}/3): {e}) time.sleep(1) self.connect() raise Exception(多次请求失败) client TdxClient(servers) # 使用示例 data client.request(get_transaction_data, 0, 000001, 0, 100)每次请求失败会自动换一台服务器重连重试3次之后再报错。这样处理之后脚本连续跑一整天基本不会中断。6.3 历史分笔数据只能取到最近几天有朋友问过我能不能用pytdx一次性把过去一年的分笔数据全部拉下来实际不太行。通达信行情服务器一般只保留最近5个交易日左右的分笔数据更早的数据需要通过客户端手动下载或者使用其他历史库。这是服务器端的存储策略pytdx本身也爱莫能助。如果你确实需要完整的历史分笔数据目前比较靠谱的做法是运行一个常驻脚本每天收盘后定时拉取当天的分笔数据保存到本地数据库。日积月累自己攒一个历史分笔数据库。我现在就是这样做的每天下午收盘后跑一次把全市场重点股票的分笔数据存进PostgreSQL一个月下来也有不小的数据量但检索非常方便。6.4 关于请求频率的注意事项虽然pytdx没有在代码层面强制限流但通达信服务器端是有风控的。如果请求频率过高可能会被临时封IP一段时间表现就是所有请求都失败。我的经验是实时监控场景每3到5秒请求一次比较稳妥。批量拉取历史分笔数据时每页之间建议间隔0.1秒左右。尽量避免在毫秒级间隔内连续发几十个请求。说白了就是别把免费服务器薅太狠大家互相留点余地长期用才稳。7. 把pytdx嵌进你的量化交易系统7.1 数据落地存成什么格式数据拉到之后不能每次都用的时候现拉最好落到本地。常用方案有这么几种CSV文件最简单但数据多了以后检索慢不适合做复杂查询。SQLite单文件数据库适合个人项目轻量好用。PostgreSQL/MySQL适合多用户、大数据量、并发查询我目前主力使用PostgreSQL。我推荐直接上PostgreSQL或者SQLite因为分笔数据天然适合按时间排序存储而且后续做聚合查询、回测取数都很方便。以SQLite为例建表语句非常简单CREATE TABLE IF NOT EXISTS tick_data ( market INTEGER, code TEXT, time TEXT, price REAL, vol INTEGER, buyorsell INTEGER, date TEXT, PRIMARY KEY (market, code, date, time) );存储时加上date字段按交易日分区存储查询时直接按日期过滤。7.2 定时任务每天自动更新用系统自带的定时任务工具把每日数据下载脚本挂上去就能实现数据自动积累Linux/macOS用户用crontab。Windows用户用任务计划程序。我的做法是每天下午15:10分执行一次下载脚本把当天全部分笔数据导入数据库# 每天15:10分执行数据下载任务 10 15 * * 1-5 cd /path/to/project python daily_tick_downloader.py logs/daily.log 21这样经过一段时间的积累你就拥有了属于自己的分笔历史数据库想怎么分析都行。7.3 策略模块如何调用这些数据数据进库之后策略模块就可以直接通过SQL查询来获取所需数据。比如想读某只股票某天的全部分笔数据import sqlite3 conn sqlite3.connect(tick_data.db) def load_tick_data(code, date): sql SELECT time, price, vol, buyorsell FROM tick_data WHERE code ? AND date ? ORDER BY time df pd.read_sql_query(sql, conn, params(code, date)) return df # 加载某只股票某天的数据 df load_tick_data(000001, 2024-12-20) print(df.head())只要保证数据能稳定入库策略开发就完全可以在本地快速迭代不再依赖任何实时接口回测和实盘之间的数据链路也能做到完全一致。8. 最后分享一点实操体会分笔数据的价值很多时候不是直接拿来当买入信号的它更像是给你配备了一台显微镜让你能看清市场内部那些粗粒度数据看不到的细节。pytdx本身的学习成本非常低真正需要花时间的是理解每一笔成交背后的市场含义。同样的数据在不同的人手里能改造成完全不同的东西有人用它做高频统计套利有人拿它监控主力动向也有人只是拿来验证自己已有的交易想法。几年下来我的习惯是任何新策略上线前都会先用pytdx拉一段时间的分笔数据做离线测算。先确认数据层面信号可靠再考虑上模拟盘和实盘。这个习惯帮我避免过很多次纸上谈兵式的策略翻车。如果你刚开始接触这块建议先从拉取一只股票的全天分笔数据开始把数据格式、字段含义都摸透再用它做一次简单的买卖力量分析。等这套流程跑顺了后面再加实时监控、自动入库、策略回测都是水到渠成的事。