量化交易数据分析——通过pytdx用 Python 获取证券分笔成交高频数据

文章标题：深入解析：用 Python 获取A股分笔成交高频数据

在量化交易和精细化的股票分析中，看日K线、分时图往往不够，我们有时需要深入到每一笔成交的记录，即 分笔成交数据（Tick Data）。这类数据记录了市场上每一个成交动作的时间、价格、成交量以及买卖方向，是分析资金流向、市场微观结构的基础。

本文将深入解析一段利用 pytdx 库从通达信数据源获取个股分笔成交数据的 Python 代码，并详细讲解其背后的逻辑和每一步的作用。

一、代码概览与核心功能

这段代码的核心功能是：获取指定股票在指定日期所有分笔成交的详细记录，并按时间分组，以便于分析。

它主要完成了以下任务：

1. 连接数据源：连接到通达信的远程服务器。
2. 智能判断市场：根据股票代码自动判断其属于上海、深圳市场还是基金。
3. 分页数据获取：高效地获取全部历史分笔数据。
4. 数据清洗与转换：对原始数据进行处理（如基金价格处理）、计算成交额并格式化。
5. 结构化输出：将数据按时间分组，并以清晰的字典形式输出。

二、逐行代码解析

1. 导入必要的库

from pytdx.hq import TdxHq_API       # 通达信行情API的核心类
from pytdx.params import TDXParams    # 通达信参数定义，如市场常量
from collections import defaultdict   # 使用默认字典来方便地分组数据

• pytdx：一个非常强大的开源库，提供了访问通达信数据的接口。
• defaultdict：这是代码的一个技巧点，它允许我们直接向不存在的键追加数据，非常适合用于按时间分组。

2. 连接数据服务器

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

• 创建 API 实例并连接到指定的通达信服务器（IP地址）。这个公共服务器通常可用，但稳定性可能随时间变化。

3. 核心函数 get_transaction_data(code, date)

这是整个代码的灵魂，我们一步步来看。

a. 判断股票市场与类型

if code.startswith(('0', '3')):
    market = TDXParams.MARKET_SZ  # 深市A股/创业板
elif code.startswith('6'):
    market = TDXParams.MARKET_SH  # 沪市A股
elif code.startswith(('5', '1')):
    market = TDXParams.MARKET_SH if code.startswith('5') else TDXParams.MARKET_SZ # 沪基/深基
else:
    raise ValueError(f"未知的股票代码格式: {code}")

• 通达信API需要指定市场参数。这里通过股票代码的前缀来智能判断：
  • 0（深市主板）、3（创业板） → 深圳市场 (MARKET_SZ)
  • 6（沪市主板） → 上海市场 (MARKET_SH)
  • 5（上海基金）、1（深圳基金） → 分别对应上海和深圳市场
• 同时，代码还通过 is_fund = code.startswith(('5', '1')) 标记了是否为基金，这对后续价格处理至关重要。

b. 分页获取数据

data, start = [], 0
while True:
    part = api.get_history_transaction_data(market, code, start, 1000, int(date))
    if not part:
        break
    data.extend(part)
    if len(part) < 1000:
        break
    start += 1000

• 这是一个经典的分页循环模式。因为服务器一次最多返回 1000 条记录。
• start 是起始位置，第一次从 0 开始，获取 1000 条。如果返回的数据不足 1000 条，说明已经是最后一页，循环终止。否则，将 start 增加 1000，继续获取下一页，直到拿完所有数据。

c. 数据清洗、转换与分组

trans = defaultdict(list)
for d in data:
    # 关键处理：基金的价格需要除以10
    price = d['price'] / 10.0 if is_fund else d['price']

    formatted_price = round(price, 3)
    formatted_amount = round(price * d['vol'] * 100, 2) # 成交额 = 价 * 量(手) * 100

    trans[d['time']].append({
        '价': formatted_price,
        '量': d['vol'],        # 单位是「手」
        'BS': d['buyorsell'],  # 0-买, 1-卖, 2-中性
        '额': formatted_amount
    })
return dict(sorted(trans.items()))

• 基金价格处理：通达信返回的基金价格是实际价格的 10 倍，所以需要 / 10.0 进行校正。这是非常关键且容易出错的一点。
• 计算成交额：原始数据只有 price（价格）和 vol（成交量，单位是手）。成交额 = 价格 × 成交量 × 100（因为1手=100股）。
• 格式化：使用 round() 函数将价格和金额格式化为保留 3 位和 2 位小数，符合日常看盘习惯。
• 按时间分组：使用 defaultdict(list)，以时间字符串（如 ‘09:30’）为键，将同一时间发生的所有成交记录放入一个列表中。最后通过 sorted(trans.items()) 按时间先后顺序返回。

4. 执行与输出

target_date = "20250909"
stock_code = "513050" # 示例代码，中概互联ETF
trans = get_transaction_data(stock_code, target_date)
print(f"股票 {stock_code} 在 {target_date} 的分笔交易数据（按时间分组）：")
print(trans)
api.disconnect()

• 设置目标日期和股票代码，调用函数并打印结果。最后记得断开连接，释放资源。

三、输出数据结构详解

运行代码后，你会得到一个嵌套的字典结构，非常清晰：

{
  '09:25': [ # 集合竞价时段可能有多笔
    {'价': 1.234, '量': 500, 'BS': 0, '额': 61700.0},
    {'价': 1.235, '量': 300, 'BS': 1, '额': 37050.0}
  ],
  '09:30': [
    {'价': 1.236, '量': 100, 'BS': 0, '额': 12360.0}
  ],
  '10:00': [
    ...
  ],
  ...
}

• 第一层Key：时间（str）。
• 第二层Value：一个列表（list），包含该时刻的所有成交明细。
• 明细字典：包含四个关键信息：
  • 价：成交价格。
  • 量：成交手数。
  • BS：买卖方向（0：主动买入，1：主动卖出，2：中性盘/不明）。
  • 额：成交金额（元）。

四、应用场景与扩展思路

获取到这些数据后，你可以进行非常多有趣且深入的分析：

1. 资金流向分析：统计不同时间段内主动买入和主动卖出的总金额，判断资金是流入还是流出。
2. 成交密集区分析：寻找成交量突然放大的时间点和高低价格，识别支撑位和压力位。
3. 大单追踪：筛选出成交量巨大的单笔成交（大单），分析主力动向。
4. 高频策略回测：为基于Tick级别的高频交易策略提供数据基础。
5. 市场微观结构研究：研究买卖挂单、成交间隔等。

扩展思路：

• 批量获取：修改代码，使其可以循环获取多只股票、多个交易日的数据。
• 数据持久化：将获取到的数据存入 CSV 文件或 SQLite/MySQL 数据库，方便后续分析。
• 可视化：使用 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) # IP地址

def get_transaction_data(code, date):
    """获取指定股票在指定日期的所有分笔数据，按时间分组返回"""
    # 判断市场类型
    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}")
    
    # 判断是否为基金（沪市5开头或深市1开头）
    is_fund = code.startswith(('5', '1'))
    
    data, start = [], 0
    while True:
        part = api.get_history_transaction_data(market, code, start, 1000, int(date))
        if not part: 
            break
        data.extend(part)
        if len(part) < 1000: 
            break
        start += 1000

    # 按时间分组并构建所需字段
    trans = defaultdict(list)
    for d in data:
        # 处理价格：如果是基金，需要除以10
        price = d['price'] / 10.0 if is_fund else d['price']
        
        # 格式化价格和成交额，保留3位小数
        formatted_price = round(price, 3)
        formatted_amount = round(price * d['vol'] * 100, 2)
        
        trans[d['time']].append({
            '价': formatted_price,     # 成交价（已处理基金价格并格式化）
            '量': d['vol'],            # 成交量（手）
            'BS': d['buyorsell'],      # 买卖方向：0-买, 1-卖, 2-中性
            '额': formatted_amount     # 成交额 = 价格 × 手数 × 100（已格式化）
        })
    return dict(sorted(trans.items()))

# === 参数设置 ===
target_date = "20250909"
stock_code = "513050"

# === 获取结果 ===
trans = get_transaction_data(stock_code, target_date)

# === 美化输出数据结构，便于查看和AI提问 ===
print(f"股票 {stock_code} 在 {target_date} 的分笔交易数据（按时间分组）：")
print(trans)

# === 断开连接 ===
api.disconnect()

六、输出效果