【Python 量化取数指南 #15】Python 做数据源灾备一家挂了自动切系列《Python 量化取数指南》连载项目 · 纯 GET 取数 · 仅依赖 requests适用怕单一数据源抖动/限流/停服导致策略中断想做自动降级的人。1. 你将得到什么一个多源取数层封装注册多个来源按优先级自动尝试一家挂了切下一家一个「实时快照 历史 K 线」双源兜底的完整示例一个离线run_check()模拟主源失败自动降级2. 本篇取数约定用第 3 篇的_get做单源请求本篇在其上包一层「源注册表 顺序兜底」同一数据可注册多个等价端点如快照类 vs 历史类互为兜底或不同证书/节点请求GET https://api.zhituapi.compath?token你的token兜底原则成功即返回全部失败才报错带超时与重试避免单源拖死3. 核心模板全系列复用 灾备层importtime,json,requests BASEhttps://api.zhituapi.comTOKEN你的tokendef_get(path,paramsNone,timeout15,retry3,backoff1.5):paramsdict(paramsor{})params[token]TOKEN urlBASEpath lastNoneforiinrange(retry):try:rrequests.get(url,paramsparams,timeouttimeout)ifr.status_code!200:lastfHTTP{r.status_code}{r.text[:120]}time.sleep(backoff*(i1));continuetry:returnr.json(),NoneexceptValueError:lastf非JSON响应:{r.text[:120]}returnNone,lastexceptrequests.RequestExceptionase:laststr(e);time.sleep(backoff*(i1))returnNone,lastdef_hit_key(d,*keys,defaultNone):ifnotisinstance(d,dict):returndefaultforkinkeys:ifkindandd[k]notin(None,,[]):returnd[k]returndefaultdef_to_float(x,defaultfloat(nan)):try:returnfloat(x)except(TypeError,ValueError):returndefault4. 跑通示例多源兜底取数deffetch_with_fallback(sources):# sources: list[(path, parser)]按优先级尝试成功即返回forpath,parserinsources:data,err_get(path,timeout8,retry2)iferr:print(f 源{path}不可用:{err})continuereturnparser(data),NonereturnNone,所有源均失败defdemo_failover():# 同一「指数行情」注册两个等价来源做兜底sources[(/hz/history/fsjy/000001.SH/d,lambdad:(difisinstance(d,list)elsed.get(data)or[])),# 主历史K线(/hz/list/hszs,lambdad:(difisinstance(d,list)elsed.get(data)or[])),# 备指数列表]rows,errfetch_with_fallback(sources)iferr:print(全部失败:,err)else:print(f 兜底成功取到{len(rows)}条)defrun_check():# 模拟主源失败、备源成功called[]deffake_get(path,**kw):called.append(path)ifpath/hz/history/fsjy/000001.SH/d:returnNone,timeout# 主源挂return[{code:000001.SH,name:上证}],Noneglobal_get _orig_get _getfake_gettry:rows,errfetch_with_fallback([(/hz/history/fsjy/000001.SH/d,lambdad:d),(/hz/list/hszs,lambdad:d)])print(f [run_check] 主源失败后降级取到{len(rows)}条调用顺序{called})finally:_get_origif__name____main__:demo_failover()run_check()返回字段说明fetch_with_fallback返回(parsed, None)或(None, 所有源均失败)sources是(path, parser)元组列表按注册顺序尝试。生产环境可把不同证书/节点/供应商分别注册为一项。5. 坑与注意事项语义等价才能兜底两个源返回结构可能不同parser 要各自对齐别直接混用。缩短单源超时兜底层单源timeout要短如 8s否则一家慢会拖垮整体。别无限重试单源retry调小如 2把「快速失败」交给下一源。记录谁挂了生产要打日志/告警别静默切换导致一直用慢源。频率预算兜底源也可能被同一证书限流切换后注意总调用量。字段名三套解析仍用_hit_key兜字段名差异。6. 常见报错速查报错 / 现象原因处理一直用慢源主源半死不活缩短 timeout 告警解析错乱两源结构不同parser 各自对齐全失败都限流/挂降频 加第三源KeyError字段名不符print(data)看真实 key7. 小结与下一篇预告小结灾备 「源注册表 按序尝试 成功即返 全部失败才报错」关键是语义等价 parser、短超时、快速失败、打告警。下一篇计划写 #16《Python 量化取数避坑8 类数据接口选型总览》汇总本系列 8 类数据的选型矩阵与避坑清单方便你一次性对照。8. 免责声明本文仅演示公开数据接口的用法所有代码示例均为演示数据不构成任何投资建议实际返回字段以接口文档与你的证书权限为准。