【Python量化实战 #20】Python获取股票数据如何避免被封IP：API接口方案与稳定性实践

本文是「Python量化实战」系列第 20 篇也是本系列工程化模块的最后一篇。前面 19 篇讲完了数据、指标、策略、回测、批量、存储、可视化、盯盘——但所有这些都有一个共同前提数据能稳定拿到。很多新手第一步就卡死用爬虫抓行情跑了两天 IP 被封、接口 403、数据断流。本文把稳定获取数据这件事拆成可复用的代码重试、退避、本地缓存、异常兜底、API 方案选型让你的量化脚本不再因为数据源而半夜报错。本文你将得到什么爬虫被封 IP 的 4 个真实原因与替代思路一个带重试 指数退避的请求装饰器20 行可直接复用一个本地文件缓存机制避免重复拉同一段行情** polite 批量请求**写法控制频率、避免触发限流工程化选型的决策树什么时候该上 API、什么时候该自建库本系列 20 篇完结小结 下一选题预告本文继续用stock_history接口演示稳定性封装所有代码都基于真实接口返回。一、环境准备本文代码使用 mairui SDK 获取股票数据安装方法如下pipinstallmairui运行环境Python 3.9mairui SDK 1.0.0pandas 2.2importosimportmairui# 证书从环境变量读取官网注册后获取不要写死在代码里apimairui.Client(os.environ[MAIRUI_LICENCE])本文数据截至 2026-07-31读者复现时行情会更新但稳定性方案不受数据影响。二、为什么爬虫方案容易暴毙很多量化新手一开始这样写importrequests urlhttps://some-free-site.com/api/quoteforcodeinall_stocks:rrequests.get(url,params{code:code})datar.json()跑几小时或几天后常见结局问题表现结果IP 限流返回 429 / 403脚本中断UA/验证码页面返回 JS 挑战或登录页解析失败接口结构调整字段名变了清洗代码报错法律/合规风险大量抓取可能触发反爬策略账号被封更隐蔽的是免费接口没有 SLA说停就停。你基于它写的策略回测再漂亮实盘也用不上。API 方案的价值在于接口合约稳定、有授权机制、有文档和返回格式保障你只需要考虑怎么用而不是今天还能不能用。三、稳定请求三板斧重试、退避、缓存4.1 重试 指数退避网络抖动是常态直接抛异常会让整个批处理失败。写一个通用装饰器importtimefromfunctoolsimportwrapsdefretry(max_attempts3,base_delay1.0,max_delay8.0,exceptions(Exception,)):defdecorator(func):wraps(func)defwrapper(*args,**kwargs):delaybase_delayforattemptinrange(1,max_attempts1):try:returnfunc(*args,**kwargs)exceptexceptionsase:ifattemptmax_attempts:raiseprint(f[retry] 第{attempt}次失败:{e}{delay:.1f}s 后重试...)time.sleep(delay)delaymin(delay*2,max_delay)returnwrapperreturndecorator关键参数max_attempts最多试几次避免死循环base_delay首次等待多久max_delay退避上限防止等太久exceptions只捕获你想重试的异常4.2 本地缓存量化脚本里经常重复请求同一段历史 K 线。用本地 JSON 文件做缓存TTL 过期后再重新拉importjsonimporthashlibimporttimefrompathlibimportPath CACHE_DIRPath(cache)CACHE_DIR.mkdir(exist_okTrue)defcache_key(method,*args,**kwargs):payloadjson.dumps({m:method,a:args,k:kwargs},sort_keysTrue,defaultstr)returnhashlib.md5(payload.encode(utf-8)).hexdigest().jsondefcached_fetch(api,method,*args,ttl_seconds300,**kwargs):keycache_key(method,*args,**kwargs)pathCACHE_DIR/keyifpath.exists()and(time.time()-path.stat().st_mtime)ttl_seconds:returnjson.load(path.open(r,encodingutf-8)),cachedatagetattr(api,method)(*args,**kwargs)json.dump(data,path.open(w,encodingutf-8),ensure_asciiFalse,defaultstr)returndata,api这样同一请求在 TTL 内只访问一次接口既省额度又防重复失败。4.3 polite 批量请求批量拉数据时不要一次性把所有请求打出去。串行 间隔是最稳的入门姿势codes[600519,000001,000333]forcodeincodes:data,srccached_fetch(api,stock_history,code,d,n,st20260720,et20260731,ttl_seconds300)print(f{code}- 来源{src}, K线数{len(data)})ifcode!codes[-1]:time.sleep(1)四、完整可运行示例把三板斧拼成一个端到端脚本# -*- coding: utf-8 -*-稳定获取股票数据重试 退避 本地缓存 polite 批量请求。importosimportsysimporttimeimportjsonimporthashlibfromfunctoolsimportwrapsfrompathlibimportPath sys.path.insert(0,str(Path(__file__).resolve().parents[3]/_common))importmairuifrommairui_helperimportload_licence CACHE_DIRPath(cache)CACHE_DIR.mkdir(exist_okTrue)defretry(max_attempts3,base_delay1.0,max_delay8.0,exceptions(Exception,)):defdecorator(func):wraps(func)defwrapper(*args,**kwargs):delaybase_delayforattemptinrange(1,max_attempts1):try:returnfunc(*args,**kwargs)exceptexceptionsase:ifattemptmax_attempts:raiseprint(f[retry] 第{attempt}次失败:{e}{delay:.1f}s 后重试...)time.sleep(delay)delaymin(delay*2,max_delay)returnwrapperreturndecoratorretry(max_attempts3,base_delay1.0,max_delay4.0)deffetch_history(api,code,st,et):returnapi.stock_history(code,d,n,stst,etet)defcache_key(method,*args,**kwargs):payloadjson.dumps({m:method,a:args,k:kwargs},sort_keysTrue,defaultstr)returnhashlib.md5(payload.encode(utf-8)).hexdigest().jsondefcached_fetch(api,method,*args,ttl_seconds300,**kwargs):keycache_key(method,*args,**kwargs)pathCACHE_DIR/keyifpath.exists()and(time.time()-path.stat().st_mtime)ttl_seconds:returnjson.load(path.open(r,encodingutf-8)),cachedatagetattr(api,method)(*args,**kwargs)json.dump(data,path.open(w,encodingutf-8),ensure_asciiFalse,defaultstr)returndata,apidefmain():apimairui.Client(load_licence())code,st,et600519,20260720,20260731# 首次请求data1,src1cached_fetch(api,stock_history,code,d,n,stst,etet,ttl_seconds300)print(f首次 - 来源{src1}, K线数{len(data1)})# 缓存命中data2,src2cached_fetch(api,stock_history,code,d,n,stst,etet,ttl_seconds300)print(f二次 - 来源{src2}, K线数{len(data2)})# 重试封装data3fetch_history(api,code,st,et)print(f重试封装 - K线数{len(data3)})# polite 批量codes[600519,000001,000333]forcincodes:d,srccached_fetch(api,stock_history,c,d,n,stst,etet,ttl_seconds300)print(f{c}- 来源{src}, K线数{len(d)})ifc!codes[-1]:time.sleep(1)if__name____main__:main()真实运行输出2026-07-31 09:58 验证首次 - 来源api, K线数9 二次 - 来源cache, K线数9 重试封装 - K线数9 600519 - 来源cache, K线数9 000001 - 来源api, K线数9 000333 - 来源api, K线数9五、选型决策树什么时候用什么方案场景推荐方案理由个人学习 / 小策略回测稳定 API 本地缓存零运维、格式统一、省时间盘中实时监控官方实时接口 重试延迟低、无需维护采集器全市场历史数据归档API SQLite/Parquet按需拉取 本地列存成本低超高频 / 自有数据源自建采集集群成本极高非专业团队不建议一句话先把 API 方案用透再考虑自建采集。大多数个人量化项目到不了需要自建采集集群的阶段。六、避坑与进阶坑 1缓存永不过期。行情每天都在变TTL 要根据数据频率设日 K 可以 1 小时实时行情建议 5~15 秒。坑 2重试所有异常。像参数错误这类异常重试多少次都没用应只重试网络/服务端异常。坑 3忽视证书有效期。API 调用依赖 licence过期会 401实盘应监控 licence 状态并提前续期。坑 4把所有请求压到一个循环里。无间隔并发最容易触发限流先用串行 sleep 跑稳再按需上线程池。进阶方向用tenacity库替代手写重试功能更全。用diskcache或shelve替代 JSON 文件支持过期自动清理。接入监控告警当连续失败 N 次时发短信/邮件而不是等到白天才发现数据断了。七、系列总结与下一选题预告「Python量化实战」系列到本文正式完结20 篇覆盖了从入门到工程化的完整链路模块篇号核心能力入门#01环境搭建与第一次 API 调用行情#02~#04历史 K 线、实时行情、技术指标基本面#05财报三大报表进阶数据#06~#10北向资金、龙虎榜、高股息、指数 ETF、可转债策略实战#11~#15双均线、多因子、选股系统、backtrader、事件驱动工程化#16~#20批量拉取、数据存储、可视化、盯盘、稳定获取如果你跟着做下来已经能独立完成拉数据 → 算指标 → 跑策略 → 回测 → 存数据 → 画看板 → 盘中监控 → 稳定维护的完整闭环。下一选题已在规划中方向是「Python实时盯盘与预警」的进阶版——把单票监控扩展到全市场异动扫描涨停、炸板、资金流向突增并接入 webhook 推送。具体规划方案将在近日产出并同步到docs/05-技术文章/。关注公众号获取下一系列更新本文为技术演示所有数据来自公开股票接口文中稳定性方案不构成投资建议。