系列：《A股分钟级高频数据完全手册》｜连载项目 · 纯 GET 取数 · 仅依赖 requests · Pro 版分钟级高频

① 你将得到什么

前 14 篇都在取数，但数据不落库就只是一次性脚本。生产级做法是：把分钟数据按 (代码,时间) 主键写入本地库，用增量水位线（每只票记录已拉到的最大时间）避免重复拉取，按存储选型（SQLite 轻量、Parquet 列式适合分析）分层。本篇把这套落库骨架讲透，并导览 15–28 的技术指标周期。

本篇产出：① 一套生产级落库模板（增量水位线 + 按主键 upsert 去重 + SQLite/Parquet 选型）；② 本周期 15–28 的 14 篇地图（Pro 技术指标三市场×4 + 综合行情指标）；③ 落地后如何低重复、低成本滚雪球攒数据。

读完这篇，你应当能独立把本篇端点数据拉下来、落本地，并跑出一份可用的加工分析表。

② 取数约定

一个端点拉数据，返回里字段名可能中英文混用（不同接口版本、不同数据源习惯不同）。演示用 你的token 占位，代码里 MOCK=True 时回放演示数据，接真实 token 把 MOCK 改成 False 即可，无需改动其他逻辑。

所有请求走 https://api.zhituapi.com + ?token=，纯 GET、仅依赖 requests，不依赖任何付费 SDK。字段口径以接口实际返回为准，本文用 _hit_key 做了大小写/中英文容错；落地前建议先用 print(r[0].keys()) 核对一次你账号下的真实字段名。

落地建议：把每天拉到的数据按 代码 + 日期 做主键写入本地 SQLite/CSV，避免重复插入；历史回看靠本地累积，而非反复调接口。本篇示例代码已包含字段容错映射与本地加工函数，你只需把 print 替换为写入语句。

③ 端点分组地图

本篇涉及字段（落地建表参考）：时间、收、成交量、指标值。

返回结构为 JSON 数组（或含数组字段的对象），每条记录即一根分钟 bar / 一笔明细；字段映射函数已按上述字段做容错映射，直接复用即可。

④ 核心模板函数（自包含·可跑）

import requests, time, copy
from urllib.parse import quote

BASE = "https://api.zhituapi.com"
# 将 "你的token" 替换成您自己的智兔API token（官网注册后在「控制台 / 接口管理」页获取，免费额度即可起步）。
TOKEN = "你的token"

# ---------- 字段容错与类型归一 ----------
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

# ---------- 统一请求：token 鉴权 + 重试退避（MOCK 开关） ----------
MOCK = True
DEMO_DATA = {}

def _get(path, params=None, timeout=10, retries=2, backoff=0.6, default=None):
    """返回 JSON；MOCK=True 时回放 DEMO_DATA；否则真实 GET 并带 token 鉴权、重试 retries 次"""
    if MOCK:
        return copy.deepcopy(DEMO_DATA.get(path, {} if default is None else default))
    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()
                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}


# ---------- 本篇技术深入：生产级落库（增量水位线 + 去重 + SQLite） ----------
import sqlite3  # 仅演示用标准库，生产可换 Parquet/DuckDB
def _mk_demo(n=12):
    rows = []
    for i in range(n):
        base = 10.0 + i * 0.1
        rows.append({"t":"2024-01-10 09:%02d:00" % (31 + i), "o":"%.2f" % base,
                     "c":"%.2f" % (base + 0.05), "h":"%.2f" % (base + 0.08),
                     "l":"%.2f" % (base - 0.02), "v":"%d" % (10000 + i * 500),
                     "a":"%d" % int((base + 0.05) * 1000)})
    return rows

DEMO_DATA = {"/hs/pro/latest/000001.SZ/1/12": _mk_demo(12)}

def fetch_and_store(code="000001.SZ", n=12):
    rows = _get("/hs/pro/latest/%s/1/%d" % (code, n), default=[])
    if not isinstance(rows, list):
        rows = []
    # 演示：把分钟 bar 落 SQLite，按 (代码,时间) 主键 upsert 去重
    conn = sqlite3.connect(":memory:")
    conn.execute("CREATE TABLE IF NOT EXISTS bar (code TEXT, t TEXT, c REAL, v REAL, PRIMARY KEY(code,t))")
    ins = 0
    for r in rows:
        t = _hit_key(r, "t", "time"); c = _to_float(_hit_key(r, "c", "close")); v = _to_float(_hit_key(r, "v", "vol", "volume"))
        conn.execute("INSERT OR IGNORE INTO bar VALUES (?,?,?,?)", (code, t, c, v)); ins += 1
    conn.commit()
    dedup = conn.execute("SELECT COUNT(*) FROM bar").fetchone()[0]
    return ins, dedup

def run_check():
    ins, dedup = fetch_and_store()
    assert ins == 12 and dedup == 12
    assert _hit_key({"T":"x"}, "t", "time") == "x"
    assert _to_float("1.5") == 1.5 and _to_float("-") is None
    print("校验通过")

if __name__ == "__main__":
    run_check()
    ins, dedup = fetch_and_store()
    print("写入=%d 去重后=%d" % (ins, dedup))

代码讲解：① SCAFFOLD 提供字段容错 _hit_key 与带 token 鉴权、重试退避的请求 _get，MOCK 开关让你无需 token 就能验证逻辑；② 本篇导览本周期 14 篇地图（见 §③），并给出当篇技术深入的完整代码（落库 / 因子计算等实质内容，函数命名见代码块）；③ 技术深入代码在本地做加工（如落库去重、因子计算与防未来函数），不消耗接口额度；④ run_check() 用断言自检，可放进 CI 持续校验。

⑤ 跑通示例（演示数据）

把上面代码保存为 hf_15.py，MOCK=True 直接跑（Python3 + requests 即可，无需其他依赖）。

预期输出：先打印 校验通过（本篇技术深入的字段容错、取数封装与逻辑自检），随后打印技术深入部分的运行结果（如落库条数 / 因子数组等）。这说明 ①token 鉴权封装可用 ②本篇端点 / 逻辑可跑通 ③关键断言成立。关掉 MOCK、填入真实 token 即可对接生产；把演示写入替换成你的存储即可落地。

⑥ 编号坑

• 落库主键务必唯一（如 代码+时间），用 INSERT OR IGNORE / upsert 去重，重复拉取不会污染数据。
• 增量水位线：每只票记录已拉到的最大时间，下次只拉其后数据，省额度又保证连续。
• 存储选型：SQLite 适合边拉边写的事务流；Parquet/CSV 适合离线分析，按场景分层。
• 时间统一为北京时间并建索引，跨日聚合与按区间回放才快。
• Pro 额度有限，落库后回测只读本地，不再反复调接口，这是省钱的关键。
• 字段口径变动（接口升级）要在落库时打版本标记，便于历史回溯。
• MOCK 演示数据虚构，实盘落库前小批量验证字段与时间连续性。
• 落库脚本建议加重试/断点，单票失败不影响整体 pipeline。

⑦ 小结与下篇预告

落库是把取数变资产的临门一脚：增量水位线+主键去重+分层存储。下一篇 #16 进入技术指标周期，用 Pro 版 MACD 历史接口算个股分钟级 MACD。

本篇是《A股分钟级高频数据完全手册》第 15/42 篇。建议按系列顺序读：主推篇把体系与地基搭好，后面的单点篇（如 #02 起）都是换端点、加加工。

下一篇预告：【A股分钟级高频数据完全手册 #16】Pro版MACD指标历史接口：1分钟MACD计算与信号

⑧ 免责声明

本文代码仅用于接口调用与本地数据落库的技术演示，所有示例数据均为演示数据（MOCK 开关开启），不构成任何投资建议。股市有风险，入市需谨慎；实盘使用前请在小资金下充分验证接口稳定性与字段口径。

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

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

把代码里的 你的token 换成你拿到的真实 token，上面的脚本就能直接拉取 Pro 版分钟级行情、实时盘口与逐笔成交数据，照着本系列 42 篇即可从零搭好属于你自己的分钟级高频数据底座。