Python 实现股票 + ETF 统一实时订阅｜腾讯云服务行情采集实践

一、业务开发常见需求

在腾讯云服务器搭建量化监控、行情采集服务时，经常需要同时监听 A 股个股与 ETF 行情。个股用于跟踪单标的资金波动，ETF 用来判断板块整体走势，两类数据结合才能完成完整的盘中分析与回测工作。

很多开发者初期会分别搭建两套 WebSocket 连接，或是采用定时轮询拉取数据。前者重复占用云带宽资源，代码冗余难维护；后者容易丢失瞬时 Tick，回测数据失真。这里分享单链路统一订阅方案，仅维护一条长连接即可批量接收多品种行情，部署在 CVM 长期运行稳定性更高。

二、统一标的管理设计思路

个股、ETF 行情返回的现价、成交量、时间戳等字段基本一致，无需拆分采集链路。我们采用统一配置列表管理全部监控标的，通过 type 字段区分品种，新增、删减标的仅修改配置，底层通信代码无需改动，适配云服务迭代开发。

# 单段完整可运行代码
import websocket, json
watch_list = [{"code":"600000","type":"stock"},{"code":"510300","type":"etf"}]
def on_open(ws):ws.send(json.dumps({"cmd":"subscribe","symbol_list":[i["code"] for i in watch_list]}))
def on_msg(ws,msg):data=json.loads(msg);print(f"标的{data['symbol']} 实时价：{data['price']}")
if __name__ == "__main__":
    cli = websocket.WebSocketApp("wss://quote.alltick.co/quote-b-ws-api",on_open=on_open,on_message=on_msg)
    cli.run_forever()

脚本部署在腾讯云 CVM、轻量应用服务器均可，仅维持一条 WebSocket 链路，带宽占用更低，批量订阅逻辑简洁。

三、云服务器长期部署关键优化点

1. 标准化证券代码：统一清洗 SH/SZ 市场前缀，写入云数据库后便于批量回测查询；
2. 时间戳统一转换：全部行情时间转为北京时间，避免分时 K 线、因子时序错位；
3. 断线自动重连：本地缓存订阅清单，网络波动重连后自动恢复批量订阅，无需人工介入；
4. 流量资源管控：单链路采集减少出口带宽消耗，降低云流量开销。

四、模块化工程目录（拓展推荐）

如需叠加数据持久化、实时告警、前端看板功能，可拆分模块部署，支持腾讯云容器弹性扩容：

• config.py：统一维护股票、ETF 监控标的配置
• websocket.py：封装长连接、批量订阅逻辑
• handler.py：行情解析、品种差异化指标计算
• storage.py：行情数据写入云时序数据库
• main.py：程序统一启动入口

五、落地总结

在腾讯云环境搭建多品种行情采集服务，单 WebSocket 统一订阅是更轻量化的方案。统一标的配置简化迭代成本，单链路设计节约云带宽资源，搭配标准化数据处理与重连容错逻辑，可稳定支撑盘中实时监控与离线历史回测。

后续拓展指数、可转债等品种仅更新标的列表，无需重构网络通信逻辑，适合个人量化研究与小型自动化策略服务。