做日内量价策略那阵子我被“怎么拿分笔成交数据”卡了整整两天。市面上Level-2数据动辄几千一年免费的API要么只给K线不给分笔要么封装得太重装完依赖比策略代码还长。后来一个做短线量化的朋友提了一句“通达信协议有人封装过了叫pytdx分笔能直接拉。”我试了一下午第一笔真实分笔数据落进CSV的时候整个策略的验证路子瞬间就通了。这篇文章就围绕pytdx怎么高效获取通达信分笔成交数据把选型理由、连接细节、核心函数边界、数据落地和坑点一次讲透。适合刚接触量化、想拿真实盘口数据练手的Python开发者也适合已经在做策略回测、但觉得K线粒度不够想往订单流和资金流向方向深挖的朋友。我会直接给代码、给参数、给实测中踩过的坑你拿到就能跑。1. 分笔成交数据到底解决什么问题为什么不用K线和分钟线1.1 K线是结果分笔才是过程K线和分钟线解决的是“价格变成什么样子”分笔成交数据解决的是“价格是怎么一步步变成这个样子的”。这两件事在量化研究里完全是不同维度。举个例子某只票在14:55还是平开震荡尾盘突然一根长下影线收回1分钟K线上只显示价格从-3%快速拉回0%。但如果你看分笔数据会发现整个过程可能是5笔800手的大单连续主动买入砸出来的每笔的时间、价格、成交量、主买主卖方向清清楚楚。对短线策略来说这种“过程信息”才是真正决定第二天情绪和资金意图的东西。做资金流统计、主力异动追踪、冲击成本测算、盘口微观结构分析都需要分笔粒度。K线最多告诉你“发生了什么”分笔能告诉你“谁在什么价位做了什么动作”。这也是为什么很多游资和私募的看盘系统核心盯的永远是分笔而不是收盘K线。1.2 行情源选型为什么最终落在pytdx上在确定pytdx之前我把市面上常见的数据获取方式筛了一遍按“是否免费、是否支持分笔、接入成本”三个维度做了个对比数据源是否免费分笔支持接入成本稳定性与限制通达信协议直连pytdx免费支持实时低pip安装即用依赖第三方服务器需要容错处理某某Level-2付费行情付费支持中有官方SDK稳定但费用门槛高baostock免费基本不支持低K线和财务数据为主akshare免费部分接口不太稳定中爬虫接口随目标网站改版波动Tushare Pro积分制积分门槛较高中接口稳定但分笔权限要求高pytdx最吸引我的点有三个纯Python实现代码轻量依赖极少直接走通达信行情服务器的通讯协议拿到的是盘中实时数据接口语义非常直接获取分笔就是get_transaction_data一个函数的事不需要理解复杂的协议报文。当然它也有代价服务器不是官方托管的分布式集群连接质量要自己维护请求过频会被断连返回的数据字段注释也不够详细很多细节得靠实测摸索。但对你我这种自己搞研究、不想在数据上花大钱的场景pytdx已经是性价比最高的入口了。2. pytdx连接层细节安装、连接与容错策略2.1 安装和最小可用代码安装没什么好说的一个pip命令pip install pytdx装完之后最基础的连接和数据获取是这样的from pytdx.hq import TdxHq_API api TdxHq_API() # ip和port填通达信行情服务器的地址port固定为7709 connected api.connect(119.147.212.81, 7709) if not connected: print(连接失败) exit() # market: 0表示深圳1表示上海code是6位股票代码字符串 # start: 起始位置从0开始count: 拉取数量单次最大800 data api.get_transaction_data(0, 000001, 0, 30) print(data) api.disconnect()这段代码如果顺利会输出最近30笔成交记录。字段含义后面细说先确保你本地能把数据拉出来。注意connect返回的是布尔值别忽略它很多断连问题都是从这里先暴露的。2.2 行情服务器选择不要死磕单个IPpytdx的config.py里内置了一批公开行情服务器地址我实测下来不同服务器的延迟和数据完整性有明显差异。有些IP响应快拉到一半会断有些响应慢但数据稳定。更麻烦的是某些公网IP可能过一阵就失效了。我的做法是写一个简单的连接池把一批候选服务器测一遍选延迟最低且能成功返回数据的那个import time from pytdx.hq import TdxHq_API from pytdx.config.hosts import hq_hosts def get_available_api(max_retry5): for item in hq_hosts: api TdxHq_API() try: ip item[1] port item[2] if api.connect(ip, port, time_out3): # 用平安银行测一下数据接口是否正常 test api.get_transaction_data(0, 000001, 0, 1) if test is not None: print(f可用服务器: {ip}:{port}) return api else: api.disconnect() except Exception as e: print(f连接 {item[1]} 失败: {e}) continue time.sleep(0.3) return Nonetime_out3这个参数很关键不加的话遇到半死不活的服务器connect会卡住很影响体验。另外pytdx的连接对象不是线程安全的如果要用多线程并发拉多只股票每个线程必须创建独立的连接实例不能共用一个api去并发请求。2.3 频率控制跑太快会被踢这是pytdx最容易被新手忽略的限制。行情服务器对单IP的请求频率有限制如果脚本里用for循环无脑拉几百只股票的分笔大概率在拉第几十只的时候连接就被服务器断掉报ConnectionResetError或者直接返回None。安全起见两个请求之间加time.sleep(0.1)到time.sleep(0.3)。多线程并发时总并发连接数控制在5到8个以内。我实测过单个连接每分钟拉300到500次请求是没问题的超过这个量级就开始不稳定。3. 核心函数实战get_transaction_data的参数边界与数据语义3.1 参数细节market、code、start、count怎么传get_transaction_data(market, code, start, count)是获取分笔成交的核心入口四个参数每一个都有坑。market的值是0或10代表深圳市场1代表上海市场。code是6位字符串比如000001、600000。这里的坑在于判断该传0还是1不能只看代码首位是不是6。科创板688开头、沪市600/601/603/605都属于上海market1而深市主板000/001、中小板002、创业板300、科创板没有301也属于深圳market0。最稳妥的做法是用一个字典或者函数做映射def get_market_by_code(code): if code.startswith((600, 601, 603, 605, 688)): return 1 return 0start是分页位置的偏移量从0开始。count单次最多800条超过800也只返回800条。获取完整分笔需要循环翻页一直翻到返回结果数小于请求数为止。3.2 返回字段拆解time、price、vol、num、buyorsell拿到的一行数据长这样{ time: datetime.datetime(1995, 1, 5, 14, 55), price: 10.25, vol: 800, num: 5, buyorsell: 1 }这里要特别注意pytdx返回的vol单位是股不是手。800代表800股也就是8手。num表示这一笔成交由几笔组成比如800股可能是5笔小单合并的结果。buyorsell字段标记方向0代表中性一般出现在集合竞价或盘口无法判定方向的成交1代表主动买盘2代表主动卖盘。这个字段是后面做资金流向统计的核心依据。所谓“聪明钱”追踪本质上就是统计主动买盘和主动卖盘的成交金额差。3.3 全量分笔拉取翻页循环的标准写法搞清楚参数边界后完整拉取某只股票当天所有分笔的代码就很清晰了import time def fetch_all_transaction(api, market, code, chunk_size800, delay0.1): rows [] start 0 while True: chunk api.get_transaction_data(market, code, start, chunk_size) if not chunk: break rows.extend(chunk) if len(chunk) chunk_size: break start len(chunk) time.sleep(delay) # 温和限速避免被服务器断开 return rows api get_available_api() if api: data fetch_all_transaction(api, 0, 000001) print(f平安银行当日分笔总数: {len(data)}) print(data[-5:]) # 打印最后5笔 api.disconnect()用这个函数一只成交活跃的股票全天分笔大概有几千到几万笔不等拉完也就几秒的事。注意我这里的delay参数之前讲过频率控制在循环翻页里同样适用。3.4 历史分笔get_history_transaction_data的局限除了实时分笔pytdx还提供了get_history_transaction_data可以按日期拉取历史分笔# 获取某只股票2024年1月5日的历史分笔 history_data api.get_history_transaction_data(0, 000001, 0, 800, 2024-01-05)这个接口的date参数格式是2024-01-05注意不同服务器的历史数据保留深度不一样有的只保留最近10来个交易日有的能保留更久。实测下来不要对这个接口的历史深度抱太大期望用于补最近几天的数据是够的想做几个月以上的分笔级回测还是得自己每日落盘积累。4. 数据落地从全量拉取到增量更新4.1 存储格式选择CSV还是SQLite分笔数据一天几万条存储格式用CSV还是SQLite取决于你怎么用。如果只是做盘后分析、回测验证CSV按日存储最简单文件名带上日期和代码比如600000_2024-01-05.csvpandas直接读。缺点是文件数量多做全市场扫描时要遍历一堆文件。如果要做实时盘中监控或者频繁查询某只票某一天的成交明细SQLite更合适。一张表搞定所有股票所有日期按code date建索引查询效率很高。我现在的策略是先全量拉取落到SQLite需要做特征工程时再用pandas拉出来。4.2 增量更新记录最后一条的序号分笔数据是实时增加的每次全量拉一遍既浪费请求次数又容易被服务器限流。更好的做法是记录上次拉到的位置增量拉取新数据。get_transaction_data的start参数是当天分笔的位置偏移所以增量更新的核心就是记住上一次拉到了哪个偏移位置class TransactionRecorder: def __init__(self, db_path): self.conn sqlite3.connect(db_path) self._init_table() def _init_table(self): self.conn.execute( CREATE TABLE IF NOT EXISTS transaction ( code TEXT, date TEXT, offset INTEGER, time TEXT, price REAL, vol INTEGER, num INTEGER, buyorsell INTEGER, PRIMARY KEY (code, date, offset) ) ) def get_last_offset(self, code, date): row self.conn.execute( SELECT MAX(offset) FROM transaction WHERE code? AND date?, (code, date) ).fetchone() return row[0] if row and row[0] is not None else 0 def save_batch(self, code, date, rows, start): data [ (code, date, start i, r[time].strftime(%H:%M:%S), r[price], r[vol], r[num], r[buyorsell]) for i, r in enumerate(rows) ] self.conn.executemany( INSERT OR REPLACE INTO transaction VALUES (?,?,?,?,?,?,?,?), data ) self.conn.commit()盘中写一个定时任务每30秒或1分钟拉一次增量def incremental_update(api, recorder, market, code, date): last_offset recorder.get_last_offset(code, date) rows fetch_all_transaction(api, market, code, startlast_offset) if rows: recorder.save_batch(code, date, rows, last_offset)注意fetch_all_transaction要改一下从指定的start往后拉而不是每次从0开始。这样不仅省流量更重要的是可以做到分钟级刷新捕捉盘中异动。4.3 崩溃恢复与数据去重增量更新最怕什么最怕拉到一半程序崩了或者服务器返回的数据里有重复。SQLite的PRIMARY KEY (code, date, offset)设计天然解决了这个问题重复写入会触发INSERT OR REPLACE直接覆盖不会产生脏数据。我还建议每天收盘后做一次完整性校验拉一次全量分笔比对总数和最后一条成交时间。如果差得太多当天凌晨补跑一次矫正脚本。这个习惯帮我抓出过好几次因为盘中服务器断连导致的数据空洞。5. 实测必踩的5个坑时间偏移、频率限制、市场代码与字段陷阱5.1 1995年时间偏移pytdx最著名的历史包袱第一次拿到分笔数据时我差点被时间字段搞崩溃所有的time字段年份都是1995。比如上面代码块里的datetime.datetime(1995, 1, 5, 14, 55)14:55这个时分秒是对的但年份明显不对。这是因为通达信协议内部用一个基准时间做偏移pytdx沿用了这个行为。处理方式很简单拉完数据之后统一补年份from datetime import datetime def fix_tdx_time(t, ref_dateNone): if ref_date is None: ref_date datetime.now() if t.year 2010: t t.replace(yearref_date.year) return t注意一个隐蔽的问题如果程序跨年跑比如12月31日拉的数据补年份补成了次年跨年修复时容易搞错。解决方案是拉数据的时候从get_security_quotes接口获取服务器当日日期以那个日期为基准补年而不是用本地时间。5.2 请求频率限制不要挑战服务器的耐心我在前面提过请求别太快这里给一个具体的量级参考单连接连续快速请求无sleep大约在几十次之内就会触发服务器断开。加了0.1秒的停顿后基本不会触发。如果你的策略需要同时监控几十只股票我建议用多线程 多个连接的方式每个连接负责一组股票连接之间完全隔离。一旦某个连接断开只需要重建那一个不影响其他线程。5.3 市场代码匹配600、688、300、000的归属这个坑单独拿出来说是因为我见过不止一个人在代码里写死if code.startswith(6): market 1然后拿688001科创板代码去拉数据拉到一半发现不对。科创板代码是688开头但它是在上海交易所交易的所以market应该是1。北交所代码以8开头通达信协议接入方式和沪深不完全一样pytdx支持度有限。最稳妥的方案是维护一个公开的代码前缀映射表或者直接从get_security_list接口拉一次全市场列表把代码和市场对应关系存下来。# 从pytdx获取全市场股票列表并缓存市场归属 def build_market_map(api): market_map {} for market in [0, 1]: count api.get_security_count(market) count min(count, 5000) # 分批拉取 start 0 while start count: lst api.get_security_list(market, start) if not lst: break for item in lst: market_map[item[code]] market start len(lst) return market_map5.4 停牌股票返回空列表停牌股票没有成交get_transaction_data返回的是空列表[]。代码里如果直接rows[0]取第一条必然报IndexError。所有对分笔数据的处理代码第一行都应该是if not rows: return。另外要留意新上市股票上市首日没有集合竞价前的历史分笔盘中数据也可能有一段是空的。处理策略数据时这些边界情况都要考虑到不能假设所有股票每天一定有完整分笔。5.5 未复权价格与涨跌停边界分笔数据里的price是未复权的原始成交价。做资金流统计时如果要跨除权日对比价格必须自己对分红送转做复权处理否则会算出虚假的跳空和异动。还有一个更隐蔽的细节涨停一字板时大量买单挂在一字板上但成交很少此时buyorsell字段会出现非常多的0中性。如果你用“主动买单占比”来判断资金意图遇到这类情况会得到明显失真的结论。过滤掉涨停一字板的时间段或者单独标记这类行是资金流策略中必须做的数据清洗。6. 数据到策略一个资金流向统计的示例6.1 计算逻辑主动买卖盘怎么聚合有了分笔数据能做的事情很多。这里给一个最实用、最基础的例子主买主卖资金流统计。逻辑很简单把每一笔的price * vol算成成交金额按buyorsell聚合得到主动买入金额主动卖出金额中性盘金额多为集合竞价净主动买入 主动买入 - 主动卖出再按阈值拆分大单和小单比如单笔成交金额大于50万的算大单。大单净买入持续为正往往意味着有资金在主动吃货。6.2 完整实现拉数、清洗、聚合一条龙import pandas as pd from pytdx.hq import TdxHq_API def money_flow(api, market, code): data fetch_all_transaction(api, market, code) if not data: return None df pd.DataFrame(data) df[amount] df[price] * df[vol] df[time] df[time].apply(fix_tdx_time) # 按方向聚合 summary df.groupby(buyorsell)[amount].sum() buy_amount summary.get(1, 0) sell_amount summary.get(2, 0) neutral_amount summary.get(0, 0) # 大单阈值单笔50万 large_orders df[df[amount] 500000] large_buy large_orders.loc[large_orders[buyorsell] 1, amount].sum() large_sell large_orders.loc[large_orders[buyorsell] 2, amount].sum() return { code: code, 主动买入(万): round(buy_amount / 10000, 2), 主动卖出(万): round(sell_amount / 10000, 2), 中性盘(万): round(neutral_amount / 10000, 2), 净主动(万): round((buy_amount - sell_amount) / 10000, 2), 大单净买入(万): round((large_buy - large_sell) / 10000, 2), } api get_available_api() if api: result money_flow(api, 0, 000001) print(result) api.disconnect()跑出来的结果类似{code: 000001, 主动买入(万): 32560.32, 主动卖出(万): 30120.18, 中性盘(万): 2040.55, 净主动(万): 2440.14, 大单净买入(万): 1820.77}这个指标配合股价走势如果股价在涨但大单净买入持续为负说明可能是散户在抬轿、大资金在出货属于典型的量价背离信号。反过来股价横盘但大单持续净流入往往是吸筹特征。6.3 从研究到实盘的注意点数据是策略的地基但地基之上还要有摩擦成本。分笔数据可以精确到每一笔的成交价和成交量但你的真实成交价一定和最后一笔分笔价格有偏差尤其是想按分笔数据追单的时候滑点可能直接吃掉策略利润。我自己用这套数据做研究时明确了两件事第一分笔数据适合做盘后特征分析和策略研究不适合直接当成实盘信号源除非你把延迟和滑点模型加得很细第二数据采集要当成基础设施来维护每天定时拉取落盘连续积累一两周后你自己就拥有了一份宝贵的高频数据库这比临时去找任何外部数据都更贴合你自己的策略场景。先把取数、落盘、增量更新这套管道跑通后面想加订单流失衡指标、逐笔主动买卖占比、盘口异动监控都是在已有数据管道上接新模块的事。工具层面的路铺顺了策略灵感才能跑得起来。
pytdx实战：高效获取通达信分笔成交数据的完整指南
做日内量价策略那阵子我被“怎么拿分笔成交数据”卡了整整两天。市面上Level-2数据动辄几千一年免费的API要么只给K线不给分笔要么封装得太重装完依赖比策略代码还长。后来一个做短线量化的朋友提了一句“通达信协议有人封装过了叫pytdx分笔能直接拉。”我试了一下午第一笔真实分笔数据落进CSV的时候整个策略的验证路子瞬间就通了。这篇文章就围绕pytdx怎么高效获取通达信分笔成交数据把选型理由、连接细节、核心函数边界、数据落地和坑点一次讲透。适合刚接触量化、想拿真实盘口数据练手的Python开发者也适合已经在做策略回测、但觉得K线粒度不够想往订单流和资金流向方向深挖的朋友。我会直接给代码、给参数、给实测中踩过的坑你拿到就能跑。1. 分笔成交数据到底解决什么问题为什么不用K线和分钟线1.1 K线是结果分笔才是过程K线和分钟线解决的是“价格变成什么样子”分笔成交数据解决的是“价格是怎么一步步变成这个样子的”。这两件事在量化研究里完全是不同维度。举个例子某只票在14:55还是平开震荡尾盘突然一根长下影线收回1分钟K线上只显示价格从-3%快速拉回0%。但如果你看分笔数据会发现整个过程可能是5笔800手的大单连续主动买入砸出来的每笔的时间、价格、成交量、主买主卖方向清清楚楚。对短线策略来说这种“过程信息”才是真正决定第二天情绪和资金意图的东西。做资金流统计、主力异动追踪、冲击成本测算、盘口微观结构分析都需要分笔粒度。K线最多告诉你“发生了什么”分笔能告诉你“谁在什么价位做了什么动作”。这也是为什么很多游资和私募的看盘系统核心盯的永远是分笔而不是收盘K线。1.2 行情源选型为什么最终落在pytdx上在确定pytdx之前我把市面上常见的数据获取方式筛了一遍按“是否免费、是否支持分笔、接入成本”三个维度做了个对比数据源是否免费分笔支持接入成本稳定性与限制通达信协议直连pytdx免费支持实时低pip安装即用依赖第三方服务器需要容错处理某某Level-2付费行情付费支持中有官方SDK稳定但费用门槛高baostock免费基本不支持低K线和财务数据为主akshare免费部分接口不太稳定中爬虫接口随目标网站改版波动Tushare Pro积分制积分门槛较高中接口稳定但分笔权限要求高pytdx最吸引我的点有三个纯Python实现代码轻量依赖极少直接走通达信行情服务器的通讯协议拿到的是盘中实时数据接口语义非常直接获取分笔就是get_transaction_data一个函数的事不需要理解复杂的协议报文。当然它也有代价服务器不是官方托管的分布式集群连接质量要自己维护请求过频会被断连返回的数据字段注释也不够详细很多细节得靠实测摸索。但对你我这种自己搞研究、不想在数据上花大钱的场景pytdx已经是性价比最高的入口了。2. pytdx连接层细节安装、连接与容错策略2.1 安装和最小可用代码安装没什么好说的一个pip命令pip install pytdx装完之后最基础的连接和数据获取是这样的from pytdx.hq import TdxHq_API api TdxHq_API() # ip和port填通达信行情服务器的地址port固定为7709 connected api.connect(119.147.212.81, 7709) if not connected: print(连接失败) exit() # market: 0表示深圳1表示上海code是6位股票代码字符串 # start: 起始位置从0开始count: 拉取数量单次最大800 data api.get_transaction_data(0, 000001, 0, 30) print(data) api.disconnect()这段代码如果顺利会输出最近30笔成交记录。字段含义后面细说先确保你本地能把数据拉出来。注意connect返回的是布尔值别忽略它很多断连问题都是从这里先暴露的。2.2 行情服务器选择不要死磕单个IPpytdx的config.py里内置了一批公开行情服务器地址我实测下来不同服务器的延迟和数据完整性有明显差异。有些IP响应快拉到一半会断有些响应慢但数据稳定。更麻烦的是某些公网IP可能过一阵就失效了。我的做法是写一个简单的连接池把一批候选服务器测一遍选延迟最低且能成功返回数据的那个import time from pytdx.hq import TdxHq_API from pytdx.config.hosts import hq_hosts def get_available_api(max_retry5): for item in hq_hosts: api TdxHq_API() try: ip item[1] port item[2] if api.connect(ip, port, time_out3): # 用平安银行测一下数据接口是否正常 test api.get_transaction_data(0, 000001, 0, 1) if test is not None: print(f可用服务器: {ip}:{port}) return api else: api.disconnect() except Exception as e: print(f连接 {item[1]} 失败: {e}) continue time.sleep(0.3) return Nonetime_out3这个参数很关键不加的话遇到半死不活的服务器connect会卡住很影响体验。另外pytdx的连接对象不是线程安全的如果要用多线程并发拉多只股票每个线程必须创建独立的连接实例不能共用一个api去并发请求。2.3 频率控制跑太快会被踢这是pytdx最容易被新手忽略的限制。行情服务器对单IP的请求频率有限制如果脚本里用for循环无脑拉几百只股票的分笔大概率在拉第几十只的时候连接就被服务器断掉报ConnectionResetError或者直接返回None。安全起见两个请求之间加time.sleep(0.1)到time.sleep(0.3)。多线程并发时总并发连接数控制在5到8个以内。我实测过单个连接每分钟拉300到500次请求是没问题的超过这个量级就开始不稳定。3. 核心函数实战get_transaction_data的参数边界与数据语义3.1 参数细节market、code、start、count怎么传get_transaction_data(market, code, start, count)是获取分笔成交的核心入口四个参数每一个都有坑。market的值是0或10代表深圳市场1代表上海市场。code是6位字符串比如000001、600000。这里的坑在于判断该传0还是1不能只看代码首位是不是6。科创板688开头、沪市600/601/603/605都属于上海market1而深市主板000/001、中小板002、创业板300、科创板没有301也属于深圳market0。最稳妥的做法是用一个字典或者函数做映射def get_market_by_code(code): if code.startswith((600, 601, 603, 605, 688)): return 1 return 0start是分页位置的偏移量从0开始。count单次最多800条超过800也只返回800条。获取完整分笔需要循环翻页一直翻到返回结果数小于请求数为止。3.2 返回字段拆解time、price、vol、num、buyorsell拿到的一行数据长这样{ time: datetime.datetime(1995, 1, 5, 14, 55), price: 10.25, vol: 800, num: 5, buyorsell: 1 }这里要特别注意pytdx返回的vol单位是股不是手。800代表800股也就是8手。num表示这一笔成交由几笔组成比如800股可能是5笔小单合并的结果。buyorsell字段标记方向0代表中性一般出现在集合竞价或盘口无法判定方向的成交1代表主动买盘2代表主动卖盘。这个字段是后面做资金流向统计的核心依据。所谓“聪明钱”追踪本质上就是统计主动买盘和主动卖盘的成交金额差。3.3 全量分笔拉取翻页循环的标准写法搞清楚参数边界后完整拉取某只股票当天所有分笔的代码就很清晰了import time def fetch_all_transaction(api, market, code, chunk_size800, delay0.1): rows [] start 0 while True: chunk api.get_transaction_data(market, code, start, chunk_size) if not chunk: break rows.extend(chunk) if len(chunk) chunk_size: break start len(chunk) time.sleep(delay) # 温和限速避免被服务器断开 return rows api get_available_api() if api: data fetch_all_transaction(api, 0, 000001) print(f平安银行当日分笔总数: {len(data)}) print(data[-5:]) # 打印最后5笔 api.disconnect()用这个函数一只成交活跃的股票全天分笔大概有几千到几万笔不等拉完也就几秒的事。注意我这里的delay参数之前讲过频率控制在循环翻页里同样适用。3.4 历史分笔get_history_transaction_data的局限除了实时分笔pytdx还提供了get_history_transaction_data可以按日期拉取历史分笔# 获取某只股票2024年1月5日的历史分笔 history_data api.get_history_transaction_data(0, 000001, 0, 800, 2024-01-05)这个接口的date参数格式是2024-01-05注意不同服务器的历史数据保留深度不一样有的只保留最近10来个交易日有的能保留更久。实测下来不要对这个接口的历史深度抱太大期望用于补最近几天的数据是够的想做几个月以上的分笔级回测还是得自己每日落盘积累。4. 数据落地从全量拉取到增量更新4.1 存储格式选择CSV还是SQLite分笔数据一天几万条存储格式用CSV还是SQLite取决于你怎么用。如果只是做盘后分析、回测验证CSV按日存储最简单文件名带上日期和代码比如600000_2024-01-05.csvpandas直接读。缺点是文件数量多做全市场扫描时要遍历一堆文件。如果要做实时盘中监控或者频繁查询某只票某一天的成交明细SQLite更合适。一张表搞定所有股票所有日期按code date建索引查询效率很高。我现在的策略是先全量拉取落到SQLite需要做特征工程时再用pandas拉出来。4.2 增量更新记录最后一条的序号分笔数据是实时增加的每次全量拉一遍既浪费请求次数又容易被服务器限流。更好的做法是记录上次拉到的位置增量拉取新数据。get_transaction_data的start参数是当天分笔的位置偏移所以增量更新的核心就是记住上一次拉到了哪个偏移位置class TransactionRecorder: def __init__(self, db_path): self.conn sqlite3.connect(db_path) self._init_table() def _init_table(self): self.conn.execute( CREATE TABLE IF NOT EXISTS transaction ( code TEXT, date TEXT, offset INTEGER, time TEXT, price REAL, vol INTEGER, num INTEGER, buyorsell INTEGER, PRIMARY KEY (code, date, offset) ) ) def get_last_offset(self, code, date): row self.conn.execute( SELECT MAX(offset) FROM transaction WHERE code? AND date?, (code, date) ).fetchone() return row[0] if row and row[0] is not None else 0 def save_batch(self, code, date, rows, start): data [ (code, date, start i, r[time].strftime(%H:%M:%S), r[price], r[vol], r[num], r[buyorsell]) for i, r in enumerate(rows) ] self.conn.executemany( INSERT OR REPLACE INTO transaction VALUES (?,?,?,?,?,?,?,?), data ) self.conn.commit()盘中写一个定时任务每30秒或1分钟拉一次增量def incremental_update(api, recorder, market, code, date): last_offset recorder.get_last_offset(code, date) rows fetch_all_transaction(api, market, code, startlast_offset) if rows: recorder.save_batch(code, date, rows, last_offset)注意fetch_all_transaction要改一下从指定的start往后拉而不是每次从0开始。这样不仅省流量更重要的是可以做到分钟级刷新捕捉盘中异动。4.3 崩溃恢复与数据去重增量更新最怕什么最怕拉到一半程序崩了或者服务器返回的数据里有重复。SQLite的PRIMARY KEY (code, date, offset)设计天然解决了这个问题重复写入会触发INSERT OR REPLACE直接覆盖不会产生脏数据。我还建议每天收盘后做一次完整性校验拉一次全量分笔比对总数和最后一条成交时间。如果差得太多当天凌晨补跑一次矫正脚本。这个习惯帮我抓出过好几次因为盘中服务器断连导致的数据空洞。5. 实测必踩的5个坑时间偏移、频率限制、市场代码与字段陷阱5.1 1995年时间偏移pytdx最著名的历史包袱第一次拿到分笔数据时我差点被时间字段搞崩溃所有的time字段年份都是1995。比如上面代码块里的datetime.datetime(1995, 1, 5, 14, 55)14:55这个时分秒是对的但年份明显不对。这是因为通达信协议内部用一个基准时间做偏移pytdx沿用了这个行为。处理方式很简单拉完数据之后统一补年份from datetime import datetime def fix_tdx_time(t, ref_dateNone): if ref_date is None: ref_date datetime.now() if t.year 2010: t t.replace(yearref_date.year) return t注意一个隐蔽的问题如果程序跨年跑比如12月31日拉的数据补年份补成了次年跨年修复时容易搞错。解决方案是拉数据的时候从get_security_quotes接口获取服务器当日日期以那个日期为基准补年而不是用本地时间。5.2 请求频率限制不要挑战服务器的耐心我在前面提过请求别太快这里给一个具体的量级参考单连接连续快速请求无sleep大约在几十次之内就会触发服务器断开。加了0.1秒的停顿后基本不会触发。如果你的策略需要同时监控几十只股票我建议用多线程 多个连接的方式每个连接负责一组股票连接之间完全隔离。一旦某个连接断开只需要重建那一个不影响其他线程。5.3 市场代码匹配600、688、300、000的归属这个坑单独拿出来说是因为我见过不止一个人在代码里写死if code.startswith(6): market 1然后拿688001科创板代码去拉数据拉到一半发现不对。科创板代码是688开头但它是在上海交易所交易的所以market应该是1。北交所代码以8开头通达信协议接入方式和沪深不完全一样pytdx支持度有限。最稳妥的方案是维护一个公开的代码前缀映射表或者直接从get_security_list接口拉一次全市场列表把代码和市场对应关系存下来。# 从pytdx获取全市场股票列表并缓存市场归属 def build_market_map(api): market_map {} for market in [0, 1]: count api.get_security_count(market) count min(count, 5000) # 分批拉取 start 0 while start count: lst api.get_security_list(market, start) if not lst: break for item in lst: market_map[item[code]] market start len(lst) return market_map5.4 停牌股票返回空列表停牌股票没有成交get_transaction_data返回的是空列表[]。代码里如果直接rows[0]取第一条必然报IndexError。所有对分笔数据的处理代码第一行都应该是if not rows: return。另外要留意新上市股票上市首日没有集合竞价前的历史分笔盘中数据也可能有一段是空的。处理策略数据时这些边界情况都要考虑到不能假设所有股票每天一定有完整分笔。5.5 未复权价格与涨跌停边界分笔数据里的price是未复权的原始成交价。做资金流统计时如果要跨除权日对比价格必须自己对分红送转做复权处理否则会算出虚假的跳空和异动。还有一个更隐蔽的细节涨停一字板时大量买单挂在一字板上但成交很少此时buyorsell字段会出现非常多的0中性。如果你用“主动买单占比”来判断资金意图遇到这类情况会得到明显失真的结论。过滤掉涨停一字板的时间段或者单独标记这类行是资金流策略中必须做的数据清洗。6. 数据到策略一个资金流向统计的示例6.1 计算逻辑主动买卖盘怎么聚合有了分笔数据能做的事情很多。这里给一个最实用、最基础的例子主买主卖资金流统计。逻辑很简单把每一笔的price * vol算成成交金额按buyorsell聚合得到主动买入金额主动卖出金额中性盘金额多为集合竞价净主动买入 主动买入 - 主动卖出再按阈值拆分大单和小单比如单笔成交金额大于50万的算大单。大单净买入持续为正往往意味着有资金在主动吃货。6.2 完整实现拉数、清洗、聚合一条龙import pandas as pd from pytdx.hq import TdxHq_API def money_flow(api, market, code): data fetch_all_transaction(api, market, code) if not data: return None df pd.DataFrame(data) df[amount] df[price] * df[vol] df[time] df[time].apply(fix_tdx_time) # 按方向聚合 summary df.groupby(buyorsell)[amount].sum() buy_amount summary.get(1, 0) sell_amount summary.get(2, 0) neutral_amount summary.get(0, 0) # 大单阈值单笔50万 large_orders df[df[amount] 500000] large_buy large_orders.loc[large_orders[buyorsell] 1, amount].sum() large_sell large_orders.loc[large_orders[buyorsell] 2, amount].sum() return { code: code, 主动买入(万): round(buy_amount / 10000, 2), 主动卖出(万): round(sell_amount / 10000, 2), 中性盘(万): round(neutral_amount / 10000, 2), 净主动(万): round((buy_amount - sell_amount) / 10000, 2), 大单净买入(万): round((large_buy - large_sell) / 10000, 2), } api get_available_api() if api: result money_flow(api, 0, 000001) print(result) api.disconnect()跑出来的结果类似{code: 000001, 主动买入(万): 32560.32, 主动卖出(万): 30120.18, 中性盘(万): 2040.55, 净主动(万): 2440.14, 大单净买入(万): 1820.77}这个指标配合股价走势如果股价在涨但大单净买入持续为负说明可能是散户在抬轿、大资金在出货属于典型的量价背离信号。反过来股价横盘但大单持续净流入往往是吸筹特征。6.3 从研究到实盘的注意点数据是策略的地基但地基之上还要有摩擦成本。分笔数据可以精确到每一笔的成交价和成交量但你的真实成交价一定和最后一笔分笔价格有偏差尤其是想按分笔数据追单的时候滑点可能直接吃掉策略利润。我自己用这套数据做研究时明确了两件事第一分笔数据适合做盘后特征分析和策略研究不适合直接当成实盘信号源除非你把延迟和滑点模型加得很细第二数据采集要当成基础设施来维护每天定时拉取落盘连续积累一两周后你自己就拥有了一份宝贵的高频数据库这比临时去找任何外部数据都更贴合你自己的策略场景。先把取数、落盘、增量更新这套管道跑通后面想加订单流失衡指标、逐笔主动买卖占比、盘口异动监控都是在已有数据管道上接新模块的事。工具层面的路铺顺了策略灵感才能跑得起来。
