欢迎光临
我们一直在努力

【Python 量化取数指南 #01】别再每次重写 requests:一个取数封装全系列直接复用

【Python 量化取数指南 #01】别再每次重写 requests:一个取数封装,全系列直接复用

系列:《Python 量化取数指南》|连载项目 · 纯 GET 取数 · 仅依赖 requests
适用:想用 Python 稳定拉取股票 / 基金 / 可转债等公开行情数据,又不想每次重写 requests 的读者。本篇把「统一请求封装 + 字段容错」这一底座钉死,后续每篇都在此之上叠加具体接口,你抄过去填 token 就能跑。

1. 你将得到什么

读完这一篇,你能拿走四样东西:

  • 一个统一请求封装 _get(path):token、超时、重试、退避一次封好,后面 16 篇所有接口都调它,不再各处散写;
  • 一个字段容错工具 _hit_key:官方接口字段名中英文 / 大小写混用也不崩,按顺序命中第一个非空值;
  • 一个类型归一工具 _to_float:空串、-、null 等脏值统一返回 None,不猜不填 0;
  • 一份 run_check() 合成数据校验 + 一个冒烟端点 /hz/list/hszs:复制即跑,逻辑先过校验,再填 token 拉真实数据。
  • 代码全部自包含,只依赖 requests,复制进 .py 直接能跑,不依赖 numpy / pandas。

    2. 本篇取数约定

    • 全部接口都是 GET + query 参数,token 放在查询串里(?token=xxx);
    • 统一基址 https://api.zhituapi.com;
    • 代码块里的 你的token 是占位符,换成你的真实 token 即可;
    • 代码 / 日期 / 阶段等参数在路径里(如 /hz/list/{code}),只有 token 是查询参数;
    • 所有接口路径均取自官方文档。

    3. 环境准备(2 分钟)

    只需要一个标准库之外的依赖:

    pip install requests

    要求:

    • Python 3.8 及以上(3.7 也基本可用,但建议升到 3.8+)
    • 能正常访问外网(接口是 HTTPS GET)

    验证装好:

    import requests
    print(requests.__version__) # 输出版本号即正常

    4. 核心模板函数

    import requests, time

    BASE = "https://api.zhituapi.com"
    TOKEN = "你的token" # 占位,换成你申请的真实 token

    # ———- 1. 字段容错:中英文 / 大小写混用也能命中 ———-
    def _hit_key(d, *cands, default=None):
    if not isinstance(d, dict):
    return default
    for c in cands:
    if c in d and d[c] not in (None, "", "-", "null"):
    return d[c]
    low = {str(k).lower(): v for k, v in d.items()}
    for c in cands:
    v = low.get(str(c).lower())
    if v not in (None, "", "-", "null"):
    return v
    return default

    # ———- 2. 类型归一:脏值返回 None,不猜不填 0 ———-
    def _to_float(v, default=None):
    try:
    if v in (None, "", "-", "null", "None"):
    return default
    return float(v)
    except (TypeError, ValueError):
    return default

    # ———- 3. 统一请求:重试 + 退避 ———-
    def _get(path, params=None, timeout=10, retries=2, backoff=0.6):
    """返回 (data, err);成功时 err 为 None,失败时 data 为 None。"""
    q = {"token": TOKEN}
    if params:
    q.update(params)
    last = ""
    for i in range(retries + 1):
    try:
    r = requests.get(BASE + path, params=q, timeout=timeout)
    if r.status_code == 200:
    try:
    return r.json(), None
    except ValueError:
    return None, "非 JSON:%s" % (r.text.strip()[:140])
    last = "HTTP %s %s" % (r.status_code, (r.text or "").strip()[:140])
    except requests.RequestException as e:
    last = "%s: %s" % (type(e).__name__, e)
    if i < retries:
    time.sleep(backoff * (i + 1))
    return None, last

    # ———- 4. 校验:合成数据,不联网 ———-
    def run_check():
    # 1) 字段容错:大小写 / 中英文混用都能命中
    assert _hit_key({"Dm": "000001", "mc": "平安银行"}, "dm", "code") == "000001"
    assert _hit_key({"code": "600519"}, "dm", "code") == "600519"
    # 2) 类型归一:空串 / 异常值返回 None
    assert _to_float("-") is None and _to_float("12.5") == 12.5
    assert _to_float(None) is None
    # 3) 请求封装:返回形态固定为 (data, err)
    data, err = _get("/hz/list/hszs")
    assert isinstance(err, (type(None), str)) # 联网态取决于 token,仅校验形态
    print("校验通过")

    if __name__ == "__main__":
    run_check() # 合成数据逻辑校验,无需联网
    # 填入真实 token 后取消下一行注释,即可拉取真实指数代码列表:
    # data, err = _get("/hz/list/hszs")
    # print(data if err is None else err)

    5. 跑通示例

    默认 python 本文件.py 只跑 run_check(),会打印 校验通过(不联网、不依赖 token)。

    把 TOKEN 换成你的真实 token,并取消 if __name__ == "__main__": 里那行注释后运行,会请求 /hz/list/hszs 并打印一串指数代码列表,返回结构类似:

    {'code': 0, 'data': [{'code': '000001', 'name': '上证指数', 'market': 'SH'},
    {'code': '399001', 'name': '深证成指', 'market': 'SZ'}, …],
    'msg': 'ok'}

    (上例为真实返回结构示意,字段名以接口实际返回为准;填对 token 即为真实数据。调用方统一用 data, err = _get(path),先判 if err is None 再消费 data,别把 try/except 散落在业务代码里。)

    6. 坑与注意事项

    坑 1:token 不要写进公开代码 / 版本库。 本地用环境变量或配置文件读取,别硬编码后推到 GitHub。

    坑 2:免费额度有频率与字段限制。 能起步,但生产环境先评估是否够用,别一上来全量依赖。

    坑 3:不同接口返回结构不一。 动手解析前,先 print(r.status_code, r.text[:500]) 看原始响应(_get 失败时会把原始响应截前 140 字塞进 err),确认是对象还是数组、字段名到底叫什么。

    坑 4:别在循环里无间隔高频请求。 容易触发 429 限流,批量拉取时加 time.sleep 或并发上限。

    坑 5:重试要有限度。 封装里的 retries=2 是兜底,不是让你无限重试;持续失败多半是 token 或权限问题,先查原因,别一味加重试。

    坑 6:路径参数 ≠ 查询参数。 代码 / 日期 / 阶段在路径里(如 /hz/list/{code}),只有 token 走查询串;拼错位置会返回 404。

    常见报错速查表:

    报错 / 现象原因处理
    HTTP 401 / err 含 token 字样 token 错或没填 检查 TOKEN 是否被占位字符串覆盖
    HTTP 403 频率或权限限制 降低并发,确认额度类型支持该接口
    HTTP 429 触发限流 调大 backoff,减少重试频次,必要时降速
    requests.exceptions.Timeout 网络慢 / 超时短 调大 timeout(如 20)
    返回 非 JSON:… 网关拦截 / 路径错 print 原始响应看状态码与文本
    中文乱码 编码问题 r.encoding = r.apparent_encoding 或 r.json() 一般已处理

    7. 小结与下一篇预告

    本篇把全系列共用的底座钉死:_get(path)(token + 超时 + 重试 + 退避,返回 (data, err))、_hit_key(字段容错)、_to_float(类型归一),外加一份合成数据 run_check()。后续每篇都在此基础上叠加具体接口,你只需关注每篇新增的部分。

    下一篇计划写 #02《Python 免费拿股票数据:6 个接口实测,0 元能拉到什么》:用本篇的 _get 拉「免费可拿」的一批接口(实时行情、板块列表、指数代码列表、可转债列表),给出可跑示例与返回字段说明。

    8. 免责声明

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

    赞(0)
    未经允许不得转载:171主机测评 » 【Python 量化取数指南 #01】别再每次重写 requests:一个取数封装全系列直接复用
    分享到: 更多 (0)

    评论 抢沙发

    • 昵称 (必填)
    • 邮箱 (必填)
    • 网址