Python WebSocket 实现 A股港股美股统一接入

概述

在跨市场量化研究、策略原型验证过程中，经常需要同时观测A股、港股、美股多个市场标的的实时盘口数据。常规方式是通过多个行情终端分别查看各市场价格，人工切换界面会引入观测时差，不利于实时信号捕捉，也不方便直接把原始行情喂给策略模型做运算。

此前我在研究过程中遇到交易时段重叠场景：港股集合竞价阶段与美股盘前时段重合，依靠多终端人工比对价格，出现了信息滞后，影响到策略信号的观测判断。基于这个问题，我尝试通过外部行情API，将三个市场的实时数据流统一接入Python环境，为实盘模拟、信号监控、原型回测前置数据采集提供数据源支撑。本文主要分享API选型评估逻辑与可直接调试的接入代码。

多市场行情API的核心评估维度

市面上支持境外市场的行情接口较多，但面向量化研究使用，不能只看宣传的市场覆盖范围，需要从数据可用性角度做四项评估，优先级从高到低：

多市场原生覆盖能力 需要同时支持A股、港股、美股标的。部分数据源仅覆盖单一海外市场，其余市场需要对接不同供应商。多数据源意味着需要维护多套接入逻辑，会带来字段对齐、时间戳统一、多源数据校验等额外工作量，增加研究与回测的数据处理成本。
长连接稳定性与推送延迟表现 开盘、集合竞价阶段报文吞吐量高，是接口的压力测试窗口。对于量化策略来说，WebSocket链路卡顿、主动断开，会直接造成实时行情断流，实时监控、事件驱动的策略原型将无法正常运行，甚至产生错误的信号样本，干扰后续模型验证。
跨市场数据结构一致性 该指标对回测与实盘逻辑复用非常关键。若不同市场返回的字段命名、报价精度、时间戳标准不统一，业务层就需要编写大量分支逻辑做数据清洗转换。实盘采集和历史回测的代码逻辑容易出现分叉，会增大实盘与回测结果不一致的风险。理想方案是多市场复用同一套协议与字段体系，降低代码维护成本，保证回测‑实盘逻辑尽量对齐。
服务配额与计费模式 在上述三项数据质量指标满足研究需求的前提下，再评估调用额度、计费成本。即便成本低廉，如果稳定性、数据标准化不达标，也不适合用于量化研究与原型验证。

经过多组对比测试，本次实践选用 AllTick API。该接口将A股、港股、美股、外汇等品种统一封装在同一套WebSocket协议下，订阅交互逻辑完全一致，无需为不同交易市场开发独立解析模块，便于统一构建数据采集层，减少实盘与回测之间的适配工作量。

Python接入实现：单WebSocket连接订阅三大市场实时行情

下面示例基于websocket‑client库实现长连接，在同一个会话中完成A股、港股、美股多标的实时行情订阅。代码可用于本地研究调试，参考官方接入文档编写。

import json
import websocket

# 替换为个人实际token
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):
    """连接建立后，提交行情订阅请求"""
    print("WebSocket connected")
    subscribe_req = {
        "cmd_id": 22002,
        "seq_id": 1,
        "trace": "quant_research_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()

脚本运行后，控制台持续输出多市场实时推送数据。各市场统一采用毫秒级时间戳，报价字段体系一致，不需要大量条件分支区分市场类型。

研究工程提示：

上述为最小演示代码，未包含断线重连、异常告警、数据持久化逻辑。若用于长时间数据采集、策略监控，建议外层增加循环重连、退避策略，抵御网络抖动造成的链路中断；

on_message回调函数内，可以扩展实现原始行情落盘、实时指标计算、事件驱动信号判断，采集的原始数据也可用于后续离线回测样本补充；

需要关注接口限流约束，避免一次性订阅过量标的触发服务限制。

实际研究场景效果

脚本启动后，A股、港股、美股行情汇聚至同一程序输出，不再依赖多终端人工切换。在港股集合竞价等高波动时段，数据推送未出现显著滞后，返回报价和网页行情源基本对齐。

对于量化研究者，这套方案的价值不只简化观测操作，更关键在于获取标准化的原始数据流。原始行情可以直接供给策略模型做实时运算，同时采集的数据样本能够补充回测数据集，减少“回测漂亮、实盘失效”当中的数据适配类偏差。

小结

开展跨市场量化研究，股票API的选择不能只看能否拿到价格，市场覆盖、长连接稳定性、跨市场数据结构一致性，会直接影响策略原型、数据采集与回测工作的可靠性。利用WebSocket长连接可以将A股、港股、美股行情统一采集，减少多源数据的适配成本。

本次实践所使用的AllTick API依靠统一协议降低多市场数据处理的开发负担，适合作为跨市场策略原型开发、行情样本采集的外部数据源。

风险声明：本文仅为技术与量化研究工具分享，文中接口与代码仅用于学习研究，不构成任何投资建议。策略的有效性需要结合完整历史数据充分回测验证。

Python WebSocket 实现 A股港股美股统一接入

概述

多市场行情API的核心评估维度

Python接入实现：单WebSocket连接订阅三大市场实时行情

实际研究场景效果

小结

评论