系列：《量化系统从零构建》｜连载项目 · 纯 GET 取数 · 仅依赖 requests
适用：想把 #03 客户端拉到的行情落进 #04 的 SQLite 库的读者；数据由智兔数服提供，在 init_db 上加行情表，演示「拉取即落库」与失败降级不崩，不依赖任何行情终端。

1. 你将得到什么

• 行情表结构：daily（日线）、kline（K线）、tick（逐笔）、quote（盘口）四张表的设计要点。
• 字段归一 _normalize_daily：把接口中英文混用的字段名（如 收盘/close）统一成本地表列，落库不再 KeyError。
• 拉取即落库 fetch_and_save_daily：调 _get 拉日线 → 归一 → 写入；接口失败时降级为「不插入、不崩」。
• 本篇交付：行情落库闭环打通，#06 起的基本面 / 资金流落库复用同一套「归一 + 写入 + 降级」骨架。

2. 本篇用到的取数约定

GET https://api.zhituapi.com/<path>?token=你的智兔token

• 鉴权：token 走查询参数 ?token=，不要放进请求头。
• 错误形态：非 200 常见 404 102:Licence证书(你的智兔token)不存在 —— 证书不存在，不代表路径错。
• 完整多市场客户端（限频 / 重试 / 缓存 / 降级）见 #03 的 ZhituClient；本篇为聚焦落库，内联一个最小 _get。数据来自 智兔数服（www.zhituapi.com）。

3. 行情表结构

本篇先把 daily 跑通（含归一 + 降级），kline/tick/quote 结构类似，照 daily 替换列名即可。

4. 核心模板函数

import sys, sqlite3, requests

# ── 配置（与 #01 同源）──
BASE = "https://api.zhituapi.com"
TOKEN = "你的智兔token"  # 占位，换成你申请的真实 token


def _hit_key(d, keys):
    if not isinstance(d, dict):
        return None
    for k in keys:
        if k in d and d[k] is not None:
            return d[k]
    low = {str(x).lower(): x for x in d.keys()}
    for k in keys:
        if k.lower() in low:
            return d[low[k.lower()]]
    return None


def _to_float(v):
    try:
        return None if v is None else float(v)
    except (TypeError, ValueError):
        return None


def _get(path, params=None, default=None):
    """最小取数：鉴权 + 错误归一 + 降级。返回 (data, err)。"""
    p = dict(params or {})
    p["token"] = TOKEN
    try:
        r = requests.get(f"{BASE}{path}", params=p, timeout=10)
    except requests.RequestException as e:
        return default, f"网络异常：{e}"
    if r.status_code != 200:
        return default, f"{r.status_code} {r.text.strip()[:140]}"
    try:
        return r.json(), None
    except ValueError:
        return default, f"非 JSON：{r.text.strip()[:140]}"


def init_daily_db(conn):
    conn.execute("""CREATE TABLE IF NOT EXISTS daily (
        code TEXT, date TEXT, open REAL, high REAL, low REAL, close REAL,
        volume REAL, amount REAL, PRIMARY KEY(code, date))""")
    conn.commit()


def _normalize_daily(data):
    """把接口中英文混用字段归一成本地表列；data 可能是 list 或 {'data':[...]}。"""
    out = []
    items = data if isinstance(data, list) else (data.get("data") if isinstance(data, dict) else [])
    for it in (items or []):
        out.append({
            "date":   _hit_key(it, ["日期", "date", "trade_date"]) or "",
            "open":   _to_float(_hit_key(it, ["开盘", "open"])),
            "high":   _to_float(_hit_key(it, ["最高", "high"])),
            "low":    _to_float(_hit_key(it, ["最低", "low"])),
            "close":  _to_float(_hit_key(it, ["收盘", "close"])),
            "volume": _to_float(_hit_key(it, ["成交量", "volume", "vol"])),
            "amount": _to_float(_hit_key(it, ["成交额", "amount"])),
        })
    return out


def save_daily(conn, code, rows):
    n = 0
    for r in rows:
        conn.execute(
            "INSERT OR REPLACE INTO daily(code,date,open,high,low,close,volume,amount)"
            " VALUES (?,?,?,?,?,?,?,?)",
            (code, r["date"], r["open"], r["high"], r["low"], r["close"], r["volume"], r["amount"]))
        n += 1
    conn.commit()
    return n


def fetch_and_save_daily(conn, code, default=None):
    """拉日线并落库；接口失败降级 default（默认 []）-> 不插入、不崩。"""
    data, err = _get(f"/hs/history/d/{code}", default=default if default is not None else [])
    if err:
        return 0, err
    return save_daily(conn, code, _normalize_daily(data)), None


def run_check():
    # 合成数据仅逻辑校验（含字段中英文混用），非真实行情
    conn = sqlite3.connect(":memory:")
    init_daily_db(conn)
    synth = [
        {"日期": "2024-01-02", "开盘": 10.0, "最高": 10.5, "最低": 9.8, "收盘": 10.2, "成交量": 1000, "成交额": 10200},
        {"date": "2024-01-03", "open": 10.3, "high": 10.6, "low": 10.1, "close": 10.4, "volume": 1200, "amount": 12480},
    ]
    rows = _normalize_daily(synth)
    assert len(rows) == 2
    assert rows[0]["close"] == 10.2 and rows[1]["close"] == 10.4
    assert save_daily(conn, "000001.SZ", rows) == 2
    cur = conn.cursor()
    cur.execute("SELECT COUNT(*) FROM daily WHERE code='000001.SZ'")
    assert cur.fetchone()[0] == 2

    # 降级路径：接口失败（返回 []）时落库 0 行、不崩（用假响应模拟，不联网）
    class FakeResp:
        def __init__(self, status, text):
            self.status_code = status
            self.text = text
    orig = requests.get
    try:
        requests.get = lambda url, params=None, timeout=10: FakeResp(500, "err")
        conn2 = sqlite3.connect(":memory:"); init_daily_db(conn2)
        n2, e2 = fetch_and_save_daily(conn2, "000001.SZ", default=[])
        assert n2 == 0 and e2 is not None
    finally:
        requests.get = orig
    print("校验通过")


if __name__ == "__main__":
    if len(sys.argv) > 1 and sys.argv[1] == "--check":
        run_check()
    else:
        # 填入你的真实 token 后即可拉取真实数据
        print("hs.history ->", _get("/hs/history/d/000001.SZ"))

5. 跑通示例

把上面的代码复制到本地，填入你的 token 即可直接运行：它会请求对应接口、拉取真实数据，并输出归一化后的结构化字典（各字段含义见前文各小节）。

6. 坑与注意事项

1. 归一在落库前做：接口字段名中英文混用，落库前务必 _normalize，否则 None/缺键会写坏表。
2. 降级默认 []：fetch_and_save_daily 失败返回 0, err，下游按 err 判断是否补拉，不要假设一定写入。
3. INSERT OR REPLACE 覆盖更新：重跑同一天会更新而非重复插；如需保留首次，改 INSERT OR IGNORE。
4. 逐笔 tick 量大：逐笔数据行数极多，落库注意批量 executemany 与索引，别逐行 commit。

7. 小结与下篇预告

本篇打通「取数 → 归一 → 落库 → 降级」闭环，daily 表跑通后，kline/tick/quote 只是换列名。这是后续所有落库章节的骨架。

下一篇计划写 #06《基本面落库：财务三表·股东·分红·解禁·股本》：复用本篇的「归一 + 写入 + 降级」骨架，把公司财务三表、股东、分红、解禁、股本落进 fundamentals 相关表。

8. 免责声明

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

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

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

把代码里的 你的智兔token 换成你拿到的真实 token，上面的脚本就能直接打印行情日线落库数据。