用 Python 获取股票分笔成交数据：pytdx 库实操与深度解析

在量化交易与精细化股票分析领域，日 K 线、分时图这类常规数据已难以满足深度研究需求。想要精准捕捉资金流向、剖析市场微观交易结构，就必须依托分笔成交数据（Tick Data）。这类数据完整记录了每一笔交易的时间、价格、成交量及买卖方向，是解锁市场深层逻辑的关键数据源。

本文将聚焦如何通过pytdx库对接通达信数据源，实现个股分笔成交数据的获取与处理，并拆解代码逻辑、梳理实操要点，助力开发者快速落地相关分析场景。

一、核心目标与整体流程
本次实现的核心目标，是获取指定个股在特定日期的全部分笔成交记录，并按时间维度完成分组整理，为后续分析铺路。整个流程可拆解为 5 个关键步骤，环环相扣保障数据获取的准确性与可用性：

对接通达信服务器：建立与数据源的稳定连接，这是数据获取的基础；

自动匹配交易市场：依据股票代码前缀，智能判定其所属的上海、深圳市场或基金品类；

分页抓取历史数据：针对服务器单页数据限制，通过循环实现全量数据的高效拉取；

数据优化处理：校正特殊品类价格、计算成交金额、规范数据格式，解决原始数据的适配问题；

结构化分组输出：以时间为维度整合数据，生成清晰易读的字典格式结果。
二、代码拆解与核心逻辑详解

1. 依赖库导入
   开展开发前，需先导入核心工具库，各库的核心作用如下：
   python
   运行
   from pytdx.hq import TdxHq_API # 提供通达信行情数据的核心调用接口
   from pytdx.params import TDXParams # 包含通达信市场类型等预设参数常量
   from collections import defaultdict # 简化按时间分组的数据存储操作
   其中defaultdict是提升代码效率的实用工具，它允许直接向未定义的键追加数据，无需额外判断键是否存在，特别适配数据分组场景。
2. 建立数据源连接
   要获取通达信数据，需先初始化 API 实例并连接至远程服务器，代码如下：
   python
   运行
   api = TdxHq_API()
   api.connect('xxx.xxx.xxx.xxx', 7709)
   此处的 IP 地址为通达信公共服务器地址，实际使用时可替换为稳定可用的服务器地址，端口 7709 为通达信行情服务的常用端口。连接成功后，方可执行后续数据请求操作。

3. 核心函数：分笔数据的获取与处理
   核心函数get_transaction_data(code, date)整合了数据获取、处理与分组的全部逻辑，下面分模块展开解析。
   （1）智能判定市场与品类
   通达信 API 请求需指定市场参数，而股票代码前缀隐含了市场及品类信息，据此可实现自动化判定：
   python
   运行
   def get_transaction_data(code, date):

   依据代码前缀匹配市场

   if code.startswith(('0', '3')):
   market = TDXParams.MARKET_SZ # 0开头为深市主板，3开头为创业板
   elif code.startswith('6'):
   market = TDXParams.MARKET_SH # 6开头为沪市主板
   elif code.startswith(('5', '1')):

   5开头为沪市基金，1开头为深市基金

   market = TDXParams.MARKET_SH if code.startswith('5') else TDXParams.MARKET_SZ

   else:
   raise ValueError(f"无法识别的股票代码：{code}")

   标记是否为基金，用于后续价格校正

   is_fund = code.startswith(('5', '1'))
   该逻辑既避免了手动输入市场参数的繁琐，也减少了因参数错误导致的请求失败。同时标记基金品类，是因为基金数据存在特殊的价格格式问题，需单独处理。

（2）分页拉取全量数据
通达信服务器单次最多返回 1000 条分笔数据，若单日成交数据超过 1000 条，需通过分页循环获取，具体实现如下：
python
运行
data, start = [], 0
while True:

每次请求1000条数据，start为起始位置

    part = api.get_history_transaction_data(market, code, start, 1000, int(date))
    if not part:  # 无数据返回时终止循环
        break
    data.extend(part)  # 拼接分页数据
    if len(part) < 1000:  # 数据不足1000条，说明已到最后一页
        break
    start += 1000  # 更新起始位置，准备获取下一页

这种分页请求模式是处理大量数据的常用方案，既能规避单次请求的数据量限制，也能保障数据的完整性。

（3）数据清洗与结构化分组
原始数据需经过格式校正、字段计算后，再按时间分组，才能满足分析需求，关键代码如下：
python
运行

用defaultdict按时间分组存储数据

trans_dict = defaultdict(list)
for record in data:
    # 关键校正：通达信基金价格为实际价格的10倍，需还原
    real_price = record['price'] / 10.0 if is_fund else record['price']
    # 格式化数据，符合日常分析的数值规范
    formatted_price = round(real_price, 3)
    # 计算成交额：价格×成交量（手）×100（1手=100股）
    transaction_amount = round(real_price * record['vol'] * 100, 2)

    # 按时间维度添加数据
    trans_dict[record['time']].append({
        '成交价格': formatted_price,
        '成交手数': record['vol'],
        '买卖方向': record['buyorsell'],  # 0=主动买，1=主动卖，2=中性
        '成交金额': transaction_amount
    })

# 按时间排序后返回字典
return dict(sorted(trans_dict.items()))

此环节的核心亮点在于基金价格校正和成交额计算 —— 这两个细节直接影响数据准确性，若忽略可能导致后续分析出现严重偏差。同时按时间排序能让输出结果更符合交易时序逻辑。

4. 函数调用与资源释放
   完成函数定义后，设置目标参数调用函数，获取数据并打印，最后断开 API 连接释放资源：
   python
   运行

   配置目标日期和股票代码（示例为中概互联ETF）

   target_date = "20250909"
   target_stock = "513050"

获取分笔数据

transaction_detail = get_transaction_data(target_stock, target_date)

输出结果

print(f"{target_stock}在{target_date}的分笔成交数据（按时间分组）：")
print(transaction_detail)

断开连接

api.disconnect()
三、输出结果与数据解读
运行代码后，将得到按时间升序排列的嵌套字典数据，结构清晰，便于后续分析，示例如下：
python
运行
513050在20250909的分笔成交数据（按时间分组）：
{
'09:25': [
{'成交价格': 1.234, '成交手数': 500, '买卖方向': 0, '成交金额': 61700.0},
{'成交价格': 1.235, '成交手数': 300, '买卖方向': 1, '成交金额': 37050.0}
],
'09:30': [
{'成交价格': 1.236, '成交手数': 100, '买卖方向': 0, '成交金额': 12360.0},
{'成交价格': 1.236, '成交手数': 200, '买卖方向': 1, '成交金额': 24720.0}
],
'14:57': [
{'成交价格': 1.235, '成交手数': 400, '买卖方向': 1, '成交金额': 49400.0},
{'成交价格': 1.236, '成交手数': 280, '买卖方向': 0, '成交金额': 34608.0}
]
}
其中各字段含义明确：
外层键为交易时间，涵盖集合竞价时段（如 09:25）和连续竞价时段的关键成交时刻；

内层列表中的每条数据对应一笔交易，买卖方向字段可直接用于判断单笔交易的主动操作类型，成交金额和成交手数则是量化资金规模的核心指标。

四、应用场景与拓展方向
获取分笔成交数据后，可支撑多种深度分析场景，同时也有诸多拓展空间提升代码实用性。

1. 核心应用场景
   资金流向研判：统计不同时段主动买入、卖出的总金额，判断个股当日资金是净流入还是净流出；

成交密集区定位：筛选成交量骤增的时段和对应价格区间，这些区间往往是个股的短期支撑位或压力位；

主力动向追踪：设置大单阈值（如单笔成交 500 手以上），筛选大单交易记录，推测主力的买卖意图；

高频策略回测：为日内高频交易策略提供 Tick 级数据支撑，验证策略在不同成交场景下的有效性。

2. 实用拓展方向
   批量数据获取：新增循环逻辑，遍历股票代码列表和日期列表，批量抓取多只个股多日的分笔数据；
   数据持久化存储：将处理后的数据写入 CSV 文件或 MySQL、SQLite 数据库，避免重复请求，同时便于长期数据积累；
   可视化呈现：借助 Matplotlib、Plotly 等库，绘制分时成交量走势图、资金流向热力图，让分析结果更直观；
   异常交易预警：添加阈值判断逻辑，当某时段成交量、价格波动超出设定范围时，触发预警提示。

五、完整可运行代码
python
运行
from pytdx.hq import TdxHq_API
from pytdx.params import TDXParams
from collections import defaultdict

初始化API并建立服务器连接

api = TdxHq_API()
api.connect('xxx.xxx.xxx.xxx', 7709)

def get_transaction_data(code, date):
"""
获取指定股票在指定日期的分笔成交数据
:param code: 股票代码
:param date: 目标日期，格式为"YYYYMMDD"
:return: 按时间分组的分笔成交数据字典
"""

判定市场类型

if code.startswith(('0', '3')):
    market = TDXParams.MARKET_SZ
elif code.startswith('6'):
    market = TDXParams.MARKET_SH
elif code.startswith(('5', '1')):
    market = TDXParams.MARKET_SH if code.startswith('5') else TDXParams.MARKET_SZ
else:
    raise ValueError(f"无法识别的股票代码：{code}")

is_fund = code.startswith(('5', '1'))
data, start = [], 0

# 分页获取全量数据
while True:
    part_data = api.get_history_transaction_data(market, code, start, 1000, int(date))
    if not part_data:
        break
    data.extend(part_data)
    if len(part_data) < 1000:
        break
    start += 1000

# 数据处理与按时间分组
trans_dict = defaultdict(list)
for record in data:
    real_price = record['price'] / 10.0 if is_fund else record['price']
    formatted_price = round(real_price, 3)
    transaction_amount = round(real_price * record['vol'] * 100, 2)

    trans_dict[record['time']].append({
        '成交价格': formatted_price,
        '成交手数': record['vol'],
        '买卖方向': record['buyorsell'],
        '成交金额': transaction_amount
    })

return dict(sorted(trans_dict.items()))

配置参数并执行

target_date = "20250909"
target_stock = "513050"
transaction_detail = get_transaction_data(target_stock, target_date)

输出结果

print(f"{target_stock}在{target_date}的分笔成交数据（按时间分组）：")
print(transaction_detail)

断开API连接

api.disconnect()

需要开通量化低门槛API 接口的（prtade qmt miniqmt，欢迎加v联系，备注：量化开户