外汇行情 API、股票、贵金属、加密货币：多资产行情 API 接入与跨市场数据归一化工程实践

标签：#API #WebSocket #Python #量化 #多源数据处理 #数据集成
阅读时长：7‑9分钟
本文面向后端开发、量化策略研发人员，分享外汇、股票、贵金属、加密货币多市场行情来源的选型、接口适配与数据归一化方案，包含可复用的原型代码，为业务系统、量化研究平台构建多资产数据链路提供实践参考。

摘要

在搭建金融业务系统、量化研究平台的开发过程中，经常需要对接多品类市场行情，覆盖外汇、贵金属、加密货币、股票等不同资产。如果直接接入多家第三方行情API，不同接口在鉴权、标的编码、时间格式、推送协议、限流策略上存在大量差异。缺少统一的适配设计，不仅会增加开发维护成本，还会造成历史回测与实时业务的数据不一致问题。

本文从实际项目踩坑经验出发，梳理多市场行情接入的典型问题，给出API选型评估维度，介绍通过内部领域模型做数据归一化的架构思路，同时提供基于 Python + WebSocket 的原型实现，帮助开发者降低多源异构行情数据的对接复杂度。

一、多市场行情集成的典型痛点

此前在开发一套多资产监控原型系统时，分别接入三家外部数据源，分别负责外汇贵金属、加密货币、股票行情。各服务商接口自成体系，鉴权规则、品种编码、时间戳、返回字段互不兼容，大量开发工作消耗在数据转换适配，业务逻辑开发占比被挤压。主要问题集中在4个方面：

1. 标的编码体系不统一

相同交易品种，不同服务商编码格式存在区别。例如欧元兑美元存在 EURUSD、EUR/USD；加密货币交易对有 BTC‑USDT、BTCUSDT；股票代码需要追加交易所后缀。直接复用其他数据源的编码，会出现查询无返回、数据缺失问题。

2. 时间戳与字段数据类型异构

时间戳存在秒、毫秒、日期字符串多种形式；价格、成交量字段有的返回数值类型，有的返回字符串。未做统一类型转换，时序统计、指标计算时极易抛出异常，影响系统稳定性与数据可信度。

3. 推送机制、心跳与访问限制差异

部分API支持WebSocket长连接实时推送，部分仅支持HTTP轮询拉取。各家的心跳保活、断线重连、访问限流规则完全独立，很难实现通用客户端，多数据源并行维护成本高。

4. 各市场交易时间模型不同

加密货币7×24小时连续交易；外汇周末休市；股票具备开盘、午休、节假日停盘。如果采用固定超时做故障判定，休市阶段会产生大量无效告警，无法区分「正常休市」和「数据源真实故障」，干扰系统运维监控。

二、多资产行情API选型评估维度

选型的核心目标：降低上层业务的适配成本，减少异构数据源带来的数据不一致风险，重点从下面四个维度进行评估：

1. 鉴权能力：多类资产是否可以复用同一Token完成身份认证，减少多套凭证管理负担；
2. 协议一致性：不同资产的订阅报文、推送返回字段命名是否稳定统一；
3. 文档完备度：是否提供完整可查阅的标的编码清单、字段说明；
4. 运行约束说明：明确的访问配额、限流规则，文档对心跳、断线恢复逻辑给出规范说明。

需要客观看待：不存在完全字段统一的第三方API。即便是同一服务商，不同资产也可能拆分多个接入端点，存在局部字段差异。工程上务实方案：优先选择协议标准化程度高的服务，在应用层构建适配器，消化剩余的格式差异。

在原型验证阶段，可以使用 AllTick API 开展测试。外汇、贵金属、加密货币共用一组WebSocket接入端点，美股、港股、A股使用另一组端点，两套订阅协议保持一致，可以有效减少适配层代码开发量。

三、架构设计：内部Quote模型隔离外部数据源差异

设计价值：将外部接口变化封闭在适配层，保证量化计算、回测、业务上层逻辑不受第三方接口改动影响。

无论外部API返回格式如何，在应用内部定义独立的行情实体 Quote，只保留业务必需核心字段：数据源标识、品种代码、价格、成交量、毫秒级时间戳。

所有外部原始行情数据流入系统，第一步执行归一化转换，映射为统一的Quote对象。上层模块，包括策略运算、指标统计、回测分析、业务监控，全部消费这套内部模型，不感知第三方原始JSON结构。

该架构带来两项工程收益：

1. 切换底层数据源时，仅修改数据转换函数，上层业务、回测逻辑无需改动；
2. 新增一类交易市场，仅补充数据源配置，不需要大规模重构现有代码。

之前项目中依靠这套抽象，新增港股、美股行情，仅半天就完成调试上线。开发建议：业务与回测逻辑，禁止直接依赖第三方API原始返回结构。

四、原型实现：单进程多路WebSocket行情接收

下面为可调试原型代码，单进程下维护多条独立WebSocket连接，各市场链路相互隔离，单一条连接故障不会影响其他数据流。适合用于前期技术验证，生产环境还需要进一步增强容错、持久化能力。

import asyncio
import json
import os
import uuid
from dataclasses import dataclass

import websockets

BASE = "wss://quote.alltick.co"

FEEDS = {
    "multi": {  # 外汇、贵金属、加密货币
        "path": "/quote-b-ws-api",
        "codes": ["EURUSD", "GOLD", "BTCUSDT"],
    },
    "stock": {  # 美股、港股、A股
        "path": "/quote-stock-b-ws-api",
        "codes": ["AAPL.US", "700.HK", "600519.SH"],
    },
}

heartbeat = {"cmd_id": 22000, "seq_id": 1, "trace": "heartbeat", "data": {}}


@dataclass
class Quote:
    feed: str
    code: str
    price: float
    volume: float
    ts_ms: int


def normalize(feed, tick):
    """外部原始tick数据归一化为内部Quote模型，供业务、回测模块调用"""
    return Quote(
        feed=feed,
        code=tick["code"],
        price=float(tick["price"]),
        volume=float(tick["volume"]),
        ts_ms=int(tick["tick_time"]),
    )


async def run_feed(name, cfg, queue):
    token = "填写申请获取的AllTick_token"
    uri = f"{BASE}{cfg['path']}?token={token}"
    subscribe = {
        "cmd_id": 22004,
        "seq_id": 1,
        "trace": str(uuid.uuid4()),
        "data": {"symbol_list": [{"code": c} for c in cfg["codes"]]},
    }
    while True:  # 断线自动重连
        try:
            async with websockets.connect(uri) as ws:
                await ws.send(json.dumps(subscribe))

                async def beat():
                    while True:
                        await asyncio.sleep(10)
                        await ws.send(json.dumps(heartbeat))

                task = asyncio.create_task(beat())
                try:
                    async for raw in ws:
                        msg = json.loads(raw)
                        if msg.get("cmd_id") == 22998:
                            await queue.put(normalize(name, msg["data"]))
                finally:
                    task.cancel()
        except (websockets.ConnectionClosed, OSError):
            await asyncio.sleep(3)


async def consumer(queue):
    while True:
        q = await queue.get()
        # 可扩展：写入数据库、推送业务模块、回测数据前置处理
        print(q.feed, q.code, q.price, q.ts_ms)


async def main():
    queue = asyncio.Queue()
    await asyncio.gather(
        consumer(queue),
        *[run_feed(n, c, queue) for n, c in FEEDS.items()],
    )


asyncio.run(main())

开发注意事项

1. 各个市场使用独立WebSocket连接，分别完成订阅、心跳保活、断线重试；解析后数据投递至公共异步队列统一消费；
2. 同一条WebSocket会话中，重复发送订阅指令会覆盖原有订阅列表。动态新增标的，需要提交完整品种集合；
3. 接入新数据源务必打印原始报文校验时间戳单位，时间单位错误，会直接造成时序、回测结果错乱。

五、归一化之后需要关注的工程细节

完成内部模型转换，不等于数据链路全部就绪，以下几点会直接影响数据可靠性，需要在开发阶段处理：

1. 分市场设计故障告警逻辑
   不要使用全局统一超时阈值。加密货币监控超时阈值可以设置较短；外汇、股票需要结合交易日历做判断。外汇周末无行情属于正常休市，不应触发故障告警。

2. 规避浮点数计算精度问题
   外汇报价普遍5位小数，A股2位小数，加密货币最高8位小数。涉及金额、净值运算场景，建议使用Decimal类型，避免浮点数累积误差，防止业务计算、回测结果出现偏移。

3. 历史K线数据同样执行归一化
   实时行情通过WebSocket获取，历史K线大多调用REST接口，两套接口返回结构通常不一样。历史数据拉取完成后，同样转换为Quote内部模型再写入数据库。实现历史回测、实时业务复用同一套数据对象，缩小实盘与回测的逻辑差异。

六、总结

多资产行情整合没有捷径。在数据源选型阶段优先选择协议规范、文档完善的服务；架构层面，利用适配层隔离第三方接口的异构差异，保障上层业务、量化回测模块的稳定性。

开发多资产相关系统的时候，优先定义统一的内部行情模型，能够有效降低后续迭代、切换数据源带来的改造成本，提升整条数据链路的可维护性。原型开发阶段，也可参考 AllTick API 这类协议一致性表现较佳的行情服务，降低适配层的开发负担。

免责声明：本文为技术实践分享，示例代码仅适用于原型验证。部署生产系统，需要补齐异常捕获、数据校验、持久化存储、监控告警、权限安全等能力。