系列：《跨市场数据实战》｜连载项目 · 纯 GET 取数 · 仅依赖 requests
适用：想做「板块轮动热力 / 主力净流入个股榜」、但被 /higg /hizj /hibk 三个系列 13 个端点绕晕的读者；数据由智兔数服提供。本篇给这三组端点的分组地图、一套字段容错归一化代码、以及一个把「主力净流入个股」和「概念板块资金流」叠成轮动热力的实战模板，全部只依赖 requests，所有示例均为演示数据，不构成收益承诺。

1. 你将得到什么

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

1. 一张分组地图：/higg（个股资金流，6）+ /hizj（资金路线图，5）+ /hibk（港股通板块，2）共 13 个端点，知道「个股净流入 / 主力散户拆分 / 板块资金路线图 / 概念板块」分别敲哪个门；
2. 一套字段容错代码：资金流字段名分散，_hit_key + _to_float 带候选键兜底；
3. 一个轮动热力模板：用 hot_map 把「主力净流入最高的个股」和「资金流入最高的概念板块」一次性拉出来；
4. 五个真实踩坑点，尤其是 /hizj/ggjd 那个「近几日」路径参数（只认 3/5/10/20）。

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

2. 本篇取数约定

• 全部接口都是 GET + query 参数，token 放在查询串里（?token=xxx），不放 header；
• 统一基址 https://api.zhituapi.com；
• 代码块里的 你的智兔token 是占位符，换成你的 token 即可；
• /hizj/ggjd 带路径参数「近几日」（只认 3 / 5 / 10 / 20）；其余端点不带路径参数；
• 所有接口路径均取自官方文档。
• 数据来自 智兔数服（www.zhituapi.com）：零 SDK、纯 GET、免费版即可起步。

3. 13 个端点分三组

A 组：/higg 个股资金流（6 端点，均按对应口径倒序排列）

B 组：/hizj 资金路线图（5 端点）

C 组：/hibk 港股通板块（2 端点，按流入资金倒序）

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):
    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. 个股资金流归一 ----------
def norm_flow(rows):
    out = []
    for r in rows or []:
        out.append({
            "代码":        _hit_key(r, "dm", "code", default="-"),
            "名称":        _hit_key(r, "mc", "name", default="-"),
            "净流入额":     _to_float(_hit_key(r, "jlr", "净流入额")),
            "净流入率":     _to_float(_hit_key(r, "jlrl", "净流入率")),
            "主力净流入额":  _to_float(_hit_key(r, "zljlr", "主力净流入额")),
            "散户净流入额":  _to_float(_hit_key(r, "shzlr", "散户净流入额")),
            "散户净流入率":  _to_float(_hit_key(r, "shjlrl", "散户净流入率")),
        })
    return out


# ---------- 4. 取数封装 ----------
# A 组：个股资金流
def fetch_jlr():    return _get("/higg/jlr", default=[])
def fetch_jlrl():   return _get("/higg/jlrl", default=[])
def fetch_zljlr():  return _get("/higg/zljlr", default=[])
def fetch_zljlrl(): return _get("/higg/zljlrl", default=[])
def fetch_shzlr():  return _get("/higg/shzlr", default=[])
def fetch_shjlrl(): return _get("/higg/shjlrl", default=[])
# B 组：资金路线图
def fetch_zjh():    return _get("/hizj/zjh", default=[])
def fetch_bk():     return _get("/hizj/bk", default=[])
def fetch_ggzl():   return _get("/hizj/ggzl", default=[])
def fetch_ggjd(d=5): return _get("/hizj/ggjd/%d" % d, default=[])   # d 只认 3/5/10/20
def fetch_lxlr():   return _get("/hizj/lxlr", default=[])
# C 组：港股通板块
def fetch_hibk_zjhhy(): return _get("/hibk/zjhhy", default=[])
def fetch_hibk_gnbk():  return _get("/hibk/gnbk", default=[])


# ---------- 5. 实战：板块轮动热力 ----------
def hot_map(top=10):
    """主力净流入最高的个股 + 资金流入最高的概念板块"""
    stocks = norm_flow(fetch_zljlr())[:top]      # 接口已按主力净流入额倒序
    sectors = fetch_hibk_gnbk()[:top]            # 接口已按流入资金倒序
    return {"主力净流入个股": stocks, "概念板块资金流": sectors}


# ---------- 6. 校验 ----------
def run_check():
    global fetch_zljlr, fetch_hibk_gnbk
    assert _hit_key({"DM": "00700", "mc": "腾讯"}, "dm") == "00700"
    assert _to_float("-") is None and _to_float("3.2") == 3.2

    fake = [{"dm": "00700", "mc": "腾讯", "zljlr": "520.0", "jlr": "400.0", "shzlr": "-120.0"},
            {"dm": "09988", "mc": "阿里", "zljlr": "300.0", "jlr": "280.0", "shzlr": "-80.0"}]
    nf = norm_flow(fake)
    assert nf[0]["代码"] == "00700" and nf[0]["主力净流入额"] == 520.0
    assert nf[0]["散户净流入额"] == -120.0

    # 无网环境：用假数据模拟 hot_map
    def _fake_zljlr():
        return fake
    def _fake_gnbk():
        return [{"bk": "AI", "je": "800"}, {"bk": "医药", "je": "500"}]
    _o1, _o2 = fetch_zljlr, fetch_hibk_gnbk
    fetch_zljlr, fetch_hibk_gnbk = _fake_zljlr, _fake_gnbk
    hm = hot_map(1)
    fetch_zljlr, fetch_hibk_gnbk = _o1, _o2
    assert hm["主力净流入个股"][0]["名称"] == "腾讯"
    assert hm["概念板块资金流"][0]["bk"] == "AI"

    print("校验通过")


if __name__ == "__main__":
    run_check()
    print("-" * 62)
    for name, path in [("个股净流入额", "/higg/jlr"),
                       ("个股净流入率", "/higg/jlrl"),
                       ("主力净流入额", "/higg/zljlr"),
                       ("主力净流入率", "/higg/zljlrl"),
                       ("散户净流入额", "/higg/shzlr"),
                       ("散户净流入率", "/higg/shjlrl"),
                       ("证监会行业路线图", "/hizj/zjh"),
                       ("概念板块路线图", "/hizj/bk"),
                       ("个股净流入总览", "/hizj/ggzl"),
                       ("个股阶段(近5日)", "/hizj/ggjd/5"),
                       ("主力连续流入", "/hizj/lxlr"),
                       ("证监会行业板块", "/hibk/zjhhy"),
                       ("概念板块", "/hibk/gnbk")]:
        data = _get(path, default=[])
        if isinstance(data, dict) and "_error" in data:
            print("%-14s %-18s -> %s" % (name, path, data["_error"][:50]))
        else:
            print("%-14s %-18s -> %d 条" % (name, path, len(data)))

5. 跑通示例

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

6. 坑与注意事项

坑 1：/hizj/ggjd 的「近几日」是路径参数，只认 3/5/10/20。
/hizj/ggjd/5 才对；写 /hizj/ggjd?d=5 或传 7 会 404。ggzl 是总览（按代码升序），不区分窗口。

坑 2：/higg/* 的排名已倒序，别再排序。
净流入额、主力净流入额等接口上游已按对应口径倒序返回；hot_map 直接切片取前 N 即可，自己再 sort 反而可能破坏上游口径。

坑 3：主力 / 散户是拆分不是加总。
zljlr（主力净流入额）+ shzlr（散户净流入额）≈ jlr（净流入额），但具体口径上游可能含「大户」等其它分项；做资金结构分析时按上游口径拆分，别硬凑等式。

坑 4：/hizj/zjh 与 /hibk/zjhhy 不是同一张表。
前者是「证监会行业资金路线图」（含近 3/5/10 天流入趋势），后者是「证监会行业资金流向」（单日倒序）；一个看趋势一个看当日，做轮动要分开用。

坑 5：/hizj/lxlr 看「连续流入天数」不是单日金额。
它按流入天数倒序排主力连续净流入/流出，适合找「持续吸筹」的票；别把它和单日净流入额（zljlr）混为一谈。

7. 小结与下篇预告

本篇把 /higg（6）+ /hizj（5）+ /hibk（2）共 13 个资金流端点打通，给出 norm_flow（个股资金流归一）与 hot_map（主力净流入个股 + 概念板块资金流）两个核心函数。ggjd 的近几日路径参数只认 3/5/10/20、排名已倒序不重排、主力/散户是拆分不是加总，是这套资金流接口最容易踩错的地方。

下一篇计划写 #08《可转债套利数据：3个接口打通比价、列表与现货》：用 /kzz（比价表、一览、实时行情）用比价 + 现货发现折价/溢价套利机会，收尾整个《跨市场数据实战》系列。

8. 免责声明

本文仅演示港股资金流与板块轮动类接口的取数与归一化方法，所有代码示例均为演示数据，未含任何真实资金流数值，不构成投资建议，亦不承诺收益。

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

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

把代码里的 你的智兔token 换成你拿到的真实 token，上面的脚本就能直接打印当日主力净流入个股榜与概念板块资金流热力。