系列：《跨市场数据实战》｜连载项目 · 纯 GET 取数 · 仅依赖 requests
适用：想看北向/南向资金、但被「日频还是季频」「万元还是百万」绕晕的读者；数据由智兔数服提供。本篇给港股通板块 21 个端点的分组地图、三套口径的归一化代码、季频降级守卫，全部只依赖 requests，所有示例均为演示数据，不构成收益承诺。

1. 你将得到什么

读完这一篇，你能拿走四样东西：

1. 一张分组地图：港股通板块 21 个端点按用途分成 8 组，知道什么数据该敲哪个门；
2. 一套归一化代码：官方接口里同一个"资金流入"有万、百万、万元三种单位，本篇统一换算到「亿元」，不用每次心算；
3. 一道季频守卫：北向个股排名接口已经从日频降级为季频，如果你还按日频去算"连续 5 日增持"，算出来的东西是空的——本篇用代码把这道错显式拦住；
4. 五个真实踩坑点，都是文档里写了、但第一次用几乎一定会踩的。

代码全部自包含，复制进 .py 直接能跑，不依赖 numpy / pandas。

2. 本篇取数约定

• 全部接口都是 GET + query 参数，token 放在查询串里（?token=xxx）；
• 统一基址 https://api.zhituapi.com；
• 代码块里的 你的智兔token 是占位符，换成你的 token 即可；
• 所有接口路径均取自官方文档。
• 数据来自 智兔数服（www.zhituapi.com）：零 SDK、纯 GET、免费版即可起步。

3. 21 个端点分 8 组

先建立地图。港股通板块一共 21 个端点，按用途分：

注意最后一行——前 7 组都是日频，只有个股排名组是季频。这就是全篇最大的坑。

4. 核心模板函数

import requests, time

BASE = "https://api.zhituapi.com"
TOKEN = "你的智兔token"

# ---------- 1. 字段容错与类型归一 ----------
def _hit_key(d, *cands, default=None):
    """字段容错：接口偶发大小写/中英文混用时，按顺序取第一个非空值"""
    if not isinstance(d, dict):
        return default
    for c in cands:
        if c in d and d[c] not in (None, "", "-", "null"):
            return d[c]
    low = {str(k).lower(): v for k, v in d.items()}
    for c in cands:
        v = low.get(str(c).lower())
        if v not in (None, "", "-", "null"):
            return v
    return default


def _to_float(v, default=None):
    try:
        if v in (None, "", "-", "null", "None"):
            return default
        return float(v)
    except (TypeError, ValueError):
        return default


# ---------- 2. 统一请求：重试 + 退避 + 降级 ----------
def _get(path, params=None, timeout=10, retries=2, backoff=0.6, default=None):
    """返回 JSON；失败重试 retries 次仍失败则返回 {'_error': 原因}"""
    q = {"token": TOKEN}
    if params:
        q.update(params)
    last = ""
    for i in range(retries + 1):
        try:
            r = requests.get(BASE + path, params=q, timeout=timeout)
            if r.status_code == 200:
                try:
                    return r.json()
                except ValueError:
                    return default
            last = "HTTP %s %s" % (r.status_code, (r.text or "").strip()[:80])
        except Exception as e:
            last = "%s: %s" % (type(e).__name__, e)
        if i < retries:
            time.sleep(backoff * (i + 1))
    return {"_error": last}


# ---------- 3. 单位归一：全部换算到「亿元」 ----------
UNIT_IN_YI = {          # 各接口原始单位 -> 1 单位等于多少亿元
    "万":   1e-4,       # lxgl 的 netbuy / netin / remain
    "百万": 1e-2,       # bxls / nxls 的 bx / hgt / sgt
    "万元": 1e-4,       # bxzl 的 bxall / hgtall / sgtall
    "元":   1e-8,
}


def to_yi(v, unit="万"):
    """原始金额 -> 亿元；None / 空串 / 异常值返回 None（不猜、不填 0）"""
    f = _to_float(v)
    if f is None:
        return None
    return round(f * UNIT_IN_YI.get(unit, 1.0), 4)


# ---------- 4. 三套口径的归一化 ----------
def norm_overview(rows):
    """/ht/nbzj/lxgl -> 只留北向四条通道（单位：亿元）"""
    out = []
    for r in rows or []:
        if _hit_key(r, "dir", default="") != "北向":
            continue
        out.append({
            "板块":   _hit_key(r, "tname", default="-"),
            "净买额": to_yi(_hit_key(r, "netbuy"), "万"),
            "净流入": to_yi(_hit_key(r, "netin"), "万"),
            "余额":   to_yi(_hit_key(r, "remain"), "万"),
            "上涨":   _to_float(_hit_key(r, "up"), 0),
            "下跌":   _to_float(_hit_key(r, "down"), 0),
            "状态":   _hit_key(r, "status", default="-"),
        })
    return out


def norm_series(rows, unit="百万"):
    """/ht/nbzj/bxls|nxls/{jd} -> 统一 [{'日期','北向','沪股通','深股通'}]，亿元"""
    out = []
    for r in rows or []:
        out.append({
            "日期":   _hit_key(r, "t", "date", default="-"),
            "北向":   to_yi(_hit_key(r, "bx", "nx"), unit),
            "沪股通": to_yi(_hit_key(r, "hgt", "ggth"), unit),
            "深股通": to_yi(_hit_key(r, "sgt", "ggts"), unit),
        })
    return out


def norm_total(d):
    """/ht/nbzj/bxzl -> 累计净流入（亿元）"""
    return {
        "北向累计":   to_yi(_hit_key(d, "bxall"), "万元"),
        "沪股通累计": to_yi(_hit_key(d, "hgtall"), "万元"),
        "深股通累计": to_yi(_hit_key(d, "sgtall"), "万元"),
    }


# ---------- 5. 季频守卫 ----------
SNAPSHOT_FIELDS = ("jrcg", "jrsz", "jrltb", "jrzgbb")              # 快照日持股，仍有效
PERIOD_FIELDS   = ("zqzc", "zqsz", "zqszzf", "zqltb", "zqzgbb")    # 周期增持，多已停更


def norm_rank(rows):
    """/ht/nbzj/bxpm|hgpm|sgpm/{zq} -> (排名列表, 元信息)；自动丢弃全 null 的周期字段"""
    if not rows:
        return [], {"季度标记": "unknown", "快照日": "-", "丢弃字段": [], "条数": 0}
    first = rows[0]
    quarterly = str(_hit_key(first, "_freq", default="")).lower() == "quarterly"
    dropped = [f for f in PERIOD_FIELDS
               if all(_hit_key(r, f) in (None, "", "null") for r in rows)]
    out = [{
        "代码":         _hit_key(r, "dm", default="-"),
        "名称":         _hit_key(r, "mc", default="-"),
        "板块":         _hit_key(r, "ssbk", default="-"),
        "持股万股":     _to_float(_hit_key(r, "jrcg")),
        "持股市值万元": _to_float(_hit_key(r, "jrsz")),
        "占流通股比":   _to_float(_hit_key(r, "jrltb")),
    } for r in rows]
    meta = {"季度标记": "quarterly" if quarterly else "daily",
            "快照日": _hit_key(first, "t", default="-"),
            "丢弃字段": dropped, "条数": len(out)}
    return out, meta


def assert_not_daily(meta):
    """把季频数据当日频用是本篇最想拦住的错误；季频返回 False"""
    return meta.get("季度标记") != "quarterly"


# ---------- 6. 成分列表交叉核对 ----------
def duplicate_slots(rows_a, rows_b):
    """ggth / ggts 两个港股通成分列表可能完全一致，返回重复代码数"""
    a = set(_hit_key(r, "dm", default="") for r in rows_a or [])
    b = set(_hit_key(r, "dm", default="") for r in rows_b or [])
    if not a or not b:
        return 0
    return len(a & b)


# ---------- 7. 取数封装 ----------
def fetch_overview():            return _get("/ht/nbzj/lxgl", default=[])
def fetch_north_series(jd="1"):  return _get("/ht/nbzj/bxls/%s" % jd, default=[])
def fetch_south_series(jd="1"):  return _get("/ht/nbzj/nxls/%s" % jd, default=[])
def fetch_north_total():         return _get("/ht/nbzj/bxzl", default={})
def fetch_slot(kind="hgtc"):     return _get("/ht/nbzj/%s" % kind, default=[])
def fetch_rank(scope="bx", zq="1"):
    m = {"bx": "bxpm", "hgt": "hgpm", "sgt": "sgpm"}
    return _get("/ht/nbzj/%s/%s" % (m.get(scope, "bxpm"), zq), default=[])


# ---------- 8. 校验 ----------
def run_check():
    # 1) 字段容错
    assert _hit_key({"Dm": "000001", "mc": "平安银行"}, "dm") == "000001"
    assert _to_float("-") is None and _to_float("12.5") == 12.5

    # 2) 单位换算：三个口径都归一到亿元
    assert to_yi(10000, "万") == 1.0       # 10000 万 = 1 亿
    assert to_yi(100, "百万") == 1.0       # 100 百万 = 1 亿
    assert to_yi(10000, "万元") == 1.0
    assert to_yi(None, "万") is None

    # 3) 概览只留北向
    fake = [
        {"dir": "北向", "tname": "沪股通(港>沪)", "netbuy": "12345", "netin": "13000",
         "remain": "98765", "up": "800", "down": "300", "status": "3"},
        {"dir": "南向", "tname": "港股通(沪>港)", "netbuy": "9999", "netin": "9999",
         "remain": "0", "up": "1", "down": "1", "status": "3"},
    ]
    ov = norm_overview(fake)
    assert len(ov) == 1 and ov[0]["板块"].startswith("沪股通")
    assert ov[0]["净买额"] == 1.2345

    # 4) 走势归一：分项之和应等于合计
    ser = norm_series([{"t": "2026-08-28", "bx": "120", "hgt": "60", "sgt": "60"}])
    assert ser[0]["日期"] == "2026-08-28" and ser[0]["北向"] == 1.2
    assert abs(ser[0]["沪股通"] + ser[0]["深股通"] - 1.2) < 1e-9

    # 5) 累计总览
    tot = norm_total({"bxall": "190000", "hgtall": "100000", "sgtall": "90000"})
    assert tot["北向累计"] == 19.0 and tot["沪股通累计"] == 10.0

    # 6) 季频守卫：zq* 全 null 应被丢弃，且不可当日频用
    fake_rank = [
        {"dm": "600519", "mc": "X", "ssbk": "沪股通", "jrcg": "9000", "jrsz": "15000000",
         "jrltb": "7.1", "zqzc": None, "zqsz": None, "zqszzf": None,
         "zqltb": None, "zqzgbb": None, "t": "2026-06-30", "_freq": "quarterly"},
        {"dm": "000858", "mc": "Y", "ssbk": "深股通", "jrcg": "5000", "jrsz": "800000",
         "jrltb": "3.2", "zqzc": None, "zqsz": None, "zqszzf": None,
         "zqltb": None, "zqzgbb": None, "t": "2026-06-30", "_freq": "quarterly"},
    ]
    rank, meta = norm_rank(fake_rank)
    assert meta["季度标记"] == "quarterly"
    assert meta["快照日"] == "2026-06-30"
    assert set(meta["丢弃字段"]) == set(PERIOD_FIELDS)
    assert assert_not_daily(meta) is False          # 季频 -> 不允许当日频用
    assert rank[0]["持股市值万元"] > rank[1]["持股市值万元"]

    # 7) 成分列表重复检测
    a = [{"dm": "00700"}, {"dm": "00001"}]
    assert duplicate_slots(a, a) == 2
    assert duplicate_slots(a, [{"dm": "09988"}]) == 0

    print("校验通过")


if __name__ == "__main__":
    run_check()
    print("-" * 62)
    for name, path in [("当日概览", "/ht/nbzj/lxgl"),
                       ("北向历史走势", "/ht/nbzj/bxls/1"),
                       ("北向历史总览", "/ht/nbzj/bxzl"),
                       ("沪股通成分股", "/ht/nbzj/hgtc"),
                       ("AH股比价", "/ht/nbzj/ah"),
                       ("北向个股排名", "/ht/nbzj/bxpm/1")]:
        data = _get(path, default=[])
        if isinstance(data, dict) and "_error" in data:
            print("%-12s %-20s -> %s" % (name, path, data["_error"][:60]))
        else:
            print("%-12s %-20s -> %d 条" % (name, path, len(data)))

5. 跑通示例

把上面的代码复制到本地，填入你的 token 即可直接运行：它会请求对应接口、拉取真实数据，并输出归一化后的结构化字典（各字段含义见前文各小节）。

6. 坑与注意事项

坑 1：个股排名已经不是日频了。
/ht/nbzj/bxpm/{zq}、hgpm/{zq}、sgpm/{zq} 三个接口的上游日频持股明细已停更，现在返回的是最近季末的持股快照。返回的 zq* 系列字段（周期增持股数、增持市值、增幅、占流通股比、占总股本比）多数为 null，只有 jrcg / jrsz / jrltb / jrzgbb 这四个快照字段有值，_freq 会标成 quarterly。
所以"北向连续 5 日净增持某某股"这类逻辑现在已经做不出来了，能做的是"最新季末北向持股排名"。norm_rank 会自动把全 null 的周期字段丢掉，assert_not_daily 会在你误用时返回 False。

坑 2：单位有三种，别混着加。
lxgl 用「万」，bxls/nxls 用「百万」，bxzl 用「万元」。直接把 bx 和 netbuy 相加会差 100 倍。to_yi() 把三者统一到亿元，这是本篇最容易被忽略、也最容易出静默错误的地方。

坑 3：成交净买额 ≠ 资金净流入。
netbuy 是买入成交额减卖出成交额，netin 是当日限额减当日余额（含挂单未成交部分）。两者口径不同、数值不同，别当成同一个指标轮着用。

坑 4：总览接口只有累计口径。
/ht/nbzj/bxzl 只返回 bxall / hgtall / sgtall 三个累计值，没有近一月/近六月/近一年的分阶段字段。想要分阶段，去用 /ht/nbzj/bxls/{jd} 自己按日期切窗口。

坑 5：两个港股通成分列表可能完全一致。
/ht/nbzj/ggth（港股通·沪）与 /ht/nbzj/ggts（港股通·深）实测返回的成分列表可能完全相同。如果你的策略要区分这两个通道，先用 duplicate_slots() 比一下再决定要不要合并去重，别默认它们天然不同。

7. 小结与下篇预告

本篇把港股通板块 21 个端点分成 8 组，给出三套口径（万/百万/万元）统一到亿元的归一化代码，并用 norm_rank + assert_not_daily 把"季频当日频用"这个错误显式拦住。

下一篇计划写 #02《基金持仓穿透：32个接口从基金列表查到重仓股变动》：用 /jh 与 /js 两组接口，从基金代码反查它持有哪些股票，再统计哪些股票被最多基金共同重仓。

8. 免责声明

本文仅演示跨市场资金数据的取数与口径归一化方法，所有代码示例均为演示数据，未含任何真实行情数值，不构成投资建议，亦不承诺收益。

免费领取证书
数据来自 智兔数服（www.zhituapi.com）：零 SDK、纯 GET、免费版即可起步。

领取路径：进入 www.zhituapi.com → 点击「请求证书」→「证书获取」→「免费版」（邮箱验证 3 步即可拿到 token）。

把代码里的 你的智兔token 换成你拿到的真实 token，上面的脚本就能直接打印北向资金与港股通成分股数据。