我奶奶都会用的港股API教程

最近接到几个开发者的咨询,大家反映在找港股实时行情接口的时候,普遍遇到一个问题:要么价格高到离谱,要么数据质量差得让人绝望。

有一个团队跟我说,他们在做一个面向海外华人的港股交易平台,早期为了省钱,用的是某个免费接口抓取港股数据,结果数据延迟严重、经常断流,用户反馈说行情总是”卡住不动”,甚至出现过腾讯(00700.HK)页面价格和实际价格差了两块钱的情况,用户一度以为平台在做假数据。

其实这个问题在港股赛道非常普遍,因为相比美股,港股的数据生态要封闭很多,免费又好用的接口几乎不存在。

这篇文章就来聊聊,开发者接入港股实时行情的几种方式,以及如何通过 Infoway API 用最低成本把实时港股行情跑起来。

一、香港股票市场简介

港股市场的核心是香港联合交易所(HKEX),在交易量和上市公司质量上都是亚洲最重要的市场之一。截至2026年,港股上市公司超过2,600家,总市值超过35万亿港元,汇聚了腾讯、阿里巴巴、美团、小米、汇丰银行等大量具有全球影响力的公司。

对于中国大陆的投资者来说,港股还有一个独特吸引力——港股通,允许大陆投资者直接用人民币账户买卖部分港股,这让两地联动越来越密切,也让针对港股的分析工具和交易平台需求快速增长。

港股的交易时间为香港时间(UTC+8):

  • 早盘竞价:09:00 – 09:30
  • 正常交易:09:30 – 12:00,13:00 – 16:00
  • 收盘竞价:16:00 – 16:10

每年还有香港本地公众假期停市,需要注意日历管理。

二、港股实时行情数据的获取成本

和其他市场一样,获取港股实时数据的方式分两类:直连香港联合交易所,或者通过第三方数据商。

直连香港交易所

HKEX官方提供数据授权服务,授权费用完全公开透明。以常见的实时L1行情(一档买卖价)为例:

费用项目 大约年费用(人民币)
港股L1实时行情授权 约 60,000 – 120,000 元
终端用户分发费(每月/每用户) 约 30 – 80 元
专线接入费 另计
增值税 额外收取

而且直连方案要求申请方具备完整的企业资质,通常需要在香港当地注册公司并签署复杂的数据协议。对于中小团队,光是合规流程就能拖上几个月。

免费接口

像 Yahoo Finance 这类免费接口确实能查到港股数据,但延迟通常在15分钟以上,且频繁被封IP,完全无法用于实时应用或交易场景。

第三方数据服务商

这是大多数开发者和中小团队的实际选择。数据服务商统一采购交易所授权,以更合理的价格向下游分发,开发者只需要接一个接口,无需处理复杂的授权和基础设施问题。

对于大多数开发者来说,Infoway API 是性价比最高的选择,能在不承担高额成本的前提下快速搭建实时港股行情应用。

三、Infoway API 港股行情覆盖范围

Infoway API 覆盖香港交易所全部挂牌产品,包括:

  • 港股主板股票(超过2,000只)
  • 港股创业板
  • 港股轮证(Warrant)
  • 恒生指数、恒生科技指数等主要指数
  • 港股暗盘(IPO前竞价)

常用的港股代码格式为 {代码}.HK,例如:

Symbol 公司名称
00700.HK 腾讯控股
09988.HK 阿里巴巴
03690.HK 美团
01810.HK 小米集团
00005.HK 汇丰控股
02318.HK 中国平安
00941.HK 中国移动

除了港股以外,Infoway API 还提供以下市场的实时数据:

  • A股、美股、日本股票、印度股票
  • 外汇货币对
  • 加密货币
  • 商品期货、CFD、贵金属

统一的接口结构,让你今天接入港股,之后扩展到其他市场时几乎不需要重构系统。

四、接入准备

在官网注册账号,注册完成后自动获得7天免费试用,可以查询所有市场的数据,无需绑定信用卡,无需实名认证。注册后在后台可以看到你的 API Key,所有请求都需要在请求头中带上这个 Key:

apiKey: YOUR_API_KEY_HERE

五、港股行情 API 使用教程

5.1 获取实时成交明细

成交明细接口返回最近一笔成交的价格、成交量等信息,是最基础的行情数据之一。

接口地址:https://data.infoway.io/stock/batch_trade/{codes}

以下示例查询腾讯(00700.HK)的最新成交:

import requests

api_url = 'https://data.infoway.io/stock/batch_trade/00700.HK'

headers = {
    'User-Agent': 'Mozilla/5.0',
    'Accept': 'application/json',
    'apiKey': 'YOUR_API_KEY_HERE'
}

response = requests.get(api_url, headers=headers)

print(f"HTTP code: {response.status_code}")
print(f"message: {response.text}")

返回示例:

{
  "ret": 200,
  "msg": "success",
  "traceId": "a3f2d1c0-1234-5678-abcd-ef0123456789",
  "data": [
    {
      "s": "00700.HK",
      "t": 1781672609632,
      "p": "468.200",
      "v": "12800",
      "vw": "5992960.00",
      "td": 1
    }
  ]
}

字段说明:

字段 说明
s 股票代码
t 成交时间戳(毫秒)
p 成交价格
v 成交量(股)
vw 成交额
td 交易方向:0=默认,1=Buy,2=Sell

如果需要同时查询多只港股,在 codes 里用逗号分隔即可,最多支持100个产品一次查询:

api_url = 'https://data.infoway.io/stock/batch_trade/00700.HK,09988.HK,03690.HK'

5.2 获取实时K线数据

K线接口支持1分钟到年线共12种周期,是构建图表最核心的数据来源。

接口地址:https://data.infoway.io/stock/v2/batch_kline(POST请求)

以下示例查询腾讯(00700.HK)的1分钟K线,返回最近10根:

import requests
import json

api_url = 'https://data.infoway.io/stock/v2/batch_kline'

headers = {
    'User-Agent': 'Mozilla/5.0',
    'Accept': 'application/json',
    'Content-Type': 'application/json',
    'apiKey': 'YOUR_API_KEY_HERE'
}

payload = {
    "klineType": 1,   # 1=1分钟K;8=日K;9=周K;10=月K
    "klineNum": 10,
    "codes": "00700.HK"
}

response = requests.post(api_url, headers=headers, data=json.dumps(payload))

print(f"HTTP code: {response.status_code}")
print(f"message: {response.text}")

返回示例:

{
  "ret": 200,
  "msg": "success",
  "traceId": "b5c3e2d1-abcd-5678-ef01-234567890abc",
  "data": [
    {
      "s": "00700.HK",
      "respList": [
        {
          "t": "1781671920",
          "h": "469.000",
          "o": "467.800",
          "l": "467.600",
          "c": "468.400",
          "v": "48600",
          "vw": "22762680.00",
          "pc": "0.17%",
          "pca": "0.800"
        }
      ]
    }
  ]
}

K线周期参数(klineType)说明:

周期
1 1分钟
2 5分钟
3 15分钟
4 30分钟
5 1小时
8 日K
9 周K
10 月K

如果需要查询历史数据,在请求体中加入 timestamp 参数(秒时间戳),接口会向前返回该时间点之前的K线:

payload = {
    "klineType": 1,
    "klineNum": 500,
    "codes": "00700.HK",
    "timestamp": 1780000000  # 向前查询该时间点的历史K线
}

分钟级历史数据支持最近3年,日K及以上不受时间限制。

5.3 获取实时盘口深度

盘口数据反映当前市场的买卖挂单分布,对于交易所和量化策略来说非常重要。Infoway API 提供港股十档买卖盘口。

接口地址:https://data.infoway.io/stock/batch_depth/{codes}

import requests

api_url = 'https://data.infoway.io/stock/batch_depth/00700.HK'

headers = {
    'User-Agent': 'Mozilla/5.0',
    'Accept': 'application/json',
    'apiKey': 'YOUR_API_KEY_HERE'
}

response = requests.get(api_url, headers=headers)

print(f"HTTP code: {response.status_code}")
print(f"message: {response.text}")

返回示例:

{
  "ret": 200,
  "msg": "success",
  "traceId": "c7d4f3e2-bcde-6789-f012-345678901bcd",
  "data": [
    {
      "s": "00700.HK",
      "t": 1781672693007,
      "a": [
        ["468.600", "468.800", "469.000", "469.200", "469.400",
         "469.600", "469.800", "470.000", "470.200", "470.400"],
        ["12800", "8400", "23600", "15200", "31000",
         "9800", "17400", "42000", "8200", "26600"]
      ],
      "b": [
        ["468.400", "468.200", "468.000", "467.800", "467.600",
         "467.400", "467.200", "467.000", "466.800", "466.600"],
        ["18000", "32400", "51200", "27800", "14600",
         "38000", "9400", "21600", "16200", "8800"]
      ]
    }
  ]
}

其中 a 为卖盘(Ask),b 为买盘(Bid)。第一个子数组是价格列表,第二个子数组是对应的挂单量,通过下标一一对应。

5.4 WebSocket 订阅港股实时推送

如果你的应用需要毫秒级的实时行情更新(例如交易所行情面板、量化策略触发器),WebSocket 是更合适的方案。相比 REST 轮询,WebSocket 建立连接后服务端主动推送,无需频繁发起请求,延迟更低,服务器压力也更小。

WebSocket 连接地址:wss://data.infoway.io/ws?business=stock&apikey=YOUR_API_KEY_HERE

下面是一个完整的 Python 示例,包含订阅成交明细、盘口、K线,以及断线重连机制:

import asyncio
import json
import uuid
import logging
from typing import Optional
import websockets
from websockets.exceptions import ConnectionClosed, WebSocketException

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[logging.StreamHandler()]
)
logger = logging.getLogger("hk-ws-client")

class HKStockWebsocketClient:
    """港股行情 WebSocket 客户端"""

    def __init__(self, api_key: str):
        self.ws_url = f"wss://data.infoway.io/ws?business=stock&apikey={api_key}"
        self.ws: Optional[websockets.WebSocketClientProtocol] = None
        self.is_connected = False
        self.reconnect_interval = 10
        self.heartbeat_interval = 30
        self.heartbeat_task: Optional[asyncio.Task] = None
        self.reconnect_task: Optional[asyncio.Task] = None

    def _trace_id(self) -> str:
        return str(uuid.uuid4())

    async def connect(self) -> None:
        if self.ws and not self.ws.closed:
            await self.ws.close()
        self.ws = await websockets.connect(self.ws_url)
        self.is_connected = True
        logger.info(f"WebSocket 连接成功")
        await self._subscribe_all()
        self._start_heartbeat()

    async def _subscribe_all(self) -> None:
        # 订阅实时成交明细
        await self.ws.send(json.dumps({
            "code": 10000,
            "trace": self._trace_id(),
            "data": {"codes": "00700.HK,09988.HK,03690.HK"}
        }))
        await asyncio.sleep(5)

        # 订阅实时盘口
        await self.ws.send(json.dumps({
            "code": 10003,
            "trace": self._trace_id(),
            "data": {"codes": "00700.HK,09988.HK,03690.HK"}
        }))
        await asyncio.sleep(5)

        # 订阅1分钟K线
        await self.ws.send(json.dumps({
            "code": 10006,
            "trace": self._trace_id(),
            "data": {
                "arr": [
                    {"type": 1, "codes": "00700.HK,09988.HK,03690.HK"}
                ]
            }
        }))

    def _start_heartbeat(self) -> None:
        if self.heartbeat_task and not self.heartbeat_task.done():
            self.heartbeat_task.cancel()

        async def heartbeat_loop():
            while self.is_connected and self.ws and not self.ws.closed:
                try:
                    await self.ws.send(json.dumps({
                        "code": 10010,
                        "trace": self._trace_id()
                    }))
                    await asyncio.sleep(self.heartbeat_interval)
                except Exception:
                    break

        self.heartbeat_task = asyncio.create_task(heartbeat_loop())

    async def _listen(self) -> None:
        while self.is_connected and self.ws and not self.ws.closed:
            try:
                message = await self.ws.recv()
                self._handle(message)
            except ConnectionClosed:
                self.is_connected = False
                break
            except Exception as e:
                logger.error(f"消息处理异常: {e}")

    def _handle(self, message: str) -> None:
        try:
            msg = json.loads(message)
            code = msg.get("code")
            data = msg.get("data", {})
            if code == 10000:
                logger.info(f"成交明细: {data}")
            elif code == 10003:
                logger.info(f"盘口数据: {data}")
            elif code == 10006:
                logger.info(f"K线数据: {data}")
            elif code == 10010:
                logger.debug("心跳响应")
        except Exception as e:
            logger.error(f"解析消息失败: {e}")

    async def _reconnect_loop(self) -> None:
        while True:
            if not self.is_connected:
                logger.info(f"尝试重连(间隔 {self.reconnect_interval}s)...")
                try:
                    await self.connect()
                except Exception:
                    await asyncio.sleep(self.reconnect_interval)
            else:
                await asyncio.sleep(1)

    async def start(self) -> None:
        self.reconnect_task = asyncio.create_task(self._reconnect_loop())
        try:
            await self.connect()
            await self._listen()
        finally:
            self.is_connected = False
            if self.heartbeat_task:
                self.heartbeat_task.cancel()
            if self.reconnect_task:
                self.reconnect_task.cancel()
            if self.ws and not self.ws.closed:
                await self.ws.close()
            logger.info("客户端已停止")


async def main():
    client = HKStockWebsocketClient(api_key="YOUR_API_KEY_HERE")
    await client.start()

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        logger.info("用户中断")

WebSocket 订阅协议号说明:

协议号 说明
10000 订阅实时成交明细
10003 订阅实时盘口
10006 订阅实时K线
10010 心跳

以上示例同时订阅了腾讯、阿里、美团三只港股,实际使用中可以根据需要调整订阅列表,支持同时订阅多个产品。

六、常见问题

港股的交易时间是什么?

正常交易日(周一至周五,非香港公众假期):早盘竞价 09:00–09:30,正式交易 09:30–12:00 及 13:00–16:00,收盘竞价 16:00–16:10。时区为 UTC+8。

数据延迟有多少?

通过 WebSocket 订阅,数据延迟通常在毫秒级。REST API 适合查询历史数据和低频刷新场景,延迟会相对高一些。

支持港股通标的查询吗?

支持,Infoway API 覆盖港交所全部挂牌产品,港股通标的包含在内,不需要做特殊区分。

可以同时订阅多只港股吗?

可以。REST API 单次最多支持100个产品同时查询,WebSocket 订阅的产品数量取决于你的套餐等级:

  • 免费试用:最多10个产品
  • 基础套餐(99 USDT/月):最多200个产品
  • 高级套餐(199 USDT/月):最多800个产品
  • 专业套餐(399 USDT/月):最多5,000个产品

如果你只需要港股数据,可以咨询我们的港股专项套餐,可订阅全部港股实时行情。

REST API 和 WebSocket 应该怎么选?

两者可以配合使用:页面首次加载时用 REST API 获取历史 K 线,渲染好初始图表;之后切换 WebSocket 持续接收实时更新。这样既避免了高频轮询带来的请求压力,也保证了用户体验的实时性。

支持免费试用吗?

支持。在我们官网注册账号后自动获得7天免费试用,可查询所有市场数据,无需申请,无需绑定信用卡,无需实名认证。有任何技术问题欢迎联系我们的客服 Telegram。

本作品采用《CC 协议》,转载必须注明作者和本文链接
讨论数量: 0
(= ̄ω ̄=)··· 暂无内容!

讨论应以学习和精进为目的。请勿发布不友善或者负能量的内容,与人为善,比聪明更重要!