开通量化交易接口用 Python 获取股票分笔成交数据：pytdx 库实操与深度解析

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

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

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

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

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

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

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

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

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

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

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

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 条，需通过分页循环获取，具体实现如下：

    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）数据清洗与结构化分组
原始数据需经过格式校正、字段计算后，再按时间分组，才能满足分析需求，关键代码如下：


    # 用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()))

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

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

# 配置目标日期和股票代码（示例为中概互联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()

三、输出结果与数据解读
运行代码后，将得到按时间升序排列的嵌套字典数据，结构清晰，便于后续分析，示例如下：


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）和连续竞价时段的关键成交时刻；

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

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

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

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

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

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

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

五、完整可运行代码


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()