通达信行情IP突然连不上了?手把手教你做一套「自动探活+选优」的动态IP方案

做过 A 股量化的人，大概率都绕不开一个老朋友——通达信行情接口。它免费、快、数据全，从日线到分时、从K线到资金流都能抓。无论是 pytdx、还是各种二次封装的库，底层都连的是通达信那套 HQ 协议、TCP 7709 端口的服务器。

但用着用着，你大概率会撞上同一个噩梦：

某天早上程序跑起来，突然全线报错，一根数据都拉不下来。

你查代码、查网络、查防火墙，折腾半天才发现——你硬编码的那台服务器 IP，挂了。

这篇文章就讲清楚一件事：为什么通多了达信的 IP 会失效，以及怎么写一套"自动探活、自动选优、自动故障转移"的动态 IP 方案，让你的量化程序从此告别"一夜之间全崩"。

一、先说结论：不是协议挂了，是服务器挂了

很多人第一反应是"是不是通达信把接口封了"。其实绝大多数情况不是。

底层协议从来没变——还是那个 TCP 7709，还是那套 HQ 命令字。真正出问题的是你连的那一台具体服务器。

通达信的行情服务器是分布在全国各地的一堆机器，IP 一抓一大把。但这些机器的稳定性参差不齐：

• 有的服务器被运营商区域性屏蔽了；
• 有的机房迁移、下线、换段；
• 有的高峰期连接数打满直接拒绝新连接；
• 有的就是单纯抽风、超时、丢包。

所以问题的本质是：你把程序绑死在了一台随时可能消失的机器上。 这台一倒，你整个程序就跟着倒了。

二、传统写法为什么脆弱

绝大多数教程和老旧代码里，连接通达信是这样的：

from pytdx.hq import TdxHq_APIapi = TdxHq_API()api.connect('60.12.136.250', 7709)   # ← 硬编码单台 IPdata = api.get_security_bars(9, 0, '000001', 0, 100)

这种写法有三个致命问题：

1. 单点依赖。 整个程序的生命周期，系在 60.12.136.250 这一台机器上。它一挂，全军覆没，没有 Plan B。

2. 没有探活。 你根本不知道这台服务器现在是不是活的，上来就 connect，失败了才发现——而且往往要等到超时（默认可能几十秒）才死心。

3. IP 列表是静态的。 pytdx 自带的 select_best_ip() 依赖一份打包进去的、常年不更新的 IP 列表。那份列表里很多 IP 早失效了，选出来的"最优"IP 可能根本连不上。

换句话说，传统写法是把"哪台服务器可用"这个会变的客观事实，当成了一成不变的常量。这在生产环境里是行不通的。

三、正确思路：服务器池 + 探活 + 选优 + 故障转移

真正稳定的做法，是把"选 IP"这件事当成一个动态决策问题来对待。核心是四个动作：

1. 准备一个候选服务器池（不要只一台，要几十台）；
2. 连接前先并发探活（快速知道池里哪些通、哪些不通）；
3. 按延迟自动排序（优先用最快的）；
4. 失败自动转移（这台连不上，顺延下一台）。

这四步合起来，就是一套完整的动态 IP 方案。下面我直接给你一套经过实战验证的源码实现。

四、动态IP方案

1. 准备服务器池

先准备一份尽量新鲜、尽量多的候选 IP 列表。可以从公开的通达信服务器列表里抓，也可以维护一份手动更新的清单。关键是要多——几十台起步，这样就算一半挂了也还有的选。

FALLBACK_HOSTS = ("116.205.183.150:7709","116.205.171.132:7709","111.230.186.52:7709","129.204.230.128:7709","110.41.2.72:7709","159.75.29.111:7709",# ... 几十台，建议优先用云厂商(腾讯云/华为云)段的IP，相对稳定)

一个小经验：云厂商 IDC 的 IP 比运营商骨干网的 IP 稳定得多。腾讯云、华为云那些段的机器，常年在线率明显高于某些地方机房的 IP。

2. 单台探活：TCP 握手 + 测延迟

探活不需要真的去拉一根 K 线，一次 TCP 握手就够了——只要 7709 端口能建立 TCP 连接，这台服务器基本就是活的。顺便把握手耗时记下来，这就是"延迟"。

import socketimport timefrom dataclasses import dataclass@dataclassclassHostProbeResult:    host: str    ok: bool    latency_ms: float | None = None    error: str | None = Nonedefprobe_host(host: str, *, timeout: float = 1.2) -> HostProbeResult:"""探活单台: TCP握手能成功就算活, 顺便测延迟。"""    address, port = host.rsplit(":", 1)    started = time.perf_counter()try:with socket.create_connection((address, int(port)), timeout=timeout):            latency_ms = round((time.perf_counter() - started) * 1000.0, 2)return HostProbeResult(host=host, ok=True, latency_ms=latency_ms)except OSError as exc:# 超时/拒绝/路由不通 → 都算死return HostProbeResult(host=host, ok=False, error=type(exc).__name__)

超时设 1.2 秒 是个比较平衡的值——既不会因为某台死机拖太久，又能让正常的服务器从容握上手。

3. 并发探活：32 线程同时 ping

这是整个方案里最关键的性能优化。如果你的池子有 43 台，串行一台台试，最坏要等 43 × 1.2 ≈ 51 秒，程序启动慢得像蜗牛。

用线程池并发探活，43 台同时测，一两秒就能把全池扫完：

from concurrent.futures import ThreadPoolExecutor, as_completeddefprobe_hosts(hosts, *, timeout=1.2, max_workers=32):"""并发探活: 32线程同时ping整个池子。"""    candidates = list(dict.fromkeys(hosts))  # 去重保序ifnot candidates:return []    worker_count = min(max(1, max_workers), len(candidates))    results = []with ThreadPoolExecutor(max_workers=worker_count,                            thread_name_prefix="tdx-probe") as executor:        futures = [executor.submit(probe_host, h, timeout=timeout)for h in candidates]for future in as_completed(futures):            results.append(future.result())return results

4. 按延迟排序选优

探活完，把结果整理一下：活的按延迟从快到慢排前面，死的全压到最后。

defsort_hosts_by_latency(hosts, *, timeout=1.2, max_workers=32):"""活的按延迟升序排前面, 死的丢最后。"""    candidates = list(dict.fromkeys(hosts))    results = probe_hosts(candidates, timeout=timeout, max_workers=max_workers)    reachable = sorted(        (r for r in results if r.ok),        key=lambda r: r.latency_ms    )    reachable_hosts = {r.host for r in reachable}    unreachable = [h for h in candidates if h notin reachable_hosts]return [r.host for r in reachable] + unreachable

返回的这个有序列表，就是你后续连接时从前往后依次尝试的顺序。

5. 连接时自动故障转移

最后，把上面这套串进连接逻辑。从最快的开始连，连上就用；某台失败，自动切到列表里的下一台，直到找到一台能用的为止。

defconnect_with_failover(hosts, max_try=None):"""从排序后的池子里依次尝试连接, 自动故障转移。"""    ordered = sort_hosts_by_latency(hosts)  # 先探测+排序    tried = ordered[:max_try] if max_try else orderedfor host in ordered:try:            api = TdxHq_API()            addr, port = host.rsplit(":", 1)if api.connect(addr, int(port)):                print(f"✅ 连上 {host}")return apiexcept Exception as e:            print(f"❌ {host} 连接失败: {e}")continueraise RuntimeError("所有候选服务器均不可用")

最后

记住那句话：别把会变的东西写死。 服务器的生与死是它的事，但你的程序应该永远有下一台可以连。