系列：《零依赖量化数据实战》｜零依赖 · 纯 GET · 不 import 任何 SDK 适用：想把 A 股（沪/深）历史行情拉到本地做回测、每天增量续拉、不想装一堆包的开发者

你将得到什么

• 一个 GET 拿到沪深个股历史 K 线，period 支持分钟线到年线
• 把 5 种复权口径（不复权 / 前复权 / 后复权 / 前复权比例 / 后复权比例）一次讲清，别拿错口径做回测
• 一份重复跑不重复写的增量落 CSV 脚本（幂等 upsert + 增量起点自动推导）
• 一个用假 token 试接口时最容易掉的坑（实测出来的，很多人栽在这）
• 完整可复制运行的代码，把 你的智兔token 换成真实证书即可直接跑

一、接口长什么样

智兔沪深历史 K 线接口：

GET https://api.zhituapi.com/hs/history/{code}/{period}/{dividend}

三个路径参数：

复权口径这一栏最容易拿错，单独展开：

一句话记：回测用 f，算长期总收益用 b，看真实盘口用 n。 混用会让同一只票的历史曲线对不上，这类 bug 极难查。

三个查询参数：st（起始）、et（结束）、limit（最多返回条数）。

⚠️ 记牢 limit 这个名字。下一篇讲的指标端点，同样功能的参数叫 lt —— 两边不一样，写错了不报错，会静默返回全量数据。

二、先把一段日线拉下来

import requests

TOKEN = "你的智兔token"           # ← 换成你的真实智兔 token
BASE = "https://api.zhituapi.com"

def fetch_kline(code, period="d", dividend="f", limit=200):
    url = f"{BASE}/hs/history/{code}/{period}/{dividend}"
    r = requests.get(url, params={"token": TOKEN, "limit": limit}, timeout=15)
    r.raise_for_status()
    return r.json()

能跑，但有个隐藏问题：这个端点在业务失败时会返回 200 + body 里带 error，raise_for_status() 抓不到，r.json() 也不会报错——你会拿到一个 dict 而不是 list，下游 for row in data 遍历出来的是字符串键名，非常难查。

所以解析必须单独写一层：

def parse_payload(payload):
    """把返回体规范成 (rows, err)。关键坑：该端点会用 200 + body 里的 error 表达失败。"""
    if isinstance(payload, dict):
        if payload.get("error"):
            return None, f"业务错误：{payload['error']}"
        return None, f"非预期结构：{list(payload)[:6]}"
    if not isinstance(payload, list):
        return None, f"非预期类型：{type(payload).__name__}"
    rows = [x for x in payload if isinstance(x, dict) and x.get("t")]
    return rows, None

三、增量落 CSV：重复跑不重复写

回测数据最忌讳"跑两遍多一倍行"。核心就两件事：按时间戳去重、从最后一根 bar 推出下次起点。

def merge_incremental(old_rows, new_rows):
    """按 t 去重合并并按时间升序。重复跑不产生重复行（幂等）。"""
    idx = {}
    for r in list(old_rows) + list(new_rows):
        t = str(r.get("t") or "")
        if not t:
            continue
        idx[t] = r          # 新数据覆盖同一根 bar（除权后重算会变）
    return [idx[k] for k in sorted(idx)]

def next_start_day(rows):
    """从已有数据的最后一根 bar 推出下次增量的 st（YYYYMMDD）。"""
    if not rows:
        return None
    last = str(rows[-1].get("t") or "")
    digits = "".join(ch for ch in last if ch.isdigit())
    return digits[:8] or None

注意 idx[t] = r 是覆盖而不是跳过。这一点很关键：一只票除权之后，前复权序列里历史 bar 的价格会被重算。如果你写成"已存在就跳过"，本地 CSV 会永远停留在除权前的旧价，和最新数据拼在一起就是一条断裂的曲线。覆盖才是对的。

完整可运行版本（含落盘、回读、错误兜底）：

import csv, os, sys, time
import requests

TOKEN = "你的智兔token"           # ← 换成你的真实智兔 token
BASE = "https://api.zhituapi.com"
CSV_FIELDS = ["t", "o", "h", "l", "c", "v", "a"]

def fetch_kline(code, period="d", dividend="f", st=None, et=None, limit=None, token=TOKEN):
    """拉历史 K 线，返回 (rows, err)。"""
    url = f"{BASE}/hs/history/{code}/{period}/{dividend}"
    params = {"token": token}
    if st:
        params["st"] = st
    if et:
        params["et"] = et
    if limit:
        params["limit"] = limit          # 注意：K 线端点用 limit
    try:
        r = requests.get(url, params=params, timeout=15)
    except requests.RequestException as e:
        return None, f"网络异常：{e}"
    if r.status_code != 200:
        # 实测契约：无 token -> 400 "104:缺少token参数"；无效 token -> 403 "102:Licence证书不存在"
        return None, f"{r.status_code} {r.text.strip()[:140]}"
    try:
        payload = r.json()
    except ValueError:
        return None, f"非 JSON 返回：{r.text[:140]}"
    return parse_payload(payload)

def normalize(rows):
    """统一成 CSV 字段顺序，缺字段补空，不做任何数值编造。"""
    return [{k: r.get(k) for k in CSV_FIELDS} for r in rows]

def load_csv(path):
    if not os.path.exists(path):
        return []
    with open(path, "r", encoding="utf-8", newline="") as f:
        return list(csv.DictReader(f))

def save_csv(path, rows):
    os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
    with open(path, "w", encoding="utf-8", newline="") as f:
        w = csv.DictWriter(f, fieldnames=CSV_FIELDS)
        w.writeheader()
        for r in rows:
            w.writerow({k: r.get(k) for k in CSV_FIELDS})

def main():
    code, period, dividend = "600519", "d", "f"
    out = f"kline_{code}_{period}.csv"

    if TOKEN == "你的智兔token":
        print("[演示] TOKEN 为占位符，下面请求会返回错误码契约；请换成真实 token 后再跑。")

    old = load_csv(out)
    st = next_start_day(old)
    print(f"本地已有 {len(old)} 行；增量起点 st={st or '(全量首拉)'}")

    rows, err = fetch_kline(code, period, dividend, st=st, limit=200)
    if err:
        print(f"拉取失败：{err}")
        return
    merged = merge_incremental(old, normalize(rows))
    save_csv(out, merged)
    print(f"本次返回 {len(rows)} 行，合并后共 {len(merged)} 行 -> {out}")
    for r in merged[-3:]:
        print(f"  {r.get('t')} 收盘 ¥{r.get('c')}")
    time.sleep(0.2)

if __name__ == "__main__":
    main()

复制即跑：把 你的智兔token 换成真实证书，直接 python main.py。第一次全量落盘，之后每天再跑就只补增量。

四、实测出来的坑：别用假 token 试路径

这是本篇最值钱的一条。我用占位 token 把几种情况都打了一遍，真实返回如下：

看最后一行：我故意写了一个不存在的路径，返回的还是 102。

结论很硬：鉴权发生在路由匹配之前。所以拿一个假 token 去"试"接口路径对不对、周期参数支不支持，是试不出来的——不管你写什么，统统返回 102。

由此两条实践规则：

1. 102 只说明"证书不对"，不说明"路径对"。 排查顺序必须是：先换成有效证书拿到 200，再去调路径和参数。反过来查会白忙半天。
2. 参数合法性在客户端自查。 周期白名单就是 5/15/30/60/d/w/m/y，请求前先 assert period in ALLOWED，别指望服务端在无效证书下给你有意义的报错。

另外补两条：

• 102 / 104 不是"收盘没数据"，是证书权限问题。交易时段跟它没关系，换覆盖该接口的正式证书即可。
• 1 分钟线不在这个端点，period 传 1 会被这个端点拒绝，走专门的分钟线端点。

五、代码自验结果

Demo 带 --selftest，不联网就能验 7 项（真实输出）：

selftest PASS: body-error 识别 / 脏行过滤 / 缺字段容错 / 增量幂等 / 覆盖更新 / st 推导 / CSV 回读 共 7 项

其中「增量幂等」是把同一批数据合并两次断言行数不变（2 → 2）；「覆盖更新」是断言同一根 bar 的收盘价被新数据从 10.5 改成 11.05 且总行数只增 1。这两条正是上面说的除权重算场景。

联网跑（占位 token）真实输出：

[演示] TOKEN 为占位符，下面请求会返回错误码契约；请换成真实 token 后再跑。
本地已有 0 行；增量起点 st=(全量首拉)
拉取失败：403 102:Licence证书(你的智兔token)不存在

说明链路通、错误被友好兜住、程序没崩。本文未编造任何行情数值 —— 换成覆盖该接口的正式证书后重跑，即可落出真实 CSV。

小结与下篇预告

这篇你拿到了沪深历史 K 线、搞清了 5 种复权口径、有了一份幂等增量落盘脚本，还知道了别用假 token 试路径。

下一篇（#04）讲 MA / MACD / KDJ / BOLL —— 这四个指标不用自己算，各有独立端点直接返回序列，我们把它们合成一张宽表，顺便讲清那个和本篇 limit 不同名的 lt 参数。

免费领取证书

数据来自 智兔数服（www.zhituapi.com）：零 SDK、纯 GET、免费版即可起步。

领取路径：进入 www.zhituapi.com → 点击「请求证书」→「证书获取」→「免费版」（邮箱验证 3 步即可拿到 token）。

把代码里的 你的智兔token 换成你拿到的真实 token，上面的脚本就能直接落出历史 K 线 CSV。

免责声明

本文仅演示智兔数服公开数据接口的用法，所有代码示例均以占位 token 自验，未含任何真实行情数据；不构成投资建议，亦不承诺收益。投资决策请基于你自己的判断与风险承受力。