港股实时行情 API 接口选型与接入实战

用户头像sh_***312kg3
2026-11-06 发布

适配聚宽社区Markdown编辑器,适合量化开发者阅读;包含服务商选型、Python完整示例、生产环境优化、踩坑排查;文末附带聚宽社区标准免责声明,直接复制粘贴即可发布。

随着内地与港股互联互通不断深化,做港股量化回测、行情分析工具、多市场策略开发,都需要稳定的港股实时行情数据源。 很多开发者接入过程经常遇到:接口延迟高、数据字段缺失、WebSocket频繁断线、token鉴权401、限流429等问题。

本文完整覆盖:服务商选型对比 → 环境与鉴权封装 → REST接口实战(实时报价/盘口深度/K线历史) → WebSocket长连接订阅 → 生产环境优化 → 报错排查 + FAQ,全部提供可运行Python代码,帮助量化开发者快速完成港股行情接口落地。

一、港股行情API服务商选型对比

从实时性、接入成本、易用性、数据完整性4个维度对比 DeepTick、Wind、Tushare三款主流数据源,方便量化开发者根据项目规模做取舍。

API 服务商 实时性 接入成本 易用性 核心优势 适用场景
DeepTick 毫秒级推送,延迟 ≤50ms 免费版支持基础行情,付费版性价比高 RESTful规范,请求头统一,返回格式简洁 港股全量覆盖,支持WebSocket实时订阅 个人量化开发者、中小型行情系统、多市场策略
Wind 金融终端API 延迟100‑300ms 年费数万元起,无免费版 需要客户端,参数复杂 数据维度丰富,附带研报、基本面 机构、专业投研团队
Tushare 港股接口 延迟1‑3秒,非实时 积分模式,高频调用付费 Python SDK友好,港股覆盖有限 A股数据完善,文档成熟 A股为主,港股低频历史回测

💡选型小结

  1. 个人量化、中小型项目,追求低延迟同时控制成本:优先 DeepTick;
  2. 机构投研,需要完整基本面、研报数据,预算充足:选择 Wind;
  3. 主要做A股,仅少量港股历史回测:可以选择 Tushare。

二、接入前置准备:环境依赖与统一请求封装

安装依赖库

pip install requests websockets pandas matplotlib

基础客户端封装(DeepTick)

token从环境变量读取,禁止代码硬编码密钥,避免密钥泄露风险。鉴权头字段为 deeptoken。

import os
import requests

API_BASE = "[https://api.deeptick.cc](https://api.deeptick.cc)"
API_TOKEN = os.getenv("DEEPTICK_API_TOKEN", "your_api_key")

headers = {
    "accept": "application/json",
    "deeptoken": API_TOKEN,
    "content-type": "application/json; charset=utf-8"
}


class DeepTickClient:
    def __init__(self, base_url=API_BASE, custom_headers=None):
        self.base_url = base_url
        self.headers = headers.copy()
        if custom_headers:
            self.headers.update(custom_headers)

    def _get(self, path, params=None):
        """统一封装GET请求,捕获网络异常、超时"""
        try:
            resp = requests.get(
                f"{self.base_url}{path}",
                headers=self.headers,
                params=params,
                timeout=10
            )
            resp.raise_for_status()
            return resp.json()
        except requests.exceptions.RequestException as e:
            print(f"请求异常: {e}")
            return None

三、REST接口实战示例

3.1 获取港股实时报价

接口文档:https://deeptick.cc/docs/rest-quote

获取标的最新价、开盘、最高最低、成交量、成交额、涨跌幅、交易状态;一次仅支持单个标的查询。

import datetime

class DeepTickClient(DeepTickClient):
    def get_quote(self, symbol="00700.HK"):
        """
        GET /api/v1/quote
        :param symbol: 标的,格式 代码.市场,例 00700.HK(腾讯控股)
        :return: 报价字典
        """
        resp = self._get("/api/v1/quote", {"symbols": symbol})
        if resp and resp.get("code") == 0 and resp.get("data"):
            return resp["data"][0]
        print(f"获取报价失败: {resp.get('msg') if resp else '未知错误'}")
        return None


# 使用示例
client = DeepTickClient()
quote = client.get_quote("00700.HK")
if quote:
    print(f"标的代码: {quote['s']}")
    print(f"最新价: {quote['ld']} 前收: {quote.get('p')}")
    print(f"开/高/低: {quote.get('o')} / {quote.get('h')} / {quote.get('l')}")
    print(f"成交量: {quote.get('v')} 成交额: {quote.get('tu')}")
    print(f"涨跌额: {quote.get('ch')} 涨跌幅: {quote.get('chp')}")

    # ts 交易状态:0正常,1停牌,2退市,3熔断,不是时间戳
    if quote.get("ts") is not None:
        status_map = ["正常", "停牌", "退市", "熔断"]
        print(f"交易状态: {status_map[quote['ts']]}")

    # t 毫秒时间戳
    if quote.get("t"):
        trade_dt = datetime.datetime.fromtimestamp(quote["t"] / 1000)
        print(f"行情时间: {trade_dt}")

3.2 获取实时盘口深度数据

接口文档:https://deeptick.cc/docs/rest-orderbook

获取买卖档位挂单价格与挂单量;买盘由高到低,卖盘由低到高,适合做盘口因子、短线策略分析。

def get_orderbook(self, symbol="00700.HK"):
    """
    GET /api/v1/orderbook
    :param symbol: 标的代码 00700.HK
    :return: dict{s,b,a,t}
    """
    resp = self._get("/api/v1/orderbook", {"symbol": symbol})
    if resp and resp.get("code") == 0 and resp.get("data") is not None:
        return resp["data"]
    print(f"获取盘口失败: {resp.get('msg') if resp else '未知错误'}")
    return None


# 使用示例
book = client.get_orderbook("00700.HK")
if book:
    print(f"标的代码: {book['s']}")
    print("=== 卖盘(价格由低到高)===")
    for ask in book["a"]:
        print(f"价格: {ask['p']} 挂单量: {ask['v']}")

    print("=== 买盘(价格由高到低)===")
    for bid in book["b"]:
        print(f"价格: {bid['p']} 挂单量: {bid['v']}")

    ask_volume = sum(float(ask["v"]) for ask in book["a"])
    bid_volume = sum(float(bid["v"]) for bid in book["b"])
    if ask_volume > 0:
        print(f"买卖盘挂单量比(买/卖): {bid_volume / ask_volume:.2%}")

3.3 获取历史K线数据(策略回测)

接口文档:https://deeptick.cc/docs/rest-candles

获取历史K线,支持多周期,返回数据转换为Pandas DataFrame,可直接计算技术指标、绘图回测。

import pandas as pd
import matplotlib.pyplot as plt

def get_candles(self, symbol="00700.HK", interval="5m", limit=50, et=None):
    """
    GET /api/v1/candles
    :param interval: 1m /5m /15m /30m /60m /4h /1d /1w /1M
    :param limit: 1~500
    :param et: 截止时间戳(毫秒)
    :return: k线列表
    """
    params = {"symbols": symbol, "interval": interval, "limit": limit}
    if et:
        params["et"] = et
    resp = self._get("/api/v1/candles", params)
    if resp and resp.get("code") == 0 and resp.get("data"):
        return resp["data"][0].get("d", [])
    print(f"获取K线失败: {resp.get('msg') if resp else '未知错误'}")
    return None


# 使用示例
candles = client.get_candles("00700.HK", interval="5m", limit=50)
if candles:
    df = pd.DataFrame(candles)
    df["t"] = pd.to_datetime(df["t"], unit="ms")
    df.set_index("t", inplace=True)
    for col in ["o", "h", "l", "c", "v"]:
        df[col] = pd.to_numeric(df[col])

    print(df.head())
    # 计算简单移动平均
    df["sma_5"] = df["c"].rolling(window=5).mean()
    df["sma_20"] = df["c"].rolling(window=20).mean()

    plt.figure(figsize=(12, 6))
    plt.plot(df.index, df["c"], label="收盘价")
    plt.plot(df.index, df["sma_5"], label="SMA5")
    plt.plot(df.index, df["sma_20"], label="SMA20")
    plt.title("腾讯控股 00700.HK 5分钟K线与均线")
    plt.legend()
    plt.show()

3.4 WebSocket实时行情推送

接口文档:https://deeptick.cc/docs/ws-subscribe

高频行情场景,优先WebSocket长连接替代轮询,减少HTTP请求,降低限流风险。 ⚠️注意:quote和depth属于不同type,不能放在同一条subscribe消息,需要分开发送,否则返回4001错误码。

import asyncio
import json
import os
import websockets

WS_URL = "wss://api.deeptick.cc/ws/v1"

class DeepTickWSClient:
    def __init__(self, token=None, url=WS_URL):
        self.url = url
        self.headers = {"deeptoken": token or os.getenv("DEEPTICK_API_TOKEN", "your_api_key")}
        self.ws = None

    async def subscribe(self):
        # quote、depth分两次发送订阅
        topics = ["quote.HK.00700", "depth.HK.00700"]
        for topic in topics:
            await self.ws.send(json.dumps({
                "type": "subscribe",
                "topics": [topic],
                "request_id": f"sub-{topic}"
            }))

    async def handle(self, msg):
        if msg.get("type") == "ack":
            print(f"订阅回执|成功:{len(msg.get('accepted',[]))},拒绝:{msg.get('rejected')}")
            return
        print(f"【行情推送】topic:{msg.get('topic')}, data:{json.dumps(msg.get('data'), ensure_ascii=False)}")

    async def run(self):
        while True:
            try:
                async with websockets.connect(self.url, additional_headers=self.headers) as ws:
                    self.ws = ws
                    print("WebSocket连接建立成功")
                    welcome = await ws.recv()
                    print(f"welcome帧:{welcome[:80]}")
                    await self.subscribe()
                    async for raw in ws:
                        await self.handle(json.loads(raw))
            except Exception as e:
                print(f"连接异常,5s后自动重连: {e}")
                await asyncio.sleep(5)


if __name__ == "__main__":
    asyncio.run(DeepTickWSClient().run())

四、生产环境落地优化建议(量化实盘注意点)

  1. Token安全:禁止硬编码密钥,使用环境变量/配置中心管理,定期轮换密钥;
  2. 交易时段适配:港股交易时段 09:30‑12:00、13:00‑16:00,非交易时段行情静止,可以减少请求、降低配额消耗;
  3. 断线补偿逻辑:WebSocket重连成功后,调用K线接口拉取断线区间数据,填补推送缺口,避免回测/实盘策略缺失数据;
  4. 缓存策略:热门标的行情写入Redis,1秒内重复请求走缓存,减少API调用次数;
  5. 限流与重试:遇到429限流,使用指数退避,不要无限循环重试;
  6. 监控告警:监控接口错误码、响应延迟、WebSocket重连次数,401/403/429要接入告警;
  7. WebSocket心跳:长连接建议30‑60s发送心跳包,防止NAT超时断开。

五、常见报错排查

401 / 403 鉴权失败

  • deeptoken 过期,重新申请密钥;
  • 请求头字段名称大小写必须完全匹配;
  • 账号未开通接口权限,核对服务商后台。

429 Too Many Requests 调用超限

  • 请求频率超过套餐QPS;
  • 优化:循环轮询改为WebSocket订阅;使用批量接口;
  • 必要升级套餐。

WebSocket频繁断开

  • 网络NAT超时,增加心跳保活;
  • 代码必须实现自动重连,重连后重新下发订阅topic;
  • 注意:不同type不能合并到一条subscribe消息,否则返回4001拒绝订阅。

返回空数据、字段缺失

  • 标的格式区分:REST查询用00700.HK,WebSocket topic格式quote.HK.00700;
  • 标的停牌、退市会返回空;核对文档字段,不要直接复用其他接口字段映射。

六、常见问题 FAQ

Q1:港股实时行情API有免费额度吗?

DeepTick提供免费基础行情额度,高频盘口、长历史回溯需要付费;建议先用免费额度做功能验证,再升级套餐。

Q2:轮询和WebSocket该如何选择?

低频快照、少量查询可以轮询;实盘、短线策略、需要毫秒级行情,必须使用WebSocket长连接。

Q3:港股代码格式为什么有两种写法?

REST接口:00700.HK(代码.市场);WebSocket Topic:quote.HK.00700(类型.市场.代码),格式写错直接无数据。

Q4:这套代码是否可以直接用于实盘?

示例代码仅演示接口调用,实盘还需要补充异常处理、数据校验、断线补偿、监控告警,不能直接照搬上线。


免责声明:本文所有代码仅作为技术学习演示,不构成任何投资建议。量化策略实盘存在风险,行情接口商用需要确认服务商授权许可,据此操作风险自担。

评论