简介：这份资源围绕TdxHqApi.dll构建的证券行情实时监测系统展开，面向具备一定C#或Java基础、希望深入理解金融行情数据采集与解析的开发者。系统通过持久化连接行情服务器，持续获取证券代码、最新成交价、分时成交量及买卖盘挂单量价等核心指标，并采用多线程异步机制对二进制流实时解码与结构化重组，配合数据校验、断线重连与缓存补偿，整体延迟控制在毫秒级。包内共299个文件，以cs源码、zbak备份、txt说明、dll动态库、java源码、config配置、csproj工程文件及class字节码为主，另含pdf、doc等文档，压缩包约105.88MB，覆盖数据接入、协议解析与业务逻辑三层模块。已有46人学习，适合用于价格预警、技术指标计算与行情图表生成等场景的二次开发参考。
1. 拆开一个只有 2MB 的行情采集器：TdxHqApi.dll 能解决什么
如果你做过 A 股行情采集，大概率绕不开一个尴尬的现实：官方行情接口要么贵，要么权限卡得死，而网页抓取又慢又不稳定，分时数据延迟几分钟是常态。TdxHqApi.dll 这个动态库之所以在个人学习圈子里被反复提起，是因为它把通达信行情服务器的通信协议封装成了几个导出函数，调用方只需要传入服务器地址和端口，就能拿到实时行情、历史 K 线和分时成交。整个库体积很小，依赖也少，适合拿来理解行情采集的完整链路——从连接服务器、发送请求、解析二进制包，到落地成本地文件或数据库。它适合谁？适合想自己动手写采集程序、又不希望从零逆向通信协议的个人开发者。不适合谁？不适合需要 Level-2 逐笔委托、需要毫秒级撮合数据的场景，那是另一套东西。下面按“能干什么 → 怎么调 → 坑在哪”的顺序拆一遍。
2. 通信模型与接口选型：为什么是 DLL 而不是 HTTP 轮询
2.1 通达信行情服务器的通信特征
通达信行情服务器走的是 TCP 长连接，请求和响应都是二进制格式，不是 HTTP。每个请求包有固定的头部结构，包含请求类型、序列号、包体长度等字段；响应包同样有头部，后面跟具体数据。TdxHqApi.dll 做的事情，就是把这套二进制协议封装成 C 风格的导出函数，调用方不需要自己拼包和拆包。常见做法是：先调用连接函数建立 TCP 连接，然后循环调用行情请求函数，拿到结构体数组后再做后续处理。
这种模型和 HTTP 轮询的本质区别在于：HTTP 每次请求都要重新建连（除非用 keep-alive），头部开销大，而且行情服务器根本不提供 HTTP 接口。用 DLL 走原生 TCP，单次请求的往返延迟可以压到几十毫秒级别，对于分钟级或秒级采集来说完全够用。另一个好处是，DLL 内部通常维护了连接状态，断线后可以重新连接，不需要调用方自己管理 socket 生命周期。
但要注意，这个 DLL 是 32 位的，因为通达信客户端本身是 32 位程序。如果你的采集程序编译成 64 位，加载会直接失败。这是第一个选型约束：宿主进程必须是 32 位，或者用 32 位的 Python/Node 来调用。
2.2 导出函数与调用约定
DLL 的导出函数通常是 stdcall 调用约定，在 C# 里用[DllImport]声明时要加CallingConvention = CallingConvention.StdCall。在 Python 里用ctypes加载时，要指定windll而不是cdll。函数名一般包括连接、断开、获取行情数量、获取行情数据、获取 K 线数据、获取分时数据等。参数里经常出现服务器 IP 和端口，通达信常用的行情服务器端口是 7709，但不同服务器可能不同，需要自己准备一份可用服务器列表。
下面是一段 Python 加载 DLL 并建立连接的示例：
import ctypes from ctypes import wintypes # 加载 32 位 DLL，注意 Python 也必须是 32 位 tdx = ctypes.windll.LoadLibrary(r"./TdxHqApi.dll") # 声明函数原型，stdcall 用 windll 加载后默认就是 stdcall # 连接服务器：参数为 IP 字符串和端口 tdx.TdxHq_Connect.argtypes = [ctypes.c_char_p, ctypes.c_ushort] tdx.TdxHq_Connect.restype = ctypes.c_bool # 获取行情数量：参数为市场代码（0 深市，1 沪市） tdx.TdxHq_GetSecurityCount.argtypes = [ctypes.c_ubyte] tdx.TdxHq_GetSecurityCount.restype = ctypes.c_ushort ip = b"119.147.212.81" # 示例服务器，实际需替换为可用地址 port = 7709 ok = tdx.TdxHq_Connect(ip, port) if not ok: raise RuntimeError("连接行情服务器失败") count = tdx.TdxHq_GetSecurityCount(0) print(f"深市证券数量: {count}")这段代码的逻辑是：先加载 DLL，然后声明两个函数的参数类型和返回类型。TdxHq_Connect接收 IP 和端口，返回布尔值表示是否连接成功。TdxHq_GetSecurityCount接收市场代码，返回该市场的证券总数。参数说明：市场代码 0 代表深圳，1 代表上海，这是通达信体系的约定。注意 IP 和端口需要替换成实际可用的服务器，不同网络环境下可用服务器不同，这是采集能否跑通的第一道门槛。
2.3 数据包结构与解析方式
DLL 返回的行情数据通常是结构体数组，每个结构体包含股票代码、最新价、买卖五档、成交量、成交额等字段。K 线数据则包含日期、开盘、最高、最低、收盘、成交量、成交额。分时数据包含时间、价格、成交量、均价。这些结构体在 DLL 的头文件里有定义，用 ctypes 调用时需要自己定义对应的Structure类，字段顺序和类型必须和头文件完全一致，否则解析出来的数据全是乱的。
常见做法是：先用TdxHq_GetSecurityQuotes拿到实时行情，再用TdxHq_GetSecurityBars拿历史 K 线，最后用TdxHq_GetMinuteTimeData拿分时。每个函数返回的是数据条数，数据通过指针参数传出。调用方需要预先分配好缓冲区，缓冲区大小要足够容纳最大返回条数，否则会截断或报错。
3. 从连接服务器到落地 CSV：一套可复现的采集流程
3.1 准备服务器列表与连接重试
采集程序最怕的就是连不上服务器。通达信的行情服务器有很多个，但并不是每个都随时可用。我一般会准备一个文本文件，每行一个IP:端口，程序启动时逐个尝试连接，连上就停。连接失败的原因可能是服务器下线、网络不通、或者并发连接数超限。重试策略上，不要死循环重连同一个 IP，而是轮换列表，每次失败后 sleep 一两秒再试下一个。
import time def load_servers(path="servers.txt"): with open(path, "r", encoding="utf-8") as f: return [line.strip() for line in f if line.strip()] def connect_any(tdx, servers): for addr in servers: ip, port = addr.split(":") if tdx.TdxHq_Connect(ip.encode(), int(port)): print(f"已连接 {addr}") return addr time.sleep(0.5) raise RuntimeError("所有服务器均连接失败")逻辑说明：load_servers从文件读取服务器地址，connect_any遍历列表逐个尝试。参数方面，time.sleep(0.5)是给 TCP 连接留出超时时间，避免瞬间大量重试被服务器拒绝。如果所有服务器都失败，直接抛异常，不要静默继续。
3.2 拉取实时行情并写入 CSV
连接成功后，就可以拉取实时行情了。以深市为例，先获取证券数量，再分批获取行情。DLL 通常限制单次最多获取 80 条行情，所以需要循环调用。每条行情包含股票代码、最新价、买卖五档价格和量、成交量、成交额等。写入 CSV 时，建议用追加模式，每次采集打一个时间戳，方便后续按时间筛选。
import csv from datetime import datetime class Quote(ctypes.Structure): _fields_ = [ ("code", ctypes.c_char * 6), ("price", ctypes.c_float), ("buy1", ctypes.c_float), ("sell1", ctypes.c_float), ("vol", ctypes.c_int), ("amount", ctypes.c_float), ] def fetch_quotes(tdx, market=0, batch=80): total = tdx.TdxHq_GetSecurityCount(market) quotes = [] for start in range(0, total, batch): buf = (Quote * batch)() ret = tdx.TdxHq_GetSecurityQuotes(market, start, batch, buf) for i in range(ret): quotes.append(buf[i]) return quotes def save_csv(quotes, path="quotes.csv"): with open(path, "a", newline="", encoding="utf-8") as f: writer = csv.writer(f) ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S") for q in quotes: writer.writerow([ts, q.code.decode(), q.price, q.buy1, q.sell1, q.vol, q.amount])逻辑说明：Quote结构体定义了行情字段，字段顺序必须和 DLL 头文件一致。fetch_quotes先拿总数，再按 80 条一批循环获取。save_csv用追加模式写入，每行带时间戳。参数说明：market=0是深市，batch=80是单次最大条数，这个值不能随意调大，超过 DLL 内部缓冲区会出错。注意code字段是定长字符数组，需要 decode 成字符串。
3.3 历史 K 线与分时数据的采集节奏
实时行情是快照，历史 K 线和分时数据才是做分析的基础。K 线数据按日期范围请求，DLL 通常要求传入起始日期和结束日期，格式是YYYYMMDD的整数。分时数据只能拿当天的，历史分时拿不到。采集节奏上，K 线可以每天收盘后跑一次，分时可以在盘中每隔几分钟拉一次。注意不要高频请求同一只股票的分时，服务器会限制频率，触发后可能直接断开连接。
def fetch_bars(tdx, code, market, start_date, end_date, count=800): buf = (Bar * count)() ret = tdx.TdxHq_GetSecurityBars(market, code.encode(), start_date, end_date, count, buf) return [buf[i] for i in range(ret)]参数说明：code是 6 位股票代码，market是市场代码，start_date和end_date是整数日期，count是最大返回条数。常见做法是每次请求不超过 800 条，超过就分段请求。返回的Bar结构体包含日期、开盘、最高、最低、收盘、成交量、成交额。
4. 避坑与排查：连接、解析、频率控制里的血泪经验
4.1 连接成功但拿不到数据
现象：TdxHq_Connect返回 true，但调用TdxHq_GetSecurityCount返回 0 或负数。原因通常是服务器虽然 TCP 连上了，但行情服务没有正常响应，或者该服务器只提供部分市场数据。解决方法是换服务器，不要在一个 IP 上死磕。我一般会准备至少 5 个不同网段的服务器地址，轮换测试。
4.2 结构体字段错位导致数据全乱
现象：拿到的价格是几万或负数，成交量是随机大数。原因几乎都是 ctypes 的Structure字段顺序或类型和 DLL 头文件不一致。比如把c_float写成了c_double，或者漏掉了某个保留字段。解决方法是找到 DLL 配套的头文件，逐字段对照，不要凭猜测写。如果没有头文件，用dumpbin /exports看导出函数名，再结合网络上的结构体定义交叉验证。
4.3 高频请求被服务器断开
现象：程序跑了几分钟后，所有请求返回失败，重连也连不上。原因是请求频率太高，触发了服务器的限流。通达信行情服务器对单 IP 的请求频率有隐性限制，具体阈值不公开。解决方法是控制采集间隔，实时行情不要低于 3 秒一次，分时不要低于 1 分钟一次。如果确实需要更高频率，考虑多服务器轮换，但不要用同一 IP 并发。
4.4 32 位与 64 位不匹配
现象：加载 DLL 时报OSError: [WinError 193] %1 不是有效的 Win32 应用程序。原因是 Python 是 64 位，而 DLL 是 32 位。解决方法是换 32 位 Python，或者用 32 位的宿主程序调用。这个问题没有绕过的方法，只能匹配位数。
4.5 日期格式传错导致 K 线为空
现象：TdxHq_GetSecurityBars返回 0 条。原因通常是日期格式不对，比如传了字符串而不是整数，或者传了2024-01-01而不是20240101。解决方法是统一用整数日期，并且在请求前打印出来确认。
5. 进阶技巧：用本地缓存和增量更新把采集成本压下来
采集跑通之后，下一个问题就是效率。全量拉取所有股票的 K 线，每次都要几十分钟，而且大部分数据是重复的。我一般会做两层缓存：第一层是本地 SQLite，按股票代码和日期建唯一索引；第二层是增量更新，每次只请求最近 N 天的数据，用INSERT OR REPLACE写入。这样第一次全量之后，后续每天只需要几分钟就能完成更新。
import sqlite3 def init_db(path="bars.db"): conn = sqlite3.connect(path) conn.execute(""" CREATE TABLE IF NOT EXISTS bars ( code TEXT, date INTEGER, open REAL, high REAL, low REAL, close REAL, vol INTEGER, amount REAL, PRIMARY KEY (code, date) ) """) return conn def upsert_bars(conn, code, bars): rows = [(code, b.date, b.open, b.high, b.low, b.close, b.vol, b.amount) for b in bars] conn.executemany("INSERT OR REPLACE INTO bars VALUES (?,?,?,?,?,?,?,?)", rows) conn.commit()逻辑说明：init_db建表时用(code, date)做联合主键，保证同一只股票同一天只有一条记录。upsert_bars用INSERT OR REPLACE实现增量更新，重复数据自动覆盖。参数说明：date存整数，和 DLL 返回的日期格式一致，方便比较。增量更新时，先查本地最大日期，再请求该日期之后的数据，避免全量拉取。
另一个技巧是分市场、分板块采集。深市和沪市分开跑，创业板和主板分开跑，这样单次请求的数据量小，失败后重试成本低。我习惯在采集前先拉一遍证券列表，把退市和停牌的股票过滤掉，减少无效请求。
验证采集结果是否正确，最简单的办法是拿几只熟悉的股票，对比通达信客户端里的价格和 K 线。如果收盘价对不上，大概率是复权问题——DLL 返回的通常是不复权数据，需要自己根据除权除息信息做前复权或后复权。这一步没有捷径，只能老老实实处理。
从那以后我每次写采集程序，都会先把服务器列表、结构体定义、日期格式这三样东西各验证一遍，再跑全量。希望帮到你。
本文还有配套的精品资源，点击获取
