Python港股实时行情API接入：恒生股票数据获取与AH股溢价监控实战

做跨境量化或者港股通策略的开发者，接港股数据是绕不开的。港股跟A股有个很特别的联动关系——同一家公司在A股和港股两地上市，价格经常不一样，这就是所谓的AH股溢价。这个价差不仅是金融从业者关心的套利机会，也是技术开发者做跨市场数据监控的典型场景。

这篇从AH股溢价监控这个实际场景出发，讲一下怎么用Python接港股实时行情数据，以及怎么同时拉A股和港股的价格做价差计算。

为什么要接港股数据

一开始我做A股量化的时候，数据都是国内的接口，用得挺顺手。后来想做AH股套利策略，才发现需要同时拿A股和港股两边的实时价格。这时候问题就来了——用两个不同的接口拿两边的数据，时间戳对不上，延迟也不一样，算出来的价差不准。

后来找了一圈，试了几个不同的金融数据接口，最后留下来一直在用的是那种同时覆盖A股和港股的统一数据源——不用维护两套代码，两边的数据时间戳一致性也靠谱很多。现在做跨市场监控，一个key打通多个市场的接口选择其实挺关键的。

港股市场的基本情况

先理一下港股的基本情况，这样看返回数据的时候才知道每个字段是什么意思。

交易时间：上午900，下午1300。比A股多半小时，中午也有1小时休市。

代码格式：港股代码是5位数字，比如00700是腾讯、09988是阿里、03690是美团。跟A股的6位数字不一样，传参数的时候要注意位数。

交易货币：港币（HKD）。算AH股溢价的时候要注意汇率换算。

region代码：港股市场region传 HK，A股传 CN。

REST拉取港股报价

先从REST开始，确认数据格式。

import requests

API_BASE = "https://api.itick.org"

TOKEN = "your_token_here"

headers = {"accept": "application/json", "token": TOKEN}

# 拉港股实时报价

def  get_hk_stock_quote(code):

resp = requests.get(

f"{API_BASE}/stock/quote",

headers=headers,

params={

"region": "HK",

"code": code

}

)

return resp.json()["data"]

# 拉几只港股大盘股

stocks = ["00700", "09988", "03690"]

for code in stocks:

q = get_hk_stock_quote(code)

sign = "+"  if q["chp"] >= 0  else  ""

print(f"{code}: {q['ld']} ({sign}{q['chp']}%)")

返回字段跟其他市场基本一致：ld最新价、chp涨跌幅、o/h/l开高低、v成交量。做行情看板这些字段够用了。

注意：港股代码是5位数字，传参数的时候要传字符串形式（”00700”），不要传数字。一开始我传了”700”，结果返回空数据——港股代码必须带前导零，5位数字。

WebSocket订阅港股

做实时监控的时候，轮询REST接口效率太低，这时候就得用WebSocket实时推送了。

import json

import threading

import time

import websocket

WS_URL = "wss://api.itick.org/stock"

TOKEN = "your_token_here"

authenticated = False

def  on_message(ws, message):

global authenticated

payload = json.loads(message)

# 先等鉴权成功

if payload.get("resAc") == "auth"  and payload.get("code") == 1  and  not authenticated:

authenticated = True

# 港股订阅格式：代码$市场

ws.send(json.dumps({

"ac": "subscribe",

"params": "00700$HK,09988$HK",

"types": "quote"

}))

return

if payload.get("resAc") == "pong":

return

data = payload.get("data") or {}

if data.get("type") == "quote":

sign = "+"  if data["chp"] >= 0  else  ""

print(f"{data['s']}: {data['ld']} ({sign}{data['chp']}%)")

港股订阅是两个参数（00700$HK），跟美股、A股一样，不是印度那种三个参数。

AH股溢价监控实战

这是港股数据最有技术含量的应用场景——同一家公司在A股和港股两地上市，价格经常不一样。

比如招商银行，A股代码是600036，港股代码是03968。两边的价格理论上应该差不多，但实际上经常有5%-30%的溢价差。这个价差就是量化交易者关心的套利机会。

# AH股溢价监控

# 招商银行：A股 600036，港股 03968

# 中国平安：A股 601318，港股 02318

def  get_ah_premium(a_code, h_code, hkd_cny_rate=0.92):

"""

计算AH股溢价率

a_code: A股代码

h_code: 港股代码

hkd_cny_rate: 港币兑人民币汇率

"""

# 拉A股价格

a_resp = requests.get(

f"{API_BASE}/stock/quote",

headers=headers,

params={"region": "CN", "code": a_code}

).json()["data"]

# 拉港股价格

h_resp = requests.get(

f"{API_BASE}/stock/quote",

headers=headers,

params={"region": "HK", "code": h_code}

).json()["data"]

# 港股价格换算成人民币

h_price_cny = h_resp["ld"] * hkd_cny_rate

# 计算溢价率

premium = (a_resp["ld"] - h_price_cny) / h_price_cny * 100

return {

"a_code": a_code,

"a_price": a_resp["ld"],

"h_code": h_code,

"h_price_hkd": h_resp["ld"],

"h_price_cny": round(h_price_cny, 2),

"premium_pct": round(premium, 2)

}

# 监控几只AH股

ah_pairs = [

("600036", "03968"), # 招商银行

("601318", "02318"), # 中国平安

("600900", "00900"), # 长江电力

]

for a_code, h_code in ah_pairs:

result = get_ah_premium(a_code, h_code)

print(f"{result['a_code']} / {result['h_code']}: "

f"A股{result['a_price']}元, 港股{result['h_price_hkd']}港币 "

f"(≈{result['h_price_cny']}元), "

f"溢价{result['premium_pct']}%")

这个脚本跑起来之后，就能看到每只AH股的溢价率了。溢价率过高或者过低的时候，就是套利的机会。

注意：汇率是动态变化的，做实时监控的时候要定期更新汇率，不能写死。

K线数据获取

K线接口跟其他市场一样，只是region传 HK。

def  get_hk_kline(code, k_type=8, limit=100):

"""

k_type: 1=1分钟 2=5分钟 5=1小时 8=日K 9=周K 10=月K

"""

resp = requests.get(

f"{API_BASE}/stock/kline",

headers=headers,

params={

"region": "HK",

"code": code,

"kType": k_type,

"limit": limit

}

)

return resp.json()["data"]

# 拉腾讯近100根日K

daily = get_hk_kline("00700", k_type=8, limit=100)

closes = [bar["c"] for bar in daily]

# 算20日均线

ma20 = sum(closes[-20:]) / 20

print(f"00700最新收盘: {closes[-1]}, 20日均线: {ma20:.2f}")

K线返回标准OHLCV：t时间戳、o/h/l/c开高低收、v成交量。做技术分析直接用这些字段。

港股特有的注意事项

代码是5位数字，带前导零。 港股代码是5位数字（00700、09988），跟A股的6位数字（600519）不一样。一开始我传了”700”，结果返回空数据——港股代码必须是5位，前面补零。

交易时间比A股长半小时。 港股上午900，下午1300。比A股多半小时，做跨市场监控的时候要注意两边的交易时间窗口。

交易货币是港币。 港股价格单位是港币（HKD），做AH股溢价计算的时候要换算成人民币。汇率是动态的，不能写死。

港股通标的有限。 不是所有港股都能通过港股通买，做策略的时候要注意标的范围。

时间戳是毫秒。 所有返回的时间戳都是毫秒，直接当秒数处理会得到一个很远的未来时间，记得除以1000。

完整脚本：港股+AH股溢价监控

把前面几块拼起来，最终的完整脚本：

import json, time, threading, requests, websocket

API_BASE = "https://api.itick.org"

WS_URL = "wss://api.itick.org/stock"

TOKEN = "your_token_here"

headers = {"accept": "application/json", "token": TOKEN}

# 港股监控列表

watchlist = ["00700", "09988", "03690"]

# 启动时拉快照

print("=== 港股快照 ===")

for code in watchlist:

r = requests.get(f"{API_BASE}/stock/quote",

headers=headers,

params={"region": "HK", "code": code}).json()["data"]

sign = "+"  if r["chp"] >= 0  else  ""

print(f" {code}: {r['ld']} ({sign}{r['chp']}%)")

# AH股溢价快照

print("\n=== AH股溢价 ===")

ah_pairs = [("600036", "03968"), ("601318", "02318")]

for a_code, h_code in ah_pairs:

a = requests.get(f"{API_BASE}/stock/quote",

headers=headers,

params={"region": "CN", "code": a_code}).json()["data"]

h = requests.get(f"{API_BASE}/stock/quote",

headers=headers,

params={"region": "HK", "code": h_code}).json()["data"]

h_cny = h["ld"] * 0.92

premium = (a["ld"] - h_cny) / h_cny * 100

print(f" {a_code}/{h_code}: A股{a['ld']}元, 港股{h['ld']}港币 "

f"(≈{h_cny:.2f}元), 溢价{premium:.1f}%")

# WebSocket实时推送港股

authenticated = False

def  on_message(ws, message):

global authenticated

payload = json.loads(message)

if payload.get("resAc") == "auth"  and payload.get("code") == 1  and  not authenticated:

authenticated = True

params = ",".join(f"{c}$HK"  for c in watchlist)

ws.send(json.dumps({

"ac": "subscribe",

"params": params,

"types": "quote"

}))

threading.Thread(target=heartbeat, args=(ws,), daemon=True).start()

return

if payload.get("resAc") == "pong":

return

data = payload.get("data") or {}

if data.get("type") == "quote":

sign = "+"  if data["chp"] >= 0  else  ""

print(f"[实时] {data['s']}: {data['ld']} ({sign}{data['chp']}%)")

def  heartbeat(ws):

while  True:

time.sleep(30)

ws.send(json.dumps({"ac": "ping", "params": str(int(time.time() * 1000))}))

ws = websocket.WebSocketApp(WS_URL, header=[f"token: {TOKEN}"], on_message=on_message)

ws.run_forever()

这个脚本跑起来之后，开盘前先看到一组港股的快照和AH股溢价率，盘中实时更新港股价格。

小结

接港股数据的技术框架跟A股、美股差不多——REST做快照和历史数据，WebSocket做实时推送。主要区别在代码格式（5位数字带前导零）、交易货币（港币）和交易时间（比A股长半小时）。最有技术含量的应用场景是AH股溢价监控——同时拉A股和港股两边的价格，算跨市场价差。

做港股量化策略或者AH股套利的话，这套流程可以直接套。具体的参数细节和返回字段说明，我一般都是边写代码边翻官方文档对照，比自己猜参数靠谱多了。

GitHub：https://github.com/itick-org/

本作品采用《CC 协议》，转载必须注明作者和本文链接