企业微信

通达信API价格获取的基础原理

通达信软件提供了一套动态链接库（DLL）形式的API，允许外部程序调用其内部行情数据。核心接口为TdxHq_API，通过TCP协议与通达信行情服务器通信。要获取价格数据，需理解其请求-响应模式：客户端发送特定格式的数据包，服务器返回包含最新价、涨跌幅等信息的字节流。

初始化连接：从零开始搭建环境

所有价格获取操作的前提是建立与服务器的稳定连接。使用Python的pytdx库是最便捷的方式。安装后，通过以下代码实现连接：


from pytdx.hq import TdxHq_API

api = TdxHq_API()

# 连接通达信行情主站，端口通常为7709

if api.connect('119.147.212.81', 7709):

    print('连接成功')

else:

    print('连接失败')

连接成功后，API对象便可用于后续数据请求。连接时需注意服务器IP的时效性，定期从官网或第三方获取最新可用地址。

行情数据请求：核心函数详解

通达信API提供了多种行情函数，其中get_security_quotes用于获取单只或多只股票的实时报价。该函数接受市场代码（0代表深圳，1代表上海）和证券代码列表，返回包含以下关键字段的字典：

如何利用通达信API获取实时价格数据

  • price：最新成交价

  • last_close：昨收价

  • open：今开价

  • high：最高价

  • low：最低价

  • volume：成交量

  • amount：成交额

示例代码如下：


# 获取平安银行（000001）的实时报价

quotes = api.get_security_quotes([(0, '000001')])

if quotes:

    data = quotes[0]

    print(f"最新价: {data['price']}")

    print(f"涨跌幅: {(data['price'] - data['last_close']) / data['last_close'] * 100:.2f}%")

else:

    print('未获取到数据')

该函数支持批量请求，传入多个元组即可一次获取多只股票的数据，极大提升效率。

数据解析与异常处理：确保数据准确性

返回的数据是原始字节流，pytdx已将其解析为Python字典。但使用中需注意字段类型转换和空值处理。价格字段可能为None，此时应作有效性判断。请求频率过快可能被服务器限制，需在循环中加入延时（如0.1秒）。以下代码展示稳健的数据获取模式：


import time

def get_price_safe(api, market, code):

    try:

        data = api.get_security_quotes([(market, code)])[0]

        if data.get('price') is not None:

            return float(data['price'])

        else:

            return None

    except Exception as e:

        print(f"获取{code}失败: {e}")

        return None

price = get_price_safe(api, 0, '000001')

if price:

    print(f"价格: {price}")

实时监控与自动刷新：构建价格追踪系统

为了持续跟踪价格变动，需定时刷新数据。结合time.sleep或定时任务，实现简单监控。以下示例每5秒打印一次指定股票的价格：


import time

def monitor(api, market, code, interval=5):

    while True:

        price = get_price_safe(api, market, code)

        if price:

            print(f"{time.strftime('%Y-%m-%d %H:%M:%S')} - {code} 价格: {price}")

        time.sleep(interval)

# 启动监控（需在独立线程中运行，此处为演示）

# monitor(api, 0, '000001')

实际应用中，可将监控放入后台线程，并在条件触发时发出警报。

实战案例：跨市场股票价格对比

假设需要同时监控沪深两市的多只股票，利用批量接口可减少请求次数。以下代码获取平安银行和浦发银行（600000）的价格：


quotes = api.get_security_quotes([(0, '000001'), (1, '600000')])

for item in quotes:

    code = item['code']

    price = item['price']

    change_pct = (price - item['last_close']) / item['last_close'] * 100

    print(f"{code}: 最新价 {price}, 涨跌幅 {change_pct:.2f}%")

输出结果可直接用于策略决策。

性能优化：多线程与缓存策略

在需要监控大量股票时，单线程循环可能不够高效。可考虑以下优化：

  • 使用concurrent.futures.ThreadPoolExecutor并发请求，但需注意API线程安全，建议每个线程独立连接。

  • 缓存价格数据，减少重复请求。例如，在字典中存储最近一次获取的价格和时间戳，仅当超过一定时间才重新请求。

  • 合理设置超时时间，避免等待过久。pytdx的connect方法支持超时参数。


from concurrent.futures import ThreadPoolExecutor

def fetch_batch(codes):

    def fetch_one(code):

        # 每个线程创建独立连接

        local_api = TdxHq_API()

        local_api.connect('119.147.212.81', 7709)

        try:

            data = local_api.get_security_quotes([code])[0]

            return data

        finally:

            local_api.disconnect()



    with ThreadPoolExecutor(max_workers=10) as executor:

        results = list(executor.map(fetch_one, codes))

    return results

codes = [(0, '000001'), (1, '600000'), (0, '000002')]

prices = fetch_batch(codes)

for d in prices:

    print(d['code'], d['price'])

常见问题与故障排除

  • 连接失败：检查网络防火墙是否屏蔽端口，更换服务器IP。

  • 数据延迟：确保频繁请求时添加适当延时，避免被服务器拉黑。

  • 数据字段缺失：某些字段在集合竞价阶段可能为空，需做好容错。

  • 线程安全问题：同一api实例不可在多线程中共享，应每个线程独立创建。

进阶方向：从价格获取到自动化策略

获取实时价格只是第一步，结合tushare或backtrader等库，可构建更复杂的交易系统。例如，基于价格突破策略，自动下单（需接入券商接口）。通达信API提供了足够的基础数据支撑，后续可扩展至K线、分时等历史数据，为策略回测提供素材。

通达信API的价格获取是量化交易的基础技能。掌握连接、请求、解析、优化等环节，便能高效利用行情数据。建议在实践中多测试不同场景，逐步完善代码的健壮性。随着对API的深入，还可探索更多高级功能，如五档盘口、逐笔成交等，为交易决策提供更全面的参考。