系列：《量化系统从零构建》｜连载项目 · 纯 GET 取数 · 仅依赖 requests
适用：想把市场资金面数据落进 #04 的 SQLite 库的读者；数据由智兔数服提供，复用 #05/#06 的「归一 + 写入 + 降级」骨架，把个股资金流向、板块分类、龙虎榜落进 money_flow / sector / lhb 表，不依赖任何行情终端。

1. 你将得到什么

• 资金面表结构：money_flow（个股资金流）、sector（板块分类）、lhb（龙虎榜）三张表。
• 归一 + 落库：以资金流向为例跑通「拉取 → 归一 → 写入 → 降级」；path 由调用方按官方文档传入，避免写死可能变动的子路径。
• 本篇交付：存储层最后一块补齐——行情（#05）、基本面（#06）、资金面（#07）三足鼎立，#08 起做初始化与增量调度。

2. 本篇用到的取数约定

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

• 鉴权：token 走查询参数 ?token=，不要放进请求头。
• 错误形态：非 200 常见 404 102:Licence证书(你的智兔token)不存在 —— 证书不存在，不代表路径错。
• 取数函数 _get / _hit_key / _to_float 沿用 #05，本篇直接复用。数据来自 智兔数服（www.zhituapi.com）。

3. 资金面表结构

本篇把 money_flow 跑通（含降级），sector / lhb 表一并建好，照同样骨架写各自 _normalize_* 即可。真实 path 以官方文档为准（如资金流向、板块分类、龙虎榜对应端点）。

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):
    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_flow_db(conn):
    conn.executescript("""
    CREATE TABLE IF NOT EXISTS money_flow (code TEXT, date TEXT, main_in REAL, main_out REAL, net REAL, PRIMARY KEY(code,date));
    CREATE TABLE IF NOT EXISTS sector      (code TEXT, name TEXT, PRIMARY KEY(code));
    CREATE TABLE IF NOT EXISTS lhb         (code TEXT, date TEXT, reason TEXT, net REAL, PRIMARY KEY(code,date));
    """)


def _normalize_flow(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"]) or "",
            "main_in":  _to_float(_hit_key(it, ["主力流入", "main_in"])),
            "main_out": _to_float(_hit_key(it, ["主力流出", "main_out"])),
            "net":      _to_float(_hit_key(it, ["主力净流入", "net", "净流入"])),
        })
    return out


def save_flow(conn, code, rows):
    n = 0
    for r in rows:
        conn.execute(
            "INSERT OR REPLACE INTO money_flow(code,date,main_in,main_out,net)"
            " VALUES (?,?,?,?,?)",
            (code, r["date"], r["main_in"], r["main_out"], r["net"]))
        n += 1
    conn.commit()
    return n


def fetch_and_save_flow(conn, code, path, default=None):
    """path 由调用方按官方文档传入（如资金流向端点）；其余逻辑同 #05。"""
    data, err = _get(path, default=default if default is not None else [])
    if err:
        return 0, err
    return save_flow(conn, code, _normalize_flow(data)), None


def run_check():
    # 合成数据仅逻辑校验，非真实行情
    conn = sqlite3.connect(":memory:"); init_flow_db(conn)
    synth = [{"日期": "2024-01-02", "主力流入": 1.2e8, "主力流出": 1.0e8, "主力净流入": 2.0e7}]
    rows = _normalize_flow(synth)
    assert rows[0]["net"] == 2.0e7
    assert save_flow(conn, "000001.SZ", rows) == 1
    cur = conn.cursor()
    cur.execute("SELECT COUNT(*) FROM money_flow WHERE code='000001.SZ'")
    assert cur.fetchone()[0] == 1
    # 降级：接口失败时不插入、不崩
    class FakeResp:
        def __init__(self, s, t):
            self.status_code = s
            self.text = t
    orig = requests.get
    try:
        requests.get = lambda u, params=None, timeout=10: FakeResp(500, "e")
        conn2 = sqlite3.connect(":memory:"); init_flow_db(conn2)
        n2, e2 = fetch_and_save_flow(conn2, "000001.SZ", "/hizj/demo", 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("ht.nbzj.lxgl ->", _get("/ht/nbzj/lxgl"))

5. 跑通示例

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

6. 坑与注意事项

1. path 以官方文档为准：资金流 / 板块 / 龙虎榜的确切子路径随文档更新，调用方传参，别把可能变动的路径写死在落库函数里。
2. 资金流金额单位要核对：净流入可能是「元」或「万元」，归一后用 _to_float 拿到数值，单位统一靠你自己约定。
3. 龙虎榜按日：lhb 主键 (code, date)，同一天同代码只一条，用 INSERT OR REPLACE 覆盖更新。
4. 板块分类变化慢：sector 不必日更，季更即可。

7. 小结与下篇预告

本篇补齐存储层最后一块：行情（#05）、基本面（#06）、资金面（#07）三张数据网全部落库。下一步是把这些表串成「全量回填 + 增量更新」的调度。

下一篇计划写 #08《初始化与增量更新调度：全量回填 + 定时增量》：写 backfill / incremental_update，用 #03 客户端循环拉取并落库，串起前面所有落库函数。

8. 免责声明

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

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

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

把代码里的 你的智兔token 换成你拿到的真实 token，上面的脚本就能直接打印资金流与龙虎榜数据。