系列：《Python 量化取数指南》｜连载项目 · 纯 GET 取数 · 仅依赖 requests · 本地库用 sqlite
数据：由智兔数服提供。更多接口见 智兔数服技术博客。

1. 你将得到什么

• 一个可复用落库封装：把历史 K 线 / 实时快照写进 sqlite
• 一个「拉 → 解析 → upsert → 读取」的完整示例
• 一个离线 run_check()，用合成数据验证落库/读取逻辑

2. 本篇取数约定

• 历史 K 线：/hz/history/fsjy/{code}.{market}/{lvl}（落库主力）
• 实时快照：/hs/real/ssjy/（落库可选）
• 请求：GET https://api.zhituapi.com<path>?token=<你的智兔token>
• 本地库用 sqlite3（Python 标准库，零依赖）；表按 kline_<code>_<lvl> 命名
• 落库前先复用第 3 篇 _get / _hit_key / _to_float

3. 核心模板（全系列复用 + 落库封装）

import time, json, sqlite3, requests

BASE = "https://api.zhituapi.com"
TOKEN = "你的智兔token"      # 演示证书（免费版）即可起步

def _get(path, params=None, timeout=15, retry=3, backoff=1.5):
    params = dict(params or {})
    params["token"] = TOKEN
    url = BASE + path
    last = None
    for i in range(retry):
        try:
            r = requests.get(url, params=params, timeout=timeout)
            if r.status_code != 200:
                last = f"HTTP {r.status_code} {r.text[:120]}"
                time.sleep(backoff * (i + 1)); continue
            try:
                return r.json(), None
            except ValueError:
                last = f"非JSON响应: {r.text[:120]}"
                return None, last
        except requests.RequestException as e:
            last = str(e); time.sleep(backoff * (i + 1))
    return None, last

def _hit_key(d, *keys, default=None):
    if not isinstance(d, dict):
        return default
    for k in keys:
        if k in d and d[k] not in (None, "", []):
            return d[k]
    return default

def _to_float(x, default=float("nan")):
    try:
        return float(x)
    except (TypeError, ValueError):
        return default

4. 跑通示例：拉 K 线 → 落 sqlite → 读取

def save_kline(db, code, lvl, rows):
    tbl = f"kline_{code.replace('.','')}_{lvl}"
    con = sqlite3.connect(db)
    con.execute(f"""CREATE TABLE IF NOT EXISTS {tbl} (
        date TEXT PRIMARY KEY, open REAL, high REAL, low REAL, close REAL, vol REAL)""")
    for r in rows:
        con.execute(f"""INSERT OR REPLACE INTO {tbl} VALUES (?,?,?,?,?,?)""",
                    (r["date"], r["open"], r["high"], r["low"], r["close"], r["vol"]))
    con.commit(); con.close()
    return tbl

def load_kline(db, code, lvl):
    tbl = f"kline_{code.replace('.','')}_{lvl}"
    con = sqlite3.connect(db)
    cur = con.execute(f"SELECT date,close FROM {tbl} ORDER BY date")
    rows = cur.fetchall(); con.close()
    return rows

def demo_store(db="quote.db"):
    data, err = _get("/hz/history/fsjy/000001.SH/d")
    if err:
        print("K线失败:", err); return
    bars = data if isinstance(data, list) else (data.get("data") or [])
    rows = [{"date": _hit_key(b, "date", "rq"),
             "open": _to_float(_hit_key(b, "open", "zk")),
             "high": _to_float(_hit_key(b, "high", "zg")),
             "low": _to_float(_hit_key(b, "low", "zd")),
             "close": _to_float(_hit_key(b, "close", "sp")),
             "vol": _to_float(_hit_key(b, "volume", "cjl"))} for b in bars]
    tbl = save_kline(db, "000001.SH", "d", rows)
    back = load_kline(db, "000001.SH", "d")
    print(f"  落库表 {tbl}，写入 {len(rows)} 根，回读 {len(back)} 根")

def run_check():
    db = ":memory:"
    save_kline(db, "SYN.SH", "d", [{"date":"20240102","open":1,"high":2,"low":1,"close":1.5,"vol":100}])
    back = load_kline(db, "SYN.SH", "d")
    print(f"  [run_check] 合成落库回读 {len(back)} 根: {back[0]}")

if __name__ == "__main__":
    demo_store()
    run_check()

返回字段说明：K 线 list 每项 date/rq、open/zk、high/zg、low/zd、close/sp、volume/cjl；落库后按 date 主键 upsert，回测直接 SELECT 读取，不再联网。

5. 坑与注意事项

1. 主键用 date：INSERT OR REPLACE 按 date 去重，重拉自动更新不重复。
2. 表名别用点：000001.SH 含点，建表前 replace('.','') 成 kline_000001SH_d。
3. 批量落库加事务：大量数据用 executemany + 一次 commit，别逐条提交。
4. 复权未处理：本篇只存原始 K 线，复权见第 14 篇，落库前先决定存前复权还是后复权。
5. 实时快照量大：/hs/real/ssjy/ 一次几千条，落库注意频率与去重。
6. 路径：quote.db 默认当前目录，生产放固定路径并加备份。

6. 常见报错速查

7. 小结与下一篇预告

小结：落库 = 拉 → 解析(_hit_key) → CREATE TABLE → INSERT OR REPLACE → SELECT；核心是按 date 主键 upsert + 表名去点 + 批量事务。

下一篇计划写 #14《数据清洗与复权对齐实战》：在落库基础上补复权与停牌/退市对齐，让回测数据干净。

8. 免责声明

本文仅演示公开数据接口的用法，所有代码示例均为演示数据，不构成任何投资建议；实际返回字段以接口文档与你的证书权限为准。数据由 智兔数服 提供，更多接口示例见 技术博客。

免费领取证书 / 查看完整接口文档，可前往 智兔数服官网。