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

用户头像Fxdund
2026-10-13 发布

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

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

为什么要接港股数据

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

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

港股市场的基本情况

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

交易时间:上午9:30-12:00,下午13:00-16:00。比A股多半小时,中午也有1小时休市。

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

交易货币:港币(HKD)。算AH股溢价的时候要注意汇率换算。

region代码:港股市场region传 HK,A股传 CN。

REST拉取港股报价

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

import requests

API_BASE = "//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股长半小时。 港股上午9:30-12:00,下午13:00-16:00。比A股多半小时,做跨市场监控的时候要注意两边的交易时间窗口。

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

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

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

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

把前面几块拼起来,最终的完整脚本:

import json, time, threading, requests, websocket

API_BASE = "//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/

评论