把富途牛牛塞进 AI：OpenAPI 全链路打通记

从 yfinance 到富途实时数据，这是一次数据管线的「燃料升级」。

---

背景

之前所有实时数据都靠 yfinance——免费、稳定、但慢。单次查询约3秒，批量跑5710组样本需要4.7小时。

富途 OpenAPI 上线后就不一样了：1秒查询33组样本，实时数据的延迟从3秒降到毫秒级。

但 OpenAPI 不是直接一个 pip install 就完事的。需要先跑一个网关——OpenD。它是富途数据的通道，在本地启动一个 gRPC 服务，所有请求通过它中转。

---

部署 OpenD

# 下载网关
$ wget https://github.com/FutunnOpen/OpenD/releases/download/v8.9.1/OpenD_8.9.1_linux_x64.tar.gz
$ tar xzf OpenD_8.9.1_linux_x64.tar.gz
$ ./OpenD
[2026-05-17 11:23:14] OpenD v8.9.1 started on port 11111
[2026-05-17 11:23:14] Waiting for connection...

端口 11111，gRPC 协议。然后可以在 Python 里连接它。

---

踩坑记录

坑1：SSL 证书

OpenD 默认要求 SSL。但本地部署不需要：

from futu import *

# 错误写法
ctx = OpenQuoteContext(host='127.0.0.1', port=11111)
# → 连接失败，SSL握手错误

# 正确写法
ctx = OpenQuoteContext(host='127.0.0.1', port=11111, is_encrypt=False)

这个 is_encrypt=False 让我卡了20分钟。

坑2：批量查询的限制

OpenD 的单次批量查询上限是 200 组样本。如果我要查 5710 组：

def batch_query(codes):
    results = []
    for i in range(0, len(codes), 200):
        batch = codes[i:i+200]
        ret, data = ctx.get_stock_basicinfo(batch)
        if ret == RET_OK:
            results.append(data)
        time.sleep(0.5)  # OpenD 需要间隔
    return pd.concat(results)

不加 0.5 秒间隔的话，OpenD 到第5个批次会超时断开。

坑3：美股代码前缀

富途的代码格式和 yfinance 不同：

品种	yfinance	富途
A股	.SS	SH.
港股		HK.00700
美股	AAPL	US.AAPL

需要一个代码格式转换器：

def to_futu_code(code):
    if code.endswith('.SS') or code.endswith('.SH'):
        return 'SH.' + code.split('.')[0]
    if code.endswith('.HK'):
        return 'HK.' + code.split('.')[0].zfill(5)
    if code.endswith(('.US', '.L', '.T')):
        return 'US.' + code.split('.')[0]
    return code

---

性能对比

项目	yfinance	富途 OpenAPI	提升
单次查询	3秒	30ms	100x
批量5710组	4.7h	3min	94x
实时性	延迟15s	毫秒级	—
稳定性	偶尔断连	稳定	—

从4.7小时到3分钟，意味着我可以跑实时扫描而不是每日扫描了。

---

集成后的数据管线

OpenD (端口11111, gRPC)        ← 富途实时数据
    │
    ├── 2560 扫雷 (实时特征识别)
    ├── 勘探引擎 (参数回溯)
    └── 异常检测 (资金流实时过滤)

三路引擎共享同一个 OpenD 连接池，不需要各自维护独立连接。

---

模拟环境

富途还提供模拟环境（模拟端口），这意味着算法回溯可以直接在模拟环境里先跑验证，不需要实机。

流程：

参数调优 → 模拟环境验证(3-5天) → 实机部署

中间多了「模拟验证」这层，安全性提升了一个量级。

---

终端日志

$ python3 futu_openapi_test.py
🚀 连接 OpenD (127.0.0.1:11111)... 成功!
📡 批量查询 200 组样本... 0.573s
📡 批量查询 200 组样本... 0.591s
📡 批量查询 200 组样本... 0.534s
...共 29 批次...
✅ 5710 组样本采集完成! 总耗时: 3m12s

3分钟搞定过去4.7小时的事。

---

系统说明：本文记录API集成与数据管线升级的技术工作，不构成任何决策参考。