supermind logo
同花顺logo

全球股票行情数据接入实战：一套 API 搞定多市场量化数据

用户头像Fxdund
2026-08-24 发布

做量化这几年，我踩过最多的坑不是策略，而是数据。

最早接美股的时候用的某家老牌数据商，文档写得像天书，SDK 只支持 Python 2，光调通一个历史K线接口就花了我三天。后来业务扩展到港股和A股，又分别接了两个不同的数据源，三套接口、三种鉴权方式、两套字段命名，每天光维护数据同步脚本就要占掉半天。

上个月团队要做一个覆盖全球主要市场的选股策略，我实在不想再维护第四套数据接入了，花了一周时间调研选型，最后用一套统一的 REST + WebSocket API 搭好了跨市场数据管道。这篇文章记录一下整个接入过程和技术细节，给同样在折腾多市场数据的朋友一个参考。

选型思路

选数据源我主要看三点：覆盖市场够不够广、接口是不是统一、延迟稳不稳定。

我最终选的这套方案覆盖的市场比较全，美股、港股、A股、台湾、新加坡、日本、印度、泰国、德国、英国、荷兰这些主流市场都有，外加外汇、指数、期货、基金、加密货币，基本上一个 API 就能搞定大部分场景。

接口统一这点对我来说是最大的痛点解决。不管是股票还是外汇，REST API 的调用方式、参数结构、返回字段都是一套规范，不用每个市场都重新学一遍。WebSocket 也是统一的发布订阅模式，连上之后订阅什么品种就收什么数据。

延迟方面我实测了一周，美股盘口数据推送基本在毫秒级，至少我这一周没遇到过断连。

第一步：REST API 拉历史K线

先从最简单的开始。这套 API 用 header 传 token 鉴权，注册账号后在控制台生成一个 token 就能用。

拉港股腾讯控股（700）和阿里巴巴（9988）最近5根5分钟K线，直接一个 GET 请求：

import requests

url = "//api.itick.org/stock/klines?region=HK&codes=700,9988&kType=2&limit=5"
headers = {
    "accept": "application/json",
    "token": "你的token"
}

response = requests.get(url, headers=headers)
print(response.json())

参数说几个关键点：

  • region 是市场代码，HK是港股，US是美股，SH/SZ是沪深
  • codes 支持批量，逗号分隔
  • kType 是K线周期，1=1分钟，2=5分钟，3=15分钟，4=30分钟，5=1小时，8=日线，9=周线，10=月线

返回的数据结构是这样的：

{
  "code": 0,
  "msg": null,
  "data": {
    "700": [
      {
        "o": 535,
        "h": 536,
        "l": 534.5,
        "c": 534.5,
        "v": 104799385,
        "tu": 56119888070.5,
        "t": 1741239000000
      }
    ],
    "9988": [
      {
        "o": 139.9,
        "h": 140.3,
        "l": 139.8,
        "c": 140.1,
        "v": 538602171,
        "tu": 75404622753.1,
        "t": 1741239000000
      }
    ]
  }
}

标准的 OHLCV 结构，t 是毫秒级时间戳，tu 是成交额。拿到数据后直接丢进 pandas 就能算指标了，不用做任何字段映射。

我对比了一下，同样的日线数据，复权处理是对的，拆股和分红都自动调整过了，不用自己再算复权因子。

第二步：WebSocket 订阅实时行情

历史数据搞定了，接下来是实时行情。做策略的都知道，轮询 REST API 既浪费调用次数又有延迟，WebSocket 才是正道。

这套 API 的 WebSocket 是发布订阅模式，连上之后发一条订阅消息就行。我用 Python SDK 来演示，比自己手写 websocket 连接省事很多：

from itick.sdk import Client
import time

# 初始化客户端
token = "你的token"
client = Client(token)

# 设置消息回调
def on_message(message):
    print(f"收到行情: {message}")

def on_error(error):
    print(f"连接错误: {error}")

client.set_message_handler(on_message)
client.set_error_handler(on_error)

# 连接股票WebSocket
client.connect_stock_websocket()

# 订阅腾讯控股实时行情
client.send_websocket_message('{"action": "subscribe", "codes": ["700"]}')

# 保持连接
try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    client.close_websocket()

SDK 内置了自动重连和心跳维护，断网后会自动重连并恢复订阅，不用自己写重连逻辑。重连间隔5秒，最多重试10次，心跳30秒一次，这些参数对大多数场景都够用了。

实时推送的数据包含最新价、买卖盘口、成交量这些，做盘中策略完全够用。我同时订阅了20只股票，推送频率很稳，没有遇到丢数据的情况。

第三步：SDK 与直接调 API 的选择

这套方案提供了 Python SDK 和 Java SDK，我两个都试了一下。

如果你的项目是 Python，用 SDK 会方便一些，pip install itick-sdk 装好，封装了所有 REST 接口和 WebSocket 连接：

from itick.sdk import Client

client = Client("你的token")

# 股票实时报价
quote = client.get_stock_quote("US", "AAPL")
print(f"苹果最新价: {quote['ld']}")

# 历史K线
kline = client.get_stock_kline("US", "AAPL", kType=8, limit=30)

# 外汇实时tick
forex_tick = client.get_forex_tick("GB", "EURUSD")

# 加密货币深度
crypto_depth = client.get_crypto_depth("BA", "BTCUSDT")

方法命名比较直观，get_stock_quote、get_stock_kline、get_forex_tick，看名字就知道干嘛的。

如果你用的语言没有官方 SDK，直接调 REST API 也很简单，就是标准的 HTTP 请求加 header 鉴权，没有什么复杂的签名算法。

数据质量验证

做量化的对数据质量都很敏感，我重点验证了几个方面：

复权处理：美股拆股、港股供股、A股分红，这些场景下的前复权数据都是对的，我抽了几只历史上有拆股的股票对比过，没有出现价格断层。

时间戳对齐：不同市场的交易时间不一样，返回的时间戳都是 UTC 毫秒，自己转成本地时间就行，不会出现时区混乱的问题。

缺失数据：停牌期间的K线会跳过，不会用0或者上一个价格填充，这点比有些数据商处理得好，不用自己再过滤异常值。

盘口深度：能拿到5档盘口，普通趋势策略够用了。

接入踩坑记录

最后说几个我接入时遇到的小问题，帮大家省点时间：

  1. region 参数别写错：美股是 US 不是 USA，港股是 HK 不是 HKG，刚开始我写错了返回空数据，查了半天才发现。
  2. WebSocket 订阅 codes 是数组：别传字符串，要传 ["700", "9988"] 这种数组格式。
  3. 注意调用频率限制：写循环的时候记得加 sleep，不然会被限流。
  4. 历史数据深度：不同市场的历史数据深度不一样，美股能到十几年，一些小市场可能只有几年，回测前先确认数据量够不够。
  5. token 别硬编码：放环境变量里，代码上传到 Git 之前检查一下，别把 token 提交上去了。

总结

用了这套方案一个月，最大的感受就是省时间。以前接三个市场的数据要维护三套代码，现在一个 SDK 全搞定，省下来的时间能多研究几个策略。

如果你也在做多市场量化，或者正在为数据源的事情头疼，建议在选型时重点关注接口统一性和市场覆盖度，这两点直接决定了后期维护成本。

评论