系列：《跨市场数据实战》｜连载项目 · 纯 GET 取数 · 仅依赖 requests
适用：想知道"某只基金到底买了什么"、"哪些股票被最多基金共同重仓"的读者；数据由智兔数服提供。本篇给三条查询链路的完整代码、多期持仓的取最新期处理、报告期回溯，全部只依赖 requests，所有示例均为演示数据，不构成收益承诺。

1. 你将得到什么

"持仓穿透"其实是两个方向的问题，本篇一次讲清：

• 自下而上：我知道几只基金代码，想把它们的最新一期股票持仓全拉出来，聚合成一张"隐形重仓股"表——哪些股票被最多基金同时持有、合计持仓市值多少；
• 自上而下：我不关心具体基金，只想直接看全市场基金重仓股排名，以及相对上一季度的增减变动。

读完你能拿走：

1. 一张 32 端点的分组地图（/jh 14 个 + /js 15 个 + /fund 3 个）；
2. 穿透聚合代码：多只基金持仓汇总，自动按报告期取最新、自动去重；
3. 报告期回溯：查询未披露的季度会返回空，代码自动往前退到最近一个已披露报告期；
4. 五个真实踩坑点。

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

2. 本篇取数约定

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

3. 32 个端点分 3 组

基金板块一共 32 个端点，分属三组，职责完全不同：

一句话区分：/jh 管单只基金的档案，/js 管全市场横向排名，/fund 管场内实时。

三条链路要用到的核心端点：

4. 核心模板函数

import requests, time
from collections import defaultdict

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. 链路 A：自下而上的穿透聚合 ----------
LIST_PATH = {
    "all":    "/jh/list/all",      # 所有基金（开放/封闭/分级）
    "etf":    "/jh/hq/etflist",    # ETF
    "lof":    "/jh/hq/loflist",    # LOF
    "close":  "/jh/hq/list",       # 封闭式
    "hs":     "/fund/list/all",    # 沪深基金列表（另一套代码体系）
    "hs_etf": "/fund/list/etf",
}


def fetch_fund_list(kind="all"):
    return _get(LIST_PATH.get(kind, LIST_PATH["all"]), default=[])


def fetch_stock_holdings(code):
    """/jh/zh/gpcc/{code} -> 历年投资股票组合，按截止时间降序"""
    return _get("/jh/zh/gpcc/%s" % code, default=[])


def fetch_bond_holdings(code):
    """/jh/zh/zccc/{code} -> 历年投资债券持仓，按截止时间降序"""
    return _get("/jh/zh/zccc/%s" % code, default=[])


def fetch_profile(code):
    """/jh/base/jjgk/{code} -> 基金概况"""
    return _get("/jh/base/jjgk/%s" % code, default={})


def latest_period(rows):
    """持仓接口返回多期数据，取最新报告期（按 t 最大）"""
    dates = sorted({_hit_key(r, "t", default="") for r in (rows or []) if _hit_key(r, "t")})
    if not dates:
        return list(rows or [])
    top = dates[-1]
    return [r for r in rows if _hit_key(r, "t", default="") == top]


def penetrate(holdings_by_fund, topn=20):
    """穿透聚合：多只基金最新一期持仓 -> 隐形重仓股

    holdings_by_fund: {基金代码: [持仓行, ...]}，各行应已用 latest_period 取过最新期
    排序：先按被持有基金数降序，再按合计持仓市值降序
    """
    agg = defaultdict(lambda: {"名称": "-", "基金数": 0,
                               "合计市值万元": 0.0, "合计持股万股": 0.0})
    for fund, rows in (holdings_by_fund or {}).items():
        seen = set()
        for r in rows:
            dm = _hit_key(r, "dm", default="")
            if not dm or dm in seen:      # 同一基金同一期重复行只算一次
                continue
            seen.add(dm)
            a = agg[dm]
            a["基金数"] += 1
            a["名称"] = _hit_key(r, "mc", default="-")
            a["合计市值万元"] += _to_float(_hit_key(r, "ccsz"), 0.0) or 0.0
            a["合计持股万股"] += _to_float(_hit_key(r, "cgs"), 0.0) or 0.0
    out = [{"代码": k, "名称": v["名称"], "基金数": v["基金数"],
            "合计市值万元": round(v["合计市值万元"], 2),
            "合计持股万股": round(v["合计持股万股"], 2)} for k, v in agg.items()]
    out.sort(key=lambda x: (-x["基金数"], -x["合计市值万元"]))
    return out[:topn]


# ---------- 4. 链路 B：自上而下的官方排名与变动 ----------
def fetch_top_stocks(year, q):
    """/js/other/jjzc/{y}_{q} -> 基金重仓个股排名，按基金覆盖面降序"""
    return _get("/js/other/jjzc/%d_%d" % (year, q), default=[])


def fetch_change(year, q):
    """/js/other/zcbd/{y}_{q} -> 重仓个股与往季变动，按覆盖面变化降序"""
    return _get("/js/other/zcbd/%d_%d" % (year, q), default=[])


def pick_report(y, q, max_back=8, fetcher=None):
    """查询未披露的报告期会返回空；向前回溯到最近一个已披露报告期"""
    fetcher = fetcher or fetch_change
    for _ in range(max_back):
        data = fetcher(y, q)
        if data and not (isinstance(data, dict) and "_error" in data):
            return (y, q, data)
        q -= 1
        if q == 0:
            y -= 1
            q = 4
    return (y, q, [])


def norm_change(rows):
    """变动解读：bhfgm>0 覆盖面扩大，cgsbh>0 加仓"""
    out = []
    for r in rows or []:
        out.append({
            "代码":         _hit_key(r, "dm", default="-"),
            "名称":         _hit_key(r, "mc", default="-"),
            "上季覆盖面":   _to_float(_hit_key(r, "qfgm")),
            "本季覆盖面":   _to_float(_hit_key(r, "bfgm")),
            "覆盖面变化":   _to_float(_hit_key(r, "bhfgm")),
            "上季持股万股": _to_float(_hit_key(r, "qcgs")),
            "本季持股万股": _to_float(_hit_key(r, "bcgs")),
            "持股数变化":   _to_float(_hit_key(r, "cgsbh")),
            "流通占比变化": _to_float(_hit_key(r, "ltbh")),
            "报告期":       _hit_key(r, "yq", default="-"),
        })
    return out


# ---------- 5. 链路 C：场内 ETF 实时 ----------
def fetch_etf_realtime(code):
    return _get("/fund/real/ssjy/%s" % code, default={})


def norm_realtime(d):
    return {
        "最新价":   _to_float(_hit_key(d, "p")),
        "涨跌幅":   _to_float(_hit_key(d, "pc")),
        "成交额":   _to_float(_hit_key(d, "cje")),
        "换手率":   _to_float(_hit_key(d, "tr")),
        "更新时间": _hit_key(d, "t", default="-"),
    }


# ---------- 6. 校验 ----------
def run_check():
    # 1) 字段容错
    assert _hit_key({"Dm": "000001"}, "dm") == "000001"
    assert _to_float("-") is None and _to_float("9.8") == 9.8

    # 2) 多期持仓只取最新一期
    rows = [
        {"jd": "2026年二季报", "t": "2026-06-30", "dm": "600519", "mc": "A",
         "jzbl": "9.8", "cgs": "100", "ccsz": "150000"},
        {"jd": "2026年一季报", "t": "2026-03-31", "dm": "600519", "mc": "A",
         "jzbl": "8.1", "cgs": "90", "ccsz": "130000"},
        {"jd": "2026年二季报", "t": "2026-06-30", "dm": "000858", "mc": "B",
         "jzbl": "5.2", "cgs": "200", "ccsz": "80000"},
    ]
    lp = latest_period(rows)
    assert len(lp) == 2
    assert all(_hit_key(r, "t") == "2026-06-30" for r in lp)

    # 3) 穿透聚合：同基金同期的重复行应被去掉；按基金数再按市值排序
    funds = {
        "F001": [{"dm": "600519", "mc": "A", "cgs": "100", "ccsz": "150000"},
                 {"dm": "600519", "mc": "A", "cgs": "100", "ccsz": "150000"},  # 重复行
                 {"dm": "000858", "mc": "B", "cgs": "200", "ccsz": "80000"}],
        "F002": [{"dm": "600519", "mc": "A", "cgs": "50", "ccsz": "75000"},
                 {"dm": "601318", "mc": "C", "cgs": "300", "ccsz": "60000"}],
        "F003": [{"dm": "000858", "mc": "B", "cgs": "80", "ccsz": "32000"}],
    }
    pen = penetrate(funds, topn=10)
    by = {x["代码"]: x for x in pen}
    assert by["600519"]["基金数"] == 2                     # 重复行已去重
    assert by["600519"]["合计市值万元"] == 225000.0
    assert pen[0]["代码"] == "600519"                      # 基金数 2 且市值最大
    assert by["000858"]["基金数"] == 2
    assert by["601318"]["基金数"] == 1

    # 4) 报告期回溯：查询未披露季度时自动退到上一个已披露报告期
    def fake_fetch(y, q):
        return [] if (y, q) == (2026, 4) else [{"dm": "x", "yq": "%d_%d" % (y, q)}]
    y, q, data = pick_report(2026, 4, fetcher=fake_fetch)
    assert (y, q) == (2026, 3) and len(data) == 1

    # 5) 变动解读：覆盖面与持股数变化的符号
    chg = norm_change([{"dm": "600519", "mc": "A", "qfgm": "100", "bfgm": "130",
                        "bhfgm": "30", "qcgs": "1000", "bcgs": "1500",
                        "cgsbh": "500", "ltbh": "0.5", "y": "2026", "q": "2",
                        "yq": "2026_2"}])
    assert chg[0]["覆盖面变化"] == 30.0
    assert chg[0]["持股数变化"] == 500.0
    assert chg[0]["报告期"] == "2026_2"

    # 6) 实时归一
    rt = norm_realtime({"p": "1.234", "pc": "0.82", "cje": "1234567",
                        "tr": "3.1", "t": "2026-08-28 15:00:00"})
    assert rt["最新价"] == 1.234 and rt["涨跌幅"] == 0.82
    assert rt["更新时间"].startswith("2026-08-28")

    print("校验通过")


if __name__ == "__main__":
    run_check()
    print("-" * 66)
    for name, path in [("所有基金列表", "/jh/list/all"),
                       ("ETF基金列表", "/jh/hq/etflist"),
                       ("基金概况", "/jh/base/jjgk/000001"),
                       ("股票持仓", "/jh/zh/gpcc/000001"),
                       ("债券持仓", "/jh/zh/zccc/000001"),
                       ("基金重仓股", "/js/other/jjzc/2026_2"),
                       ("重仓股变动", "/js/other/zcbd/2026_2"),
                       ("ETF实时行情", "/fund/real/ssjy/159001")]:
        data = _get(path, default=[])
        if isinstance(data, dict) and "_error" in data:
            print("%-12s %-26s -> %s" % (name, path, data["_error"][:60]))
        else:
            print("%-12s %-26s -> %d 条" % (name, path, len(data)))

5. 跑通示例

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

6. 坑与注意事项

坑 1：持仓是季报数据，不是实时。
/jh/zh/gpcc/{code} 与 /jh/zh/zccc/{code} 的更新频率是每周六 13:00，内容来自基金季报。你看到的"某基金持有某股票"，实际是最近一个报告期的快照，可能已经过去两三个月。别拿它做当日决策。

坑 2：/jh/list/all 里没有 ETF 和 LOF。
它的 tp 字段只有三个值：1 开放式、2 封闭式、3 分级子基金。要拿 ETF 与 LOF 代码，得单独调 /jh/hq/etflist 与 /jh/hq/loflist。这不是同一个列表的三种过滤，是几份独立的清单。

坑 3：报告期没披露时返回空，不是报错。
/js/other/jjzc/{y}_{q} 与 /js/other/zcbd/{y}_{q} 查一个还没披露的季度，返回的是空列表，HTTP 状态码仍然是 200。如果代码只判断"没有异常"，就会拿着空结果往下走。pick_report 的做法是：检测到空就往前退一个季度，最多退 8 期。

坑 4：穿透聚合要去重，否则会重复计数。
同一只基金、同一个报告期，接口可能返回多行相同股票代码（不同持仓类别）。penetrate 里用 seen 集合按 (基金, 股票代码) 去重，否则"被 N 只基金持有"这个数字会虚高。

坑 5：/fund/list/all 与 /jh/list/all 是两套代码体系。
前者是沪深场内基金，后者涵盖开放式/封闭式/分级子基金。取代码前先确认你要查的是哪种基金，传错体系的码会查不到。

7. 小结与下篇预告

本篇把基金板块 32 个端点分成 /jh（档案）、/js（排名）、/fund（场内实时）三组，给出三条链路：穿透聚合（latest_period + penetrate）、官方重仓与变动（pick_report + norm_change）、场内实时（norm_realtime）。

下一篇计划写 #03《龙虎榜与机构席位追踪：9个接口看穿游资与机构的买卖动向》：用 /hilh（龙虎榜 5 个）与 /hijg（机构持仓 4 个）两组接口，从每日榜单明细追到营业部与机构席位的历史行为。

8. 免责声明

本文仅演示基金持仓数据的取数与穿透聚合方法，所有代码示例均为演示数据，未含任何真实持仓数值，不构成投资建议，亦不承诺收益。

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

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

把代码里的 你的智兔token 换成你拿到的真实 token，上面的脚本就能直接打印基金持仓与重仓股变动数据。