跨市场金融行情数据接入实践：使用Python WebSocket完成A股、港股、美股实时数据采集

标签：Python、WebSocket、股票API、金融数据、数据采集

摘要

在金融业务开发、量化研究场景中，经常需要同时获取A股、港股、美股多市场实时盘口数据。多终端人工查看的模式不仅效率低下，还会引入观测时差，不利于业务系统、策略模型消费实时数据流。本文从工程落地角度，分享多市场行情API的选型评估思路，同时提供可调试的Python WebSocket示例代码，实现单条长连接订阅三大市场行情，为数据处理、业务原型开发提供数据基础。

背景

在做多市场金融数据处理、量化原型验证时，往往需要并行跟踪A股、港股、港股、美股的标的行情。传统方案依靠多款行情客户端分别查看不同市场，当不同市场交易时段发生重叠，例如港股集合竞价与美股盘前时段叠加时，人工切换界面比对价格会带来不可避免的时间差。

该问题会带来两方面工程影响：一是人工观测无法对接程序，行情数据不能直接输入下游计算模块；二是人为延迟会干扰信号判断，不利于模型验证与业务逻辑测试。因此考虑引入外部行情API，将多市场实时行情统一收敛至Python服务，完成行情接收、后续可扩展数据落盘与指标计算。

多市场行情API工程化评估要点

市面上不少服务宣称支持全球金融行情，但是面向开发与业务原型场景，需要从工程可靠性角度建立评估标准，优先级从高到低：

1. 多市场原生覆盖能力
   需要原生支持A股、港股、美股三类标的。部分API仅覆盖单一海外市场，其余市场需要引入第三方数据源。多数据源架构会带来多套SDK、字段不一致、时间戳对齐、多源校验等额外开发工作量，提升整体系统的维护成本。

2. 长连接稳定性与推送时延
   开盘、集合竞价阶段行情报文流量陡增，是WebSocket链路的压力测试阶段。若链路出现卡顿、非正常断开，会直接造成实时行情断流，基于事件驱动的业务逻辑、量化原型将失效，甚至产生错误样本，影响业务调试。在云环境部署服务时，还需要考虑网络环境对长连接的影响。

3. 跨市场数据结构标准化
   该点直接影响业务代码复杂度，也关系测试环境与模拟逻辑的一致性。如果不同市场返回的字段名称、价格精度、时间戳格式存在差异，业务层需要编写大量分支逻辑做数据清洗转换，容易造成不同市场处理逻辑不一致，扩大开发、测试成本。优先选择多市场复用同一套协议、字段规范的接口。

4. 调用配额与计费策略
   在稳定性、数据规范性满足业务需求的前提下，再评估额度与成本。即使服务价格低廉，如果数据质量、链路稳定性不达标，也不适合用于业务原型、数据采集场景。

经过对比测试，本次实践选用AllTick API。该服务将A股、港股、美股、外汇等数据封装在同一套WebSocket协议，订阅交互逻辑统一，不需要针对不同市场开发独立解析逻辑，便于在应用侧搭建统一的数据采集层，降低多市场适配的开发成本。

Python代码实现：单WebSocket长连接订阅多市场实时行情

示例基于websocket‑client库，实现单条长连接同时订阅A股、港股、美股行情。代码可在本地或者云服务器环境调试运行，参考官方接入文档编写。

import json
import websocket

# 替换为实际的访问令牌
TOKEN = "YOUR_TOKEN"

WS_URL = (
    "wss://quote.alltick.co/quote-stock-b-ws-api"
    f"?token={TOKEN}"
)

# A股、美股、港股测试标的列表
symbols = [
    {"code": "688036.SH"},
    {"code": "AAPL.US"},
    {"code": "700.HK"},
]


def on_open(ws):
    """WebSocket握手完成，发起行情订阅请求"""
    print("WebSocket connected")
    subscribe_req = {
        "cmd_id": 22002,
        "seq_id": 1,
        "trace": "huaweicloud_demo",
        "data": {
            "symbol_list": [
                {"code": item["code"], "depth_level": 1}
                for item in symbols
            ]
        }
    }
    ws.send(json.dumps(subscribe_req))


def on_message(ws, message):
    """行情推送回调，可扩展数据落盘、指标计算业务逻辑"""
    try:
        payload = json.loads(message)
        # 扩展点：写入数据库、消息队列、实时指标运算
        print(payload)
    except json.JSONDecodeError:
        print("Invalid JSON message")


def on_error(ws, error):
    print(f"WebSocket error: {error}")


def on_close(ws, code, msg):
    print(f"WebSocket closed, code:{code}, msg:{msg}")


if __name__ == "__main__":
    ws_app = websocket.WebSocketApp(
        WS_URL,
        on_open=on_open,
        on_message=on_message,
        on_error=on_error,
        on_close=on_close
    )
    ws_app.run_forever()

脚本运行后，控制台持续输出多市场实时行情数据，全部市场统一毫秒级时间戳，报价字段体系统一，无需大量条件分支区分市场。

📌工程部署提示

1. 示例仅为最小Demo，缺少断线重连、异常告警、持久化逻辑；部署在云服务器做长期采集任务时，建议增加循环重连、退避重试机制，应对公网网络抖动；
2. on_message回调中可以扩展逻辑，将原始行情写入消息队列、时序数据库，供下游业务模块消费；
3. 注意接口限流约束，避免一次性订阅大量标的触发服务限制；
4. 云环境运行需要确认出站网络策略放开wss协议访问，防止长连接被防火墙策略阻断。

落地效果与应用场景

程序启动后，A股、港股、美股行情汇总至同一进程输出，不再依赖多个人工终端。港股集合竞价等高流量时段，数据推送无明显滞后，返回报价与公开网页行情基本对齐。

该采集方案可以应用于多种场景：金融原型系统开发、量化研究实时数据采集、行情样本数据集构建；标准化的原始数据流可以减少多源数据适配带来的逻辑差异，提升开发调试效率。

总结

构建跨市场金融行情采集服务，不能只关注接口是否可以获取价格，市场覆盖度、长连接稳定性、跨市场数据格式一致性，会直接影响上层业务与原型系统的可靠性。借助WebSocket长连接，可以实现A股、港股、美股行情统一采集，降低多市场数据处理的适配成本。

本次实践使用的AllTick API依靠统一协议，简化多市场数据接入工作，适合用于金融数据原型开发、行情样本采集等技术验证场景。