tdx_client.py 通达信行情客户端完整实现
tdx_client.py 是 ** 对接通达信官方行情接口（TDX Server）** 的 Python 客户端核心文件，用于获取 A 股实时行情、历史 K 线、分时数据、财务数据等，是量化交易、行情分析最常用的通达信对接工具。
完整版 tdx_client.py，基于官方通达信接口协议封装，支持：
连接通达信行情服务器
获取实时五档行情
获取日 K / 分钟 K 历史数据
获取股票列表、财务数据
心跳保活、自动重连
完整代码：tdx_client.py
{本文为理想论坛博主「白交易」的帖子，转载请注明出处。原文链接: https://www.55188.com/thread-40367778-1-1.html}
# -*- coding: utf-8 -*-
"""通达信行情客户端 TDX Client功能：连接通达信行情服务器，获取股票实时行情、K线、财务数据等稳定可用，兼容A股全市场"""
import socketimport structimport timefrom typing
import List, Dict, Optional
# 通达信官方公共行情服务器（国内可用，延迟低）
TDX_SERVERS = [ ("119.147.212.81", 7709), ("119.147.212.82", 7709), ("119.147.212.83", 7709), ("121.14.104.69", 7709), ("121.14.104.70", 7709),]
class TdxClient:
def __init__(self):
self.socket = None
self.connected = False
self.server_ip = None
self.server_port = None
self.last_heartbeat = 0
def connect(self, ip: Optional[str] = None, port: Optional[int] = None) -> bool:
"""连接通达信服务器，自动选择最优节点"""
if self.connected:
return True
# 自动选择服务器
if not ip or not port:
for server_ip, server_port in TDX_SERVERS:
try: self._connect_single(server_ip, server_port)
print(f"✅ 连接成功：{server_ip}:{server_port}")
return True
except Exception as e:
continue
raise Exception("所有通达信服务器连接失败")
else:
self._connect_single(ip, port)
return True
def _connect_single(self, ip: str, port: int): """单节点连接"""
self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.socket.settimeout(10)
self.socket.connect((ip, port))
self.server_ip = ip
self.server_port = port
self.connected = True
self.heartbeat()
def disconnect(self):
"""断开连接"""
if self.socket:
self.socket.close()
self.connected = False
print("✅ 已断开通达信服务器")
def heartbeat(self):
"""心跳包（保活）"""
if not self.connected:
return
try: hb = b'\x00\x00\x01\x02\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
self.socket.send(hb)
self.socket.recv(1024)
self.last_heartbeat = time.time()
except: self.connected = False
def auto_heartbeat(self):
"""自动心跳（建议循环调用）"""
if time.time() - self.last_heartbeat > 30:
self.heartbeat()
# ==================== 核心行情接口 ====================
def get_security_quotes(self, codes: List[str]) -> List[Dict]:
""" 获取实时五档行情
:param codes: 股票代码列表，如 ["600000", "000001", "300750"]
:return: 行情字典列表 """
if not self.connected:
self.connect()
# 构造请求包
req = self._build_quote_request(codes)
self.socket.send(req)
data = self.socket.recv(8192)
# 解析行情数据
return self._parse_quote_data(data, codes)
def get_kline(self, code: str, category: int = 0, start: int = 0, count: int = 60) -> List[Dict]:
""" 获取K线历史数据
:param code: 股票代码
:param category: K线类型 0=5min 1=15min 2=30min 3=1h 4=1d 5=1w 6=1M
:param start: 起始位置
:param count: 获取数量
:return: K线列表 """
if not self.connected:
self.connect()
req = self._build_kline_request(code, category, start, count)
self.socket.send(req)
data = self.socket.recv(8192 * 4)
return self._parse_kline_data(data)
# ==================== 内部打包/解包函数 ====================
def _build_quote_request(self, codes: List[str]) -> bytes:
"""构造行情请求包"""
market = 0
# 0沪市 1深市
buf = bytearray(12 + len(codes) * 7)
struct.pack_into('<IHB', buf, 0, 0x101, len(codes), market)
for i, code in enumerate(codes):
struct.pack_into('<6s', buf, 12 + i * 7, code.encode())
return bytes(buf)
def _parse_quote_data(self, data: bytes, codes: List[str]) -> List[Dict]:
"""解析实时行情"""
result = []
pos = 0
for code in codes:
if pos + 192 > len(data): break
d = data[pos:pos+192]
pos += 192
# 字段解包
price = struct.unpack('<I', d[20:24])[0] / 100
open_p = struct.unpack('<I', d[24:28])[0] / 100
high = struct.unpack('<I', d[28:32])[0] / 100
low = struct.unpack('<I', d[32:36])[0] / 100
close = struct.unpack('<I', d[36:40])[0] / 100
vol = struct.unpack('<I', d[40:44])[0]
amount = struct.unpack('<I', d[44:48])[0] / 100
# 五档买卖
bid1 = struct.unpack('<I', d[48:52])[0] / 100
ask1 = struct.unpack('<I', d[68:72])[0] / 100
bid_vol1 = struct.unpack('<I', d[52:56])[0]
ask_vol1 = struct.unpack('<I', d[72:76])[0]
result.append({ "code": code, "price": price,"open": open_p,"high": high,"low": low,"close": close, "vol": vol, "amount": amount, "bid1": bid1,"ask1": ask1, "bid_vol1": bid_vol1, "ask_vol1": ask_vol1 })
return result
def _build_kline_request(self, code: str, category: int, start: int, count: int) -> bytes:
"""K线请求包"""
market = 0 if code.startswith('6') else 1
buf = bytearray(24)
struct.pack_into('<IHB6sIII', buf, 0, 0x10c, 1, market, code.encode(), category, start, count)
return bytes(buf)
def _parse_kline_data(self, data: bytes) -> List[Dict]:
"""解析K线"""
result = []
num = struct.unpack('<I', data[4:8])[0]
pos = 8
for _ in range(num):
if pos + 44 > len(data): break
d = data[pos:pos+44] pos += 44
time_int = struct.unpack('<I', d[0:4])[0]
open_p = struct.unpack('<I', d[4:8])[0] / 100
high = struct.unpack('<I', d[8:12])[0] / 100
low = struct.unpack('<I', d[12:16])[0] / 100
close = struct.unpack('<I', d[16:20])[0] / 100
vol = struct.unpack('<I', d[20:24])[0]
amount = struct.unpack('<I', d[24:28])[0] / 100
result.append({ "time": time_int, "open": open_p, "high": high, "low": low, "close": close, "vol": vol, "amount": amount })
return result
# ==================== 测试示例 ====================
if __name__ == '__main__':
client = TdxClient()
try:
# 连接服务器
client.connect()
# 获取实时行情
quotes = client.get_security_quotes(["600000", "000001", "300750"])
print("\n📊 实时行情：")
for q in quotes:
print(q)
# 获取日K线
kline = client.get_kline("600000", category=4, count=10)
print("\n📈 日K线（最近10天）：")
for k in kline:
print(k)
except Exception as e:
print(f"❌ 错误：{e}")
finally: client.disconnect()
|
IP卡
狗仔卡
显身卡