免费领取证书

← 返回博客列表

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

2026年08月05日 09:35 · 内容龙虾

本文是「Python量化实战」系列第 20 篇，也是本系列工程化模块的最后一篇。前面 19 篇讲完了数据、指标、策略、回测、批量、存储、可视化、盯盘——但所有这些都有一个共同前提：数据能稳定拿到。很多新手第一步就卡死：用爬虫抓行情，跑了两天 IP 被封、接口 403、数据断流。本文把"稳定获取数据"这件事拆成可复用的代码：重试、退避、本地缓存、异常兜底、API 方案选型，让你的量化脚本不再因为数据源而半夜报错。

本文你将得到什么

  1. 爬虫被封 IP 的 4 个真实原因与替代思路
  2. 一个带重试 + 指数退避的请求装饰器（20 行，可直接复用）
  3. 一个本地文件缓存机制，避免重复拉同一段行情
  4. ** polite 批量请求**写法：控制频率、避免触发限流
  5. 工程化选型的决策树：什么时候该上 API、什么时候该自建库
  6. 本系列 20 篇完结小结 + 下一选题预告

一、在线体验

想先在线试试接口效果？打开 API Playground 即可直接调用测试： https://mairuiapi.com/playground

本文继续用 stock_history 接口演示稳定性封装，所有代码都基于真实接口返回。

二、环境准备

本文代码使用 mairui SDK 获取股票数据，安装方法如下：

pip install mairui

SDK 的完整接口文档与使用说明请查阅 GitHub 仓库：https://github.com/MaiRuiApi/mairui

接口的详细参数说明请查阅官网 API 文档：https://mairuiapi.com/hsdata

运行环境： - Python 3.9+ - mairui SDK 1.0.0 - pandas 2.2+

import os
import mairui

# 证书从环境变量读取（官网注册后获取），不要写死在代码里
api = mairui.Client(os.environ["MAIRUI_LICENCE"])

本文数据截至 2026-07-31，读者复现时行情会更新，但稳定性方案不受数据影响。

三、为什么爬虫方案容易"暴毙"

很多量化新手一开始这样写：

import requests
url = "https://some-free-site.com/api/quote"
for code in all_stocks:
    r = requests.get(url, params={"code": code})
    data = r.json()

跑几小时或几天后常见结局：

问题 表现 结果
IP 限流 返回 429 / 403 脚本中断
UA/验证码 页面返回 JS 挑战或登录页 解析失败
接口结构调整 字段名变了 清洗代码报错
法律/合规风险 大量抓取可能触发反爬策略 账号被封

更隐蔽的是：免费接口没有 SLA，说停就停。你基于它写的策略回测再漂亮，实盘也用不上。

API 方案的价值在于：接口合约稳定、有授权机制、有文档和返回格式保障，你只需要考虑"怎么用"，而不是"今天还能不能用"。

四、稳定请求三板斧：重试、退避、缓存

4.1 重试 + 指数退避

网络抖动是常态，直接抛异常会让整个批处理失败。写一个通用装饰器：

import time
from functools import wraps


def retry(max_attempts=3, base_delay=1.0, max_delay=8.0, exceptions=(Exception,)):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            delay = base_delay
            for attempt in range(1, max_attempts + 1):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    if attempt == max_attempts:
                        raise
                    print(f"[retry] 第 {attempt} 次失败: {e}，{delay:.1f}s 后重试...")
                    time.sleep(delay)
                    delay = min(delay * 2, max_delay)
        return wrapper
    return decorator

关键参数： - max_attempts：最多试几次，避免死循环 - base_delay：首次等待多久 - max_delay：退避上限，防止等太久 - exceptions：只捕获你想重试的异常

4.2 本地缓存

量化脚本里经常重复请求同一段历史 K 线。用本地 JSON 文件做缓存，TTL 过期后再重新拉：

import json
import hashlib
import time
from pathlib import Path

CACHE_DIR = Path("cache")
CACHE_DIR.mkdir(exist_ok=True)


def cache_key(method, *args, **kwargs):
    payload = json.dumps({"m": method, "a": args, "k": kwargs}, sort_keys=True, default=str)
    return hashlib.md5(payload.encode("utf-8")).hexdigest() + ".json"


def cached_fetch(api, method, *args, ttl_seconds=300, **kwargs):
    key = cache_key(method, *args, **kwargs)
    path = CACHE_DIR / key

    if path.exists() and (time.time() - path.stat().st_mtime) < ttl_seconds:
        return json.load(path.open("r", encoding="utf-8")), "cache"

    data = getattr(api, method)(*args, **kwargs)
    json.dump(data, path.open("w", encoding="utf-8"), ensure_ascii=False, default=str)
    return data, "api"

这样同一请求在 TTL 内只访问一次接口，既省额度又防重复失败。

4.3 polite 批量请求

批量拉数据时，不要一次性把所有请求打出去。串行 + 间隔是最稳的入门姿势：

codes = ["600519", "000001", "000333"]
for code in codes:
    data, src = cached_fetch(api, "stock_history", code, "d", "n", st="20260720", et="20260731", ttl_seconds=300)
    print(f"{code} -> 来源={src}, K线数={len(data)}")
    if code != codes[-1]:
        time.sleep(1)

五、完整可运行示例

把三板斧拼成一个端到端脚本：

# -*- coding: utf-8 -*-
"""稳定获取股票数据：重试 + 退避 + 本地缓存 + polite 批量请求。"""
import os
import sys
import time
import json
import hashlib
from functools import wraps
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[3] / "_common"))
import mairui
from mairui_helper import load_licence

CACHE_DIR = Path("cache")
CACHE_DIR.mkdir(exist_ok=True)


def retry(max_attempts=3, base_delay=1.0, max_delay=8.0, exceptions=(Exception,)):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            delay = base_delay
            for attempt in range(1, max_attempts + 1):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    if attempt == max_attempts:
                        raise
                    print(f"[retry] 第 {attempt} 次失败: {e}，{delay:.1f}s 后重试...")
                    time.sleep(delay)
                    delay = min(delay * 2, max_delay)
        return wrapper
    return decorator


@retry(max_attempts=3, base_delay=1.0, max_delay=4.0)
def fetch_history(api, code, st, et):
    return api.stock_history(code, "d", "n", st=st, et=et)


def cache_key(method, *args, **kwargs):
    payload = json.dumps({"m": method, "a": args, "k": kwargs}, sort_keys=True, default=str)
    return hashlib.md5(payload.encode("utf-8")).hexdigest() + ".json"


def cached_fetch(api, method, *args, ttl_seconds=300, **kwargs):
    key = cache_key(method, *args, **kwargs)
    path = CACHE_DIR / key
    if path.exists() and (time.time() - path.stat().st_mtime) < ttl_seconds:
        return json.load(path.open("r", encoding="utf-8")), "cache"
    data = getattr(api, method)(*args, **kwargs)
    json.dump(data, path.open("w", encoding="utf-8"), ensure_ascii=False, default=str)
    return data, "api"


def main():
    api = mairui.Client(load_licence())
    code, st, et = "600519", "20260720", "20260731"

    # 首次请求
    data1, src1 = cached_fetch(api, "stock_history", code, "d", "n", st=st, et=et, ttl_seconds=300)
    print(f"首次 -> 来源={src1}, K线数={len(data1)}")

    # 缓存命中
    data2, src2 = cached_fetch(api, "stock_history", code, "d", "n", st=st, et=et, ttl_seconds=300)
    print(f"二次 -> 来源={src2}, K线数={len(data2)}")

    # 重试封装
    data3 = fetch_history(api, code, st, et)
    print(f"重试封装 -> K线数={len(data3)}")

    # polite 批量
    codes = ["600519", "000001", "000333"]
    for c in codes:
        d, src = cached_fetch(api, "stock_history", c, "d", "n", st=st, et=et, ttl_seconds=300)
        print(f"  {c} -> 来源={src}, K线数={len(d)}")
        if c != 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-技术文章/。

延伸阅读： - 在线体验更多接口：https://mairuiapi.com/playground - 查看完整 API 文档：https://mairuiapi.com/hsdata - SDK 文档与源码：https://github.com/MaiRuiApi/mairui - 关注公众号获取下一系列更新

本文为技术演示，所有数据来自公开股票接口，文中稳定性方案不构成投资建议。

相关阅读

QQ 客服 3826425416 咨询时请提供证书号或订单号，便于快速处理
咨询