系列：《跨市场数据实战》｜连载项目 · 纯 GET 取数 · 仅依赖 requests
适用：想从龙虎榜里分清「游资接力」和「机构真买」、但被一堆上榜统计和席位明细绕晕的读者；数据由智兔数服提供。本篇给龙虎榜 + 机构席位两条线共 9 个端点的分组地图、一套字段容错归一化代码、以及把一只涨停股背后买卖盘还原出来的实战模板，全部只依赖 requests，所有示例均为演示数据，不构成收益承诺。

1. 你将得到什么

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

1. 一张分组地图：龙虎榜 5 个端点（/hilh）+ 机构席位 4 个端点（/hijg），知道「今日概览 / 个股上榜 / 营业部上榜 / 机构追踪 / 成交明细 / 机构·基金·社保·QFII 重仓」分别敲哪个门；
2. 一套字段容错代码：上游返回的字段名常有大小写/中英混用，本篇用 _hit_key 按顺序兜底，不用每次为字段名抓狂；
3. 一个还原模板：用「机构席位成交明细 + 机构持股汇总」两只接口，把某只涨停股背后游资和机构的进出轨迹拼出来；
4. 五个真实踩坑点，都是文档里写了、但第一次用几乎一定会踩的（尤其是那几个路径参数）。

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

2. 本篇取数约定

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

3. 9 个端点分两组

先建立地图。龙虎榜与机构席位两条线一共 9 个端点：

注意两个参数坑：/hilh/ggsb、yybsb、jgxw 用路径参数 {n}，且 n 只能是 5/10/30/60；/hijg/* 四个接口用 /{年}/{季度} 路径参数，季度 1=一季报、2=中报、3=三季报、4=年报，年份取 1989~当年。

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. 龙虎榜·每日概览 (/hilh/mrxq) ----------
def norm_daily(rows):
    """今日龙虎榜概览 -> [{代码,名称,原因,买入额,卖出额,净额}]"""
    out = []
    for r in rows or []:
        out.append({
            "代码":   _hit_key(r, "dm", "code", default="-"),
            "名称":   _hit_key(r, "mc", "name", default="-"),
            "原因":   _hit_key(r, "yy", "reason", "sm", default="-"),
            "买入额": _to_float(_hit_key(r, "mrje", "mr", "buy")),
            "卖出额": _to_float(_hit_key(r, "mcje", "mc_", "sell")),
            "净额":   _to_float(_hit_key(r, "je", "net", "netje")),
        })
    return out


# ---------- 4. 龙虎榜·个股上榜 (/hilh/ggsb/{n}) ----------
def norm_ggsb(rows):
    """近 n 日上榜个股 -> [{代码,名称,上榜次数,净额}]"""
    out = []
    for r in rows or []:
        out.append({
            "代码":     _hit_key(r, "dm", default="-"),
            "名称":     _hit_key(r, "mc", default="-"),
            "上榜次数": _to_float(_hit_key(r, "count", "sbsc")),
            "净额":     _to_float(_hit_key(r, "je", "net")),
        })
    return out


# ---------- 5. 龙虎榜·营业部上榜 (/hilh/yybsb/{n}) ----------
def norm_yybsb(rows):
    """营业部上榜统计 -> [{营业部,上榜次数,买入席位数,买入前三}]"""
    out = []
    for r in rows or []:
        out.append({
            "营业部":     _hit_key(r, "yybmc", "yyb", default="-"),
            "上榜次数":   _to_float(_hit_key(r, "count")),
            "买入席位数": _to_float(_hit_key(r, "xs", "mrxs")),
            "买入前三":   _hit_key(r, "top3", "mr3", default="-"),
        })
    return out


# ---------- 6. 龙虎榜·机构追踪 (/hilh/jgxw/{n}) ----------
def norm_jgxw(rows):
    """机构席位追踪 -> [{代码,名称,买入席位数,卖出次数,净额万}]"""
    out = []
    for r in rows or []:
        out.append({
            "代码":       _hit_key(r, "dm", default="-"),
            "名称":       _hit_key(r, "mc", default="-"),
            "买入席位数": _to_float(_hit_key(r, "xs")),
            "卖出次数":   _to_float(_hit_key(r, "ende", "mcxs")),
            "净额万":     _to_float(_hit_key(r, "je", "net")),
        })
    return out


# ---------- 7. 龙虎榜·机构成交明细 (/hilh/xwmx) ----------
def norm_xwmx(rows):
    """机构席位成交明细 -> [{代码,名称,机构净额万,原因}]"""
    out = []
    for r in rows or []:
        out.append({
            "代码":       _hit_key(r, "dm", default="-"),
            "名称":       _hit_key(r, "mc", default="-"),
            "机构净额万": _to_float(_hit_key(r, "je", "orgnet", "net")),
            "原因":       _hit_key(r, "yy", "reason", default="-"),
        })
    return out


# ---------- 8. 机构持仓汇总 (/hijg/jgcghz/{y}/{q}) ----------
def norm_org_hold(rows):
    """机构持股汇总 -> [{代码,名称,持股数,市值,占流通比}]"""
    out = []
    for r in rows or []:
        out.append({
            "代码":     _hit_key(r, "dm", default="-"),
            "名称":     _hit_key(r, "mc", default="-"),
            "持股数":   _to_float(_hit_key(r, "cgs", "hold", "shares")),
            "市值":     _to_float(_hit_key(r, "sz", "mv", "value")),
            "占流通比": _to_float(_hit_key(r, "ltb", "ratio")),
        })
    return out


# ---------- 9. 取数封装 ----------
def fetch_daily():               return _get("/hilh/mrxq", default=[])
def fetch_ggsb(n=5):             return _get("/hilh/ggsb/%s" % n, default=[])
def fetch_yybsb(n=5):            return _get("/hilh/yybsb/%s" % n, default=[])
def fetch_jgxw(n=5):             return _get("/hilh/jgxw/%s" % n, default=[])
def fetch_xwmx():                return _get("/hilh/xwmx", default=[])
def fetch_org_hold(y=2021, q=1): return _get("/hijg/jgcghz/%s/%s" % (y, q), default=[])
def fetch_fund_hold(y=2021, q=1):return _get("/hijg/jj/%s/%s" % (y, q), default=[])
def fetch_sb_hold(y=2021, q=1):  return _get("/hijg/sb/%s/%s" % (y, q), default=[])
def fetch_qfii_hold(y=2021, q=1):return _get("/hijg/qf/%s/%s" % (y, q), default=[])


# ---------- 10. 实战：还原一只涨停股背后的买卖盘 ----------
def trace_one_stock(dm, n=5):
    """用 xwmx(机构成交明细) + jgxw(机构追踪) 拼出某只股票的机构进出概览"""
    detail = norm_xwmx(fetch_xwmx())
    track = norm_jgxw(fetch_jgxw(n))
    row_d = [x for x in detail if _hit_key(x, "dm") == dm]
    row_t = [x for x in track if _hit_key(x, "dm") == dm]
    return {
        "代码":       dm,
        "机构成交明细": row_d[0] if row_d else None,
        "机构席位追踪": row_t[0] if row_t else None,
    }


# ---------- 11. 校验 ----------
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) 每日概览
    daily = [{"dm": "000001", "mc": "平安银行", "yy": "涨停",
              "mrje": "1000", "mcje": "800", "je": "200"}]
    d = norm_daily(daily)
    assert d[0]["净额"] == 200.0 and d[0]["名称"] == "平安银行"

    # 3) 个股上榜
    g = [{"dm": "600519", "mc": "贵州茅台", "count": "3", "je": "1500"}]
    assert norm_ggsb(g)[0]["上榜次数"] == 3.0

    # 4) 营业部上榜
    y = [{"yybmc": "华泰某营业部", "count": "10", "xs": "8", "top3": "600519,000001,300750"}]
    yy = norm_yybsb(y)
    assert yy[0]["买入席位数"] == 8.0 and "华泰" in yy[0]["营业部"]

    # 5) 机构追踪
    j = [{"dm": "300750", "mc": "宁德", "xs": "5", "ende": "2", "je": "900"}]
    assert norm_jgxw(j)[0]["净额万"] == 900.0

    # 6) 机构成交明细
    x = [{"dm": "300750", "mc": "宁德", "je": "1200", "yy": "机构净买"}]
    assert norm_xwmx(x)[0]["机构净额万"] == 1200.0

    # 7) 机构持股汇总
    o = [{"dm": "600519", "mc": "茅台", "cgs": "900", "sz": "15000000", "ltb": "7.1"}]
    assert norm_org_hold(o)[0]["占流通比"] == 7.1

    print("校验通过")


if __name__ == "__main__":
    run_check()
    print("-" * 62)
    for name, path in [("龙虎榜每日", "/hilh/mrxq"),
                       ("个股上榜(5日)", "/hilh/ggsb/5"),
                       ("营业部上榜(5日)", "/hilh/yybsb/5"),
                       ("机构追踪(5日)", "/hilh/jgxw/5"),
                       ("机构成交明细", "/hilh/xwmx"),
                       ("机构持股汇总", "/hijg/jgcghz/2021/1")]:
        data = _get(path, default=[])
        if isinstance(data, dict) and "_error" in data:
            print("%-14s %-22s -> %s" % (name, path, data["_error"][:60]))
        else:
            print("%-14s %-22s -> %d 条" % (name, path, len(data)))

5. 跑通示例

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

6. 坑与注意事项

坑 1：/hilh/ggsb、yybsb、jgxw 的 {n} 不是任意整数。
文档明确：n 只能是 5 / 10 / 30 / 60。传 7、20 这类值上游多半报错或返回空。fetch_ggsb(n=5) 默认 5，要改就从这四档里挑。

坑 2：/hijg/* 四个接口是 /{年}/{季度} 路径参数，不是查询参数。
/hijg/jgcghz/2021/1 才对；年份取 1989~当年，季度 1=一季报 / 2=中报 / 3=三季报 / 4=年报。写成 ?year=2021&quarter=1 会 404。

坑 3：龙虎榜两条更新频率不一样。
/hilh/mrxq 是每日 20:00 更新，ggsb / yybsb / jgxw / xwmx 是每日 15:40。如果你在 15:40 之前去拉 mrxq，可能拿到的是前一天的 stale 数据；要拼「当日完整买卖盘」，建议 20:00 之后再跑 mrxq，15:40 后即可跑其余四个。

坑 4：xwmx 是固定近 5 个交易日窗口，没有参数。
想看更长窗口的机构成交，去用 /hilh/jgxw/{n}（n 最大 60）而不是 xwmx。

坑 5：机构四条线（jgcghz/jj/sb/qf）是周更（每周六 14:00）。
季度数据有滞后，做「当季机构调仓」分析时，注意 /hijg/jj 这类返回的是最近一个已披露季度的重仓，不是实时。

7. 小结与下篇预告

本篇把龙虎榜（5 个端点）与机构席位（4 个端点）两条线打通，给出一套字段容错归一化代码，并用 trace_one_stock 把「机构成交明细 + 机构追踪」拼成一只股票背后的机构进出概览。路径参数（{n} 四档、{年}/{季度}）是这套接口最容易踩错的地方，已在取数封装里固定好默认值。

下一篇计划写 #04《沪深A股异动排名与指数历史K线：23个接口覆盖连涨连跌、估值排名与BOLL/KDJ/MA/MACD》：用 /himk（沪深A股成交量异动、连涨连跌、市盈率/市净率/ROE 排名）与 /hz（指数实时分时、历史 BOLL/KDJ/MA/MACD）两套接口，监控沪深A股异动并叠加指数技术指标。

8. 免责声明

本文仅演示龙虎榜与机构席位数据的取数与字段归一化方法，所有代码示例均为演示数据，未含任何真实行情数值，不构成投资建议，亦不承诺收益。

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

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

把代码里的 你的智兔token 换成你拿到的真实 token，上面的脚本就能直接打印龙虎榜与机构席位的买卖盘数据。