如何用通达信API获取实时行情数据并用于量化交易
摘要:
通达信API获取股票期货实时行情，通过pytdx库连接服务器，订阅行情，处理数据，构建量化策略。数据延迟低，适合日内交易。
通达信API实时行情接入实战
通达信API的核心能力
通达信提供多种数据接口，包括本地行情文件读取、网络协议直接通信、DLL动态链接库调用。本地文件读取延迟高，适合历史回测。网络协议直接通信延迟低，适合实盘。DLL调用性能最优但开发复杂。量化交易优先选择网络协议，通过Python的pytdx库即可实现。pytdx封装了通达信标准行情协议，支持股票、期货、期权、指数。
环境搭建与服务器连接
安装pytdx：
pip install pytdx
连接通达信行情服务器。服务器列表可从网上获取，选择延迟低的节点。
from pytdx.hq import TdxHq_API
api = TdxHq_API()
# 服务器地址和端口，常见如 119.147.212.81:7709
servers = [
('119.147.212.81', 7709),
('114.80.63.12', 7709),
('218.108.98.244', 7709),
]
for ip, port in servers:
if api.connect(ip, port):
print(f'已连接 {ip}:{port}')
break
else:
raise ConnectionError('所有服务器连接失败')
连接成功后，调用api.get_security_count获取市场股票数量，验证连接有效性。
实时行情数据获取
pytdx提供get_security_quotes方法获取单只或多只股票的实时行情。参数为市场代码和代码列表。市场代码：0为深圳，1为上海。期货市场代码不同，中金所为47，上期所为30，大商所为29，郑商所为28。
# 股票实时行情
stocks = [(0, '000001'), (1, '600000')]
quotes = api.get_security_quotes(stocks)
for q in quotes:
print(q['code'], q['price'], q['last_close'], q['open'], q['high'], q['low'], q['vol'], q['amount'])
返回字段包括：price当前价，last_close昨收，open今开，high最高，low最低，vol成交量（手），amount成交额（元），bid1~bid5买五档，ask1~ask5卖五档，bid_vol1~bid_vol5买量，ask_vol1~ask_vol5卖量。
期货行情获取类似，市场代码需正确。
# 期货实时行情，以上期所螺纹钢为例
futures = [(30, 'rb2310')]
quotes = api.get_security_quotes(futures)
for q in quotes:
print(q['code'], q['price'], q['open'], q['high'], q['low'], q['vol'], q['amount'])
注意期货代码格式：品种字母+合约月份，如rb2310。郑商所代码为品种字母+三位数字，如TA310。
行情订阅与轮询机制
通达信API不提供推送订阅，需轮询。轮询频率影响延迟和服务器负载。股票行情每3秒一次，期货行情每0.5秒一次。高频策略需用多线程或异步IO。
import time
from pytdx.hq import TdxHq_API
api = TdxHq_API()
api.connect('119.147.212.81', 7709)
symbols = [(0, '000001'), (1, '600000'), (30, 'rb2310')]
while True:
try:
quotes = api.get_security_quotes(symbols)
for q in quotes:
# 处理实时数据，写入队列或数据库
print(f"{q['code']} {q['price']} {q['vol']}")
except Exception as e:
print(f'获取失败: {e}')
# 重连逻辑
api.disconnect()
time.sleep(1)
api.connect('119.147.212.81', 7709)
time.sleep(0.5)
轮询频率过高可能导致服务器断开。建议单服务器每秒不超过10次请求。多品种时，分批获取，每批不超过50只。
数据存储与清洗
实时行情需持久化。选择时序数据库如InfluxDB，或关系数据库如PostgreSQL。写入前清洗数据：过滤价格为0的无效记录，检查成交量突变，处理涨跌停板。
import pandas as pd
from sqlalchemy import create_engine
engine = create_engine('postgresql://user:pass@localhost:5432/market')
def save_quotes(quotes):
df = pd.DataFrame(quotes)
df = df[df['price'] > 0]
df = df[df['vol'] >= 0]
df['timestamp'] = pd.Timestamp.now()
df.to_sql('realtime_quotes', engine, if_exists='append', index=False)
存储字段包括时间戳、代码、市场、最新价、成交量、成交额、五档买卖。注意时区统一为交易所时区。
量化交易策略集成
实时行情驱动策略。常见策略：双均线、布林带、动量。策略接收实时行情，计算信号，发送订单。下单接口可用券商API或CTP。
class DualMovingAverage:
def __init__(self, short_window=5, long_window=20):
self.short = short_window
self.long = long_window
self.prices = []
def on_tick(self, price):
self.prices.append(price)
if len(self.prices) < self.long:
return None
short_ma = sum(self.prices[-self.short:]) / self.short
long_ma = sum(self.prices[-self.long:]) / self.long
if short_ma > long_ma and self.prices[-2] <= self.prices[-1]:
return 'BUY'
elif short_ma < long_ma and self.prices[-2] >= self.prices[-1]:
return 'SELL'
return None
期货策略需考虑保证金和杠杆。股票策略考虑T+1限制。策略回测用历史数据，实盘用实时数据。注意滑点和手续费。
性能优化与风险控制
多线程轮询提高吞吐量。每个线程负责一组品种。使用队列传递数据。避免GIL限制，可用多进程。网络延迟优化：选择就近服务器，使用TCP_NODELAY。
风险控制：设置最大持仓，单笔亏损限额，日内回撤限制。行情断线自动重连。数据异常报警。
import threading
import queue
q = queue.Queue()
def poll_worker(symbols):
api = TdxHq_API()
api.connect('119.147.212.81', 7709)
while True:
quotes = api.get_security_quotes(symbols)
for quote in quotes:
q.put(quote)
time.sleep(0.5)
threads = []
for i in range(4):
t = threading.Thread(target=poll_worker, args=(symbols[i::4],))
t.start()
threads.append(t)
期货夜盘行情需单独处理。股票集合竞价数据特殊。行情数据与交易信号解耦。
常见问题与解决方案
连接失败：更换服务器IP和端口。获取数据为空：检查市场代码和代码格式。返回延迟大：减少单次请求数量，增加轮询线程。数据错乱：检查返回字段索引，不同版本pytdx字段可能不同。期货代码错误：参考交易所合约代码规则。内存泄漏：定期重启进程。
通达信API获取实时行情是量化交易的基础。掌握pytdx连接、行情获取、轮询、存储、策略集成，即可构建稳定实盘系统。股票期货通用，注意市场差异。
声明
转载声明：欢迎分享本文，转载请注明出处！
点击复制：