适配聚宽社区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股为主,港股低频历史回测 |
💡选型小结
- 个人量化、中小型项目,追求低延迟同时控制成本:优先 DeepTick;
- 机构投研,需要完整基本面、研报数据,预算充足:选择 Wind;
- 主要做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())
四、生产环境落地优化建议(量化实盘注意点)
- Token安全:禁止硬编码密钥,使用环境变量/配置中心管理,定期轮换密钥;
- 交易时段适配:港股交易时段
09:30‑12:00、13:00‑16:00,非交易时段行情静止,可以减少请求、降低配额消耗; - 断线补偿逻辑:WebSocket重连成功后,调用K线接口拉取断线区间数据,填补推送缺口,避免回测/实盘策略缺失数据;
- 缓存策略:热门标的行情写入Redis,1秒内重复请求走缓存,减少API调用次数;
- 限流与重试:遇到429限流,使用指数退避,不要无限循环重试;
- 监控告警:监控接口错误码、响应延迟、WebSocket重连次数,401/403/429要接入告警;
- 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:这套代码是否可以直接用于实盘?
示例代码仅演示接口调用,实盘还需要补充异常处理、数据校验、断线补偿、监控告警,不能直接照搬上线。
免责声明:本文所有代码仅作为技术学习演示,不构成任何投资建议。量化策略实盘存在风险,行情接口商用需要确认服务商授权许可,据此操作风险自担。

