系列：《量化系统从零构建》｜连载项目 · 纯 GET 取数 · 仅依赖 requests
适用：想把 #02 的客户端骨架补成「真能天天用」的多市场取数层，加上限频、重试、缓存与失败降级的读者；数据由智兔数服提供，在 #01/#02 的 _get 底座上继续装配，不依赖任何行情终端。

1. 你将得到什么

• 多市场统一客户端 ZhituClient（v2）：在 #02 骨架上加了四类健壮性——限频（避免被接口限流）、重试退避（抖动网络自愈）、本地缓存（重复请求不重复打接口）、失败降级（出错时返回默认值而非抛异常）。
• 分市场方法：A股日线、港股财务、北交所日线、基金K线、可转债清单、北向资金历史、龙虎榜，各封装成一个语义化方法，调用方不用再拼 URL。
• 可组合的取数契约：所有方法统一返回 (data, err)，出错走降级，业务代码不必到处 try/except。
• 本篇交付：后续 #04–#20 的落库 / 信号 / 回测全部基于这个客户端取数，本篇把它钉死。

2. 本篇用到的取数约定

GET https://api.zhituapi.com/<path>?token=你的智兔token

• 鉴权：token 走查询参数 ?token=，不要放进请求头。
• 错误形态：非 200 时接口先过鉴权再路由，常见 404 102:Licence证书(你的智兔token)不存在 —— 这是「证书（token）不存在」，不代表路径写错。
• 字段名不稳定：同一含义字段可能中英文混杂，继续用 _hit_key 候选键命中（见 #01）。数据来自 智兔数服（www.zhituapi.com）。

3. 健壮性四件套

4. 核心模板函数

import sys, time, json, requests

# ── 项目配置（与 #01 同源）──
BASE = "https://api.zhituapi.com"
TOKEN = "你的智兔token"  # 占位，换成你申请的真实 token


def _hit_key(d, keys):
    """候选键命中：字段名中英文混杂也不怕。"""
    if not isinstance(d, dict):
        return None
    for k in keys:
        if k in d and d[k] is not None:
            return d[k]
    low = {str(x).lower(): x for x in d.keys()}
    for k in keys:
        if k.lower() in low:
            return d[low[k.lower()]]
    return None


def _to_float(v):
    try:
        return None if v is None else float(v)
    except (TypeError, ValueError):
        return None


class ZhituClient:
    """多市场统一客户端（v2，带健壮性）。"""
    def __init__(self, token=TOKEN, rate_limit=0.2, retries=3, backoff=0.5, cache=None):
        self.token = token
        self.rate_limit = rate_limit      # 两次请求最小间隔（秒）
        self.retries = retries            # 失败重试次数
        self.backoff = backoff            # 重试退避基数（秒）
        self.cache = cache if cache is not None else {}
        self._last = 0.0

    def _throttle(self):
        now = time.time()
        gap = self.rate_limit - (now - self._last)
        if gap > 0:
            time.sleep(gap)
        self._last = time.time()

    def request(self, path, params=None, default=None):
        """统一取数：限频 + 重试退避 + 缓存 + 降级。返回 (data, err)。"""
        key = (path, json.dumps(params or {}, sort_keys=True))
        if key in self.cache:
            return self.cache[key], None
        self._throttle()
        last_err = None
        for i in range(1, self.retries + 1):
            p = dict(params or {})
            p["token"] = self.token
            try:
                r = requests.get(f"{BASE}{path}", params=p, timeout=10)
            except requests.RequestException as e:
                last_err = f"网络异常：{e}"
            else:
                if r.status_code == 200:
                    try:
                        data = r.json()
                    except ValueError:
                        last_err = f"非 JSON：{r.text.strip()[:140]}"
                    else:
                        self.cache[key] = data
                        return data, None
                else:
                    last_err = f"{r.status_code} {r.text.strip()[:140]}"
            if i < self.retries:
                time.sleep(self.backoff * i)
        return default, last_err

    # ── 分市场方法：把拼 URL 收口成语义调用 ──
    def hs_daily(self, code, default=None):
        """沪深A股日线。code 形如 000001.SZ。"""
        return self.request(f"/hs/history/d/{code}", default=default)

    def hk_fin(self, code, default=None):
        """港股财务三表之一（示例取盈利表）。code 为港股代码。"""
        return self.request(f"/hicw/yl/{code}", default=default)

    def bj_daily(self, code, default=None):
        """北交所日线。code 形如 830799.BJ。"""
        return self.request(f"/bj/history/d/{code}", default=default)

    def fund_kline(self, code, level="d", default=None):
        """基金历史K线。level 为 d/w/m 等周期。"""
        return self.request(f"/jh/hq/lskx/{code}/{level}", default=default)

    def kzz_list(self, default=None):
        """可转债清单。"""
        return self.request("/kzz/list", default=default)

    def hgt_north(self, stage=1, default=None):
        """北向资金历史。stage 为阶段序号（如 1）。"""
        return self.request(f"/ht/nbzj/bxls/{stage}", default=default)

    def lhb(self, sub, default=None):
        """龙虎榜（sub 由调用方填具体子路径，如 details/xxx）。"""
        return self.request(f"/hilh/{sub}", default=default)


def run_check():
    # 用假响应模拟 requests.get，验证缓存与降级（合成数据，非真实行情）
    class FakeResp:
        def __init__(self, status, text):
            self.status_code = status
            self.text = text
        def json(self):
            if self.status_code == 200:
                return {"hit": True}
            raise ValueError("bad")

    c = ZhituClient(token="T", retries=2, backoff=0)
    orig = requests.get
    try:
        # 1) 缓存命中：相同请求只打一次接口
        calls = {"n": 0}
        def counting(url, params=None, timeout=10):
            calls["n"] += 1
            return FakeResp(200, '{"hit": true}')
        requests.get = counting
        c.request("/hs/history/d/000001.SZ")
        c.request("/hs/history/d/000001.SZ")
        assert calls["n"] == 1, calls

        # 2) 失败降级：持续 500 -> 返回 default 与错误，不抛异常
        def always_fail(url, params=None, timeout=10):
            return FakeResp(500, "server error")
        requests.get = always_fail
        d, e = c.request("/x", default=[])
        assert d == [] and e is not None
    finally:
        requests.get = orig
    print("校验通过")


if __name__ == "__main__":
    if len(sys.argv) > 1 and sys.argv[1] == "--check":
        run_check()
    else:
        c = ZhituClient()
        # 填入你的真实 token 后即可拉取真实数据
        print("A股日线 /hs/history/d/000001.SZ ->", c.hs_daily("000001.SZ"))
        print("北向历史 /ht/nbzj/bxls/1 ->", c.hgt_north(1))

5. 跑通示例

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

6. 坑与注意事项

1. 限频间隔别设 0：连续高频易触发接口限流，rate_limit 给 0.1–0.3 秒更稳；批量回填时尤其要限频。
2. 缓存 key 要含 params：(path, params) 一起做 key，否则不同参数的请求会串台。
3. 降级值类型要对：default 应是「空结果」的同类型（如列表给 []、字典给 {}），下游才能无脑遍历。
4. 北向 / 南向在 /ht/nbzj 下：hgt_north 走 /ht/nbzj/bxls/{stage}，不是独立分组（见 #02 注册表）。
5. 龙虎榜子路径交给调用方：lhb(sub) 的 sub 由你填具体值，客户端只负责带 token / 限频 / 缓存。

7. 小结与下篇预告

本篇把 #02 的客户端骨架升级成「带限频、重试、缓存、降级」的 ZhituClient（v2），并把 7 类市场的取数收口成语义方法。这是后续所有落库与信号计算的唯一取数入口。

下一篇计划写 #04《存储设计：选型·建库·交易日历》：在取数层之上选型本地存储（SQLite 落库 + Parquet/DuckDB 做分析），建好数据库与交易日历表，为 #05 行情落库铺路。

8. 免责声明

本文仅演示公开数据接口的用法，所有代码示例均为演示数据，未含任何真实数据；文中示例数据仅作演示用途，不构成投资建议，亦不承诺收益。

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

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

把代码里的 你的智兔token 换成你拿到的真实 token，上面的脚本就能直接打印多市场行情与基本面数据。