沪深港通数据实战：从北向资金到择时信号的全链路量化（第 5 篇）：AH股比价：溢价率自研计算与套利信号

沪深港通数据实战：从北向资金到择时信号的全链路量化（第 5 篇）：AH股比价：溢价率自研计算与套利信号

一、前言

同一家公司同时在 A 股与港股上市时，两地价格因流动性、投资者结构、汇率等不同而存在价差。AH 溢价率（A 股价格 / H 股价格 − 1）是观察两个市场定价分歧的核心指标。本篇接入 ht/nbzj/ah，自研计算溢价率并与官方字段交叉校验，给出简单的「高溢价」提示信号。

二、接口与字段

2.1 AH 股比价 /ht/nbzj/ah

字段	含义（按官方字段名推断）
mc	名称
hdm / adm	港股代码 / A 股代码
hzxj / hzdf	港股最新价 / 港股涨跌幅
azxj / azdf	A 股最新价 / A 股涨跌幅
ahbj	AH 比价（A/H 价格比）
ahyj	AH 溢价（百分比）
t	时间

三、自研衍生指标

1. 自研 AH 溢价率：(azxj / hzxj − 1) × 100。
2. 交叉校验：与官方 ahyj 求绝对偏差，偏差应极小。
3. 高溢价信号：自研溢价率超过阈值（如 50%）标记为关注（仅演示，非投资建议）。

四、数据表设计

CREATE TABLE ht_ah (mc TEXT, hdm TEXT, hzxj TEXT, hzdf TEXT, adm TEXT, azxj TEXT,
  azdf TEXT, ahbj TEXT, ahyj TEXT, t TEXT);

五、完整代码

import urllib.request, json, sqlite3, time, logging
import pandas as pd

LICENCE = "YOUR_LICENCE_HERE"
DB_PATH = "quant.db"
LOG_FILE = "season4_article5.log"
BASE = "https://api.biyingapi.com"

logging.basicConfig(filename=LOG_FILE, level=logging.INFO,
                    format="%(asctime)s %(levelname)s %(message)s")

AH_COLS = ["mc", "hdm", "hzxj", "hzdf", "adm", "azxj", "azdf", "ahbj", "ahyj", "t"]


def biying_api_get_retry(url, timeout=15, max_retry=3):
    last = None
    for i in range(max_retry):
        try:
            req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
            with urllib.request.urlopen(req, timeout=timeout) as resp:
                return json.loads(resp.read().decode("utf-8"))
        except Exception as e:
            last = e
            logging.warning("retry %d/%d %s %s", i + 1, max_retry, url, e)
            time.sleep(1)
    logging.error("api failed %s %s", url, last)
    return None


def _as_list(data):
    return data if isinstance(data, list) else [data]


def save_ah():
    url = f"{BASE}/ht/nbzj/ah/{LICENCE}"
    data = biying_api_get_retry(url)
    if not data:
        return
    rows = _as_list(data)
    conn = sqlite3.connect(DB_PATH)
    coldef = ",".join(f"{c} TEXT" for c in AH_COLS)
    conn.execute(f"CREATE TABLE IF NOT EXISTS ht_ah ({coldef})")
    ph = ",".join("?" * len(AH_COLS))
    for r in rows:
        conn.execute(f"INSERT INTO ht_ah VALUES ({ph})",
                     [r.get(c) for c in AH_COLS])
    conn.commit(); conn.close()
    logging.info("ah saved %d rows", len(rows))


def ah_premium(threshold=50.0):
    conn = sqlite3.connect(DB_PATH)
    df = pd.read_sql("SELECT mc, hzxj, hzdf, azxj, azdf, ahyj FROM ht_ah", conn)
    conn.close()
    df["hzxj"] = pd.to_numeric(df["hzxj"], errors="coerce")
    df["azxj"] = pd.to_numeric(df["azxj"], errors="coerce")
    df["ahyj"] = pd.to_numeric(df["ahyj"], errors="coerce")
    df["self_premium"] = (df["azxj"] / df["hzxj"] - 1) * 100
    df["diff"] = (df["self_premium"] - df["ahyj"]).abs()
    df["flag"] = df["self_premium"] > threshold
    print(df[["mc", "self_premium", "ahyj", "diff", "flag"]].head(10).to_string())
    print("最大偏差:", df["diff"].max(), " 高溢价关注数:", int(df["flag"].sum()))


if __name__ == "__main__":
    save_ah()
    ah_premium(threshold=50.0)

六、业务关键点

• 汇率与红利税等因素会让 AH 长期共存溢价，溢价率高不等于「必然回归」，本信号仅作数据演示。
• 港股价格以港币计、A 股以人民币计，官方 ahbj/ahyj 已处理换算；自研校验能验证我们对字段含义的理解是否正确。
• hzxj / azxj 为 0 或缺失时会产生除零 / NaN，已用 pd.to_numeric(errors="coerce") 兜底。

七、拓展练习

1. 把高溢价标的与第 4 篇港股通行情的 pc 结合，看高溢价是否伴随港股更强。
2. 计算 AH 溢价率的历史分位（需长期采集），用分位替代固定阈值。
3. 统计 ahyj 与自研 self_premium 偏差最大的若干标的，排查字段口径差异。

八、下篇预告

下一篇《沪、深股通十大成交股：北向当日偏好》将接入 ht/nbzj/hgts 与 ht/nbzj/sgts，把北向（沪/深股通）当日最活跃成交股落库，量化北向资金的「主攻方向」。

免责申明：文中所有数据处理逻辑仅为编程演示，仅为数据演示，不构成投资建议。市场有风险，投资需谨慎。

转载自 CSDN-专业IT技术社区

原文链接：https://blog.csdn.net/2601_96840033/article/details/166773311