系列：《Python 量化取数指南》｜连载项目 · 纯 GET 取数 · 仅依赖 requests
数据：由智兔数服提供。更多接口见 智兔数服技术博客。

1. 你将得到什么

• 一个多源取数层封装：注册多个来源，按优先级自动尝试，一家挂了切下一家
• 一个「实时快照 + 历史 K 线」双源兜底的完整示例
• 一个离线 run_check()，模拟主源失败自动降级

2. 本篇取数约定

• 用第 3 篇的 _get 做单源请求；本篇在其上包一层「源注册表 + 顺序兜底」
• 同一数据可注册多个等价端点（如快照类 vs 历史类互为兜底），或不同证书/节点
• 请求：GET https://api.zhituapi.com<path>?token=<你的智兔token>
• 兜底原则：成功即返回，全部失败才报错；带超时与重试避免单源拖死

3. 核心模板（全系列复用 + 灾备层）

import time, json, requests

BASE = "https://api.zhituapi.com"
TOKEN = "你的智兔token"      # 演示证书（免费版）即可起步

def _get(path, params=None, timeout=15, retry=3, backoff=1.5):
    params = dict(params or {})
    params["token"] = TOKEN
    url = BASE + path
    last = None
    for i in range(retry):
        try:
            r = requests.get(url, params=params, timeout=timeout)
            if r.status_code != 200:
                last = f"HTTP {r.status_code} {r.text[:120]}"
                time.sleep(backoff * (i + 1)); continue
            try:
                return r.json(), None
            except ValueError:
                last = f"非JSON响应: {r.text[:120]}"
                return None, last
        except requests.RequestException as e:
            last = str(e); time.sleep(backoff * (i + 1))
    return None, last

def _hit_key(d, *keys, default=None):
    if not isinstance(d, dict):
        return default
    for k in keys:
        if k in d and d[k] not in (None, "", []):
            return d[k]
    return default

def _to_float(x, default=float("nan")):
    try:
        return float(x)
    except (TypeError, ValueError):
        return default

4. 跑通示例：多源兜底取数

def fetch_with_fallback(sources):
    for path, parser in sources:
        data, err = _get(path, timeout=8, retry=2)
        if err:
            print(f"  源 {path} 不可用: {err}")
            continue
        return parser(data), None
    return None, "所有源均失败"

def demo_failover():
    sources = [
        ("/hz/history/fsjy/000001.SH/d", lambda d: (d if isinstance(d,list) else d.get("data") or [])),
        ("/hz/list/hszs", lambda d: (d if isinstance(d,list) else d.get("data") or [])),
    ]
    rows, err = fetch_with_fallback(sources)
    if err:
        print("全部失败:", err)
    else:
        print(f"  兜底成功，取到 {len(rows)} 条")

def run_check():
    called = []
    def fake_get(path, **kw):
        called.append(path)
        if path == "/hz/history/fsjy/000001.SH/d":
            return None, "timeout"
        return [{"code":"000001.SH","name":"上证"}], None
    global _get
    _orig = _get
    _get = fake_get
    try:
        rows, err = fetch_with_fallback([("/hz/history/fsjy/000001.SH/d", lambda d:d),
                                         ("/hz/list/hszs", lambda d:d)])
        print(f"  [run_check] 主源失败后降级，取到 {len(rows)} 条，调用顺序 {called}")
    finally:
        _get = _orig

if __name__ == "__main__":
    demo_failover()
    run_check()

返回字段说明：fetch_with_fallback 返回 (parsed, None) 或 (None, "所有源均失败")；sources 是 (path, parser) 元组列表，按注册顺序尝试。生产环境可把不同证书/节点/供应商分别注册为一项。

5. 坑与注意事项

1. 语义等价才能兜底：两个源返回结构可能不同，parser 要各自对齐，别直接混用。
2. 缩短单源超时：兜底层单源 timeout 要短（如 8s），否则一家慢会拖垮整体。
3. 别无限重试：单源 retry 调小（如 2），把「快速失败」交给下一源。
4. 记录谁挂了：生产要打日志/告警，别静默切换导致一直用慢源。
5. 频率预算：兜底源也可能被同一证书限流，切换后注意总调用量。
6. 字段名三套：解析仍用 _hit_key 兜字段名差异。

6. 常见报错速查

7. 小结与下一篇预告

小结：灾备 = 「源注册表 + 按序尝试 + 成功即返 + 全部失败才报错」；关键是语义等价 parser、短超时、快速失败、打告警。

下一篇计划写 #16《量化数据源选型总览与避坑清单》：汇总本系列 8 类数据的选型矩阵与避坑清单，方便你一次性对照。

8. 免责声明

本文仅演示公开数据接口的用法，所有代码示例均为演示数据，不构成任何投资建议；实际返回字段以接口文档与你的证书权限为准。数据由 智兔数服 提供，更多接口示例见 技术博客。

免费领取证书 / 查看完整接口文档，可前往 智兔数服官网。