系列：《Python量化实战全攻略》｜连载项目 · 纯 GET 取数 · 取数层仅依赖 requests
适用：想做择时 / 动量 / 回测，但被「日线从哪取」「周月线有没有独立接口」「复权因子去哪找」绕晕的读者；数据由智兔数服提供。本篇给沪深 A 股「日 K 线 / 分钟 K 线 / 技术指标（MACD·MA·BOLL·KDJ）/ 分红复权」9 个端点的分组地图、一次跑通的代码、pandas 多周期聚合与手工前复权，所有示例均为演示数据，不构成投资建议。

1. 你将得到什么

读完这一篇，你能拿走四样东西：

1. 一张分组地图：9 个历史端点按「日 K / 分钟 K / 技术指标 / 分红」分成 4 组；
2. 一次跑通的代码：/hs/history 拿日线，/hs/pro/history 拿分钟线，/hs/history/{macd|ma|boll|kdj} 直出指标；
3. 多周期聚合：日线 → 周线 / 月线用 pandas resample，不必找"周月线专用接口"（官方没有单列）；
4. 手工前复权：用 /hs/gs/jnff 分红数据自己算，不再等"复权因子"接口。

代码全部自包含，复制进 .py 直接能跑，取数层不依赖 numpy / pandas，分析层仅需 pandas。

2. 本篇取数约定

• 全部接口都是 GET + 路径参数，token 放在查询串里（?token=xxx）；
• 统一基址 https://api.zhituapi.com；
• 代码块里的 你的智兔token 是占位符，换成你的 token 即可；
• 所有接口路径均取自官方已验证文档，跨篇零重复（本篇专注历史序列，与第 1 篇盘中实时不重叠）。
• 数据来自 智兔数服（www.zhituapi.com）：零 SDK、纯 GET、免费版即可起步。

3. 9 个端点分 4 组

先建立地图。历史行情类一共 9 个端点，按用途分：

{code}.{m} 为带交易所后缀的代码，如 600519.SH、000001.SZ；n 为条数。周 / 月线没有独立端点：拿到日线后用 pandas resample 聚合即可，本文第 5 节给出代码。5/15/30/60 分钟线同理用 resample("5min") 等聚合分钟 K。

4. 核心模板函数

import requests, time

BASE = "https://api.zhituapi.com"
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


def _to_float(v, default=None):
    try:
        if v in (None, "", "-", "null", "None"):
            return default
        return float(v)
    except (TypeError, ValueError):
        return default


# ---------- 2. 统一请求：重试 + 退避 + 降级 ----------
def _get(path, timeout=10, retries=2, backoff=0.6, default=None):
    q = {"token": TOKEN}
    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()
                except ValueError:
                    return default
            last = "HTTP %s %s" % (r.status_code, (r.text or "").strip()[:80])
        except Exception as e:
            last = "%s: %s" % (type(e).__name__, e)
        if i < retries:
            time.sleep(backoff * (i + 1))
    return {"_error": last}


# ---------- 3. 取数封装 ----------
_MOCK_DAILY = None
def fetch_daily(code, market="SZ", n=5):
    if _MOCK_DAILY is not None:
        return _MOCK_DAILY
    return _get("/hs/history/%s.%s/d/%d" % (code, market, n), default=[])

def fetch_minute(code, market="SZ", n=240):
    return _get("/hs/pro/history/%s.%s/1/%d" % (code, market, n), default=[])

def fetch_indicator(kind, code, market="SZ", n=5):
    # kind in {'macd','ma','boll','kdj'}
    return _get("/hs/history/%s/%s.%s/d/%d" % (kind, code, market, n), default=[])

def fetch_dividend(code):
    return _get("/hs/gs/jnff/%s" % code, default=[])


# ---------- 4. 字段归一：统一成标准 OHLC ----------
def norm_kline(rows):
    """原始行 -> [{'date','open','high','low','close','volume','amount'}]"""
    out = []
    for r in rows or []:
        out.append({
            "date":    _hit_key(r, "date", "t", "时间", default="-"),
            "open":    _to_float(_hit_key(r, "open", "o")),
            "high":    _to_float(_hit_key(r, "high", "h")),
            "low":     _to_float(_hit_key(r, "low", "l")),
            "close":   _to_float(_hit_key(r, "close", "c")),
            "volume":  _to_float(_hit_key(r, "volume", "vol")),
            "amount":  _to_float(_hit_key(r, "amount", "amt")),
        })
    return out


# ---------- 5. 校验 ----------
def run_check():
    fake = [
        {"date": "2026-09-01", "open": 10.20, "high": 10.55, "low": 10.18,
         "close": 10.50, "volume": 120000, "amount": 1.26e8},
        {"date": "2026-09-02", "open": 10.50, "high": 10.62, "low": 10.40,
         "close": 10.45, "volume":  98000, "amount": 1.02e8},
    ]
    global _MOCK_DAILY
    _MOCK_DAILY = fake          # 演示数据替换网络调用
    rows = norm_kline(fetch_daily("600519", "SH", 5))
    _MOCK_DAILY = None
    assert len(rows) == 2 and rows[0]["close"] == 10.50

    # 指标端点可取
    assert fetch_indicator("macd", "600519", "SH", 5) is not None
    # 分红端点可取
    assert fetch_dividend("600519") is not None
    print("校验通过")


if __name__ == "__main__":
    run_check()
    print("-" * 62)
    code, market = "600519", "SH"
    for name, fn in [("日线", lambda: fetch_daily(code, market, 5)),
                     ("分钟K", lambda: fetch_minute(code, market, 240)),
                     ("MACD", lambda: fetch_indicator("macd", code, market, 5)),
                     ("BOLL", lambda: fetch_indicator("boll", code, market, 5)),
                     ("近年分红", fetch_dividend)]:
        data = fn() if name != "近年分红" else fn(code)
        if isinstance(data, dict) and "_error" in data:
            print("%-8s -> %s" % (name, data["_error"][:60]))
        else:
            print("%-8s -> %d 条" % (name, len(data) if isinstance(data, list) else 1))

5. 跑通示例：拿到日线后怎么用

把上面的代码跑通后，下面演示「日线 → 周 / 月线聚合」与「用分红数据手工前复权」——这两步是回测前必做的，也是官方没有单列接口、必须自己算的部分。

import pandas as pd

# —— 演示数据：真实调用替换为 fetch_daily('600519','SH',N) 的返回并经 norm_kline 归一 ——
demo = [
    ("2026-09-01", 10.20, 10.55, 10.18, 10.50, 120000, 1.26e8),
    ("2026-09-02", 10.50, 10.62, 10.40, 10.45,  98000, 1.02e8),
    ("2026-09-03", 10.45, 10.70, 10.43, 10.68, 150000, 1.58e8),
    ("2026-09-04", 10.68, 10.80, 10.60, 10.75, 132000, 1.41e8),
    ("2026-09-07", 10.75, 10.90, 10.70, 10.88, 110000, 1.18e8),
]
df = pd.DataFrame(demo, columns=["date", "open", "high", "low", "close", "volume", "amount"])
df["date"] = pd.to_datetime(df["date"])
df = df.set_index("date").sort_index()

# 日线 -> 周线 / 月线聚合（官方无周月线独立接口，用 resample）
weekly = df.resample("W").agg({"open": "first", "high": "max", "low": "min",
                               "close": "last", "volume": "sum", "amount": "sum"})
monthly = df.resample("ME").agg({"open": "first", "high": "max", "low": "min",
                                 "close": "last", "volume": "sum", "amount": "sum"})
print("周线：\n", weekly)
print("月线：\n", monthly)

# 手工前复权：用近年分红 /hs/gs/jnff 取到的除权事件
def forward_adjust(close, events):
    """
    close : pd.Series，index=日期(升序)，value=未复权收盘价
    events: list of (ex_date:str, div:float, pre_close:float)
            pre_close = 除权前一交易日收盘价
    """
    s = close.astype(float).copy()
    for ex_date, div, pre_close in events:
        ex = pd.Timestamp(ex_date)
        factor = (pre_close - div) / pre_close
        s.loc[s.index < ex] = s.loc[s.index < ex] * factor
    return s

raw = df["close"].copy()
adj = forward_adjust(raw, [("2026-06-10", 0.50, 12.00)])
print(pd.DataFrame({"未复权": raw, "前复权": adj.round(3)}))

若你的 K 线端点本身支持复权参数（前复权 / 后复权），直接传参最省事；否则用上面的分红法自行复权。回测全程必须统一口径，不能把复权价和未复权价混在一起算收益。

6. 坑与注意事项

坑 1：周 / 月线没有独立接口，别拼不存在的路径。
官方只提供日线（d）和一分钟线（1），周 / 月线是用日线 resample 聚合出来的。直接去猜 /hs/history/xxx/w/n 这类路径会 404。

坑 2：复权因子没有单列接口，要用分红自己算。
找不到"复权因子"端点时，用 /hs/gs/jnff 的近年分红数据，按"复权因子 =（除权前收盘 − 每股分红）/ 除权前收盘"向前连乘即可（见第 5 节）。

坑 3：技术指标端点返回的是已算好的值，不是原始价。
/hs/history/macd 等返回 DIF/DEA/柱（或 K/D/J、上中下轨），直接消费即可，不必自己再实现一遍指标，但字段名要先 print 看一行确认。

坑 4：分钟线量大，聚合前先想清楚范围。
Pro 分钟线一次可能返回几百上千根，做 5/15/30/60 分钟聚合前先限定 n，否则内存和限频都吃紧。

坑 5：停牌日缺失，resample 不会补 NaN。
resample 只会对已有日期聚合，停牌日不会自动补空。需要完整日历的，用交易日历 reindex + fillna 处理。

7. 小结与下篇预告

本篇把 9 个历史端点分成 4 组，给出日线 / 分钟线 / 技术指标 / 分红的取数封装，并用 pandas 演示了周月线聚合与手工前复权——这两块是官方没单列、必须自己处理的核心。

下一篇预告：《【Python量化实战全攻略 #03】指数与板块行情接口全拆解：大盘·行业·概念一键取（附代码）》：用 /hz 与 /hs/list/sectors 一组接口，把指数实时、行业板块、概念板块的成分与涨跌一次取全，择时与行业中性化都靠它。

8. 免责声明

本文仅演示沪深 A 股历史行情数据的取数与聚合方法，所有代码示例均为演示数据，未含任何真实行情数值，不构成投资建议，亦不承诺收益。

关于数据来源：本文接口来自智兔数服（www.zhituapi.com）官方文档，覆盖沪深A股、指数、基金、可转债等 200+ 已验证接口路径，注册可领免费 token 做研究测试。
更多实战手册见技术博客：zhituapi.com/blog.html。
本系列《Python量化实战全攻略》共 32 篇，下一篇（#03）讲指数与板块行情接口：大盘、行业、概念一键取。