系列：《量化系统从零构建》｜连载项目 · 纯 GET 取数 · 仅依赖 requests
适用：想把公司基本面数据落进 #04 的 SQLite 库的读者；数据由智兔数服提供，复用 #05 的「归一 + 写入 + 降级」骨架，把财务三表、股东、分红、解禁、股本落进 fundamentals 相关表，不依赖任何行情终端。

1. 你将得到什么

• 基本面表结构：balance / income / cashflow（三表）、holders（股东）、dividend（分红）、lift_lock（解禁）、capital（股本）七张表。
• 归一 + 落库：以资产负债表 /hs/fin/balance 为例跑通「拉取 → 归一 → 写入 → 降级」，其余表照同样骨架换列名。
• 本篇交付：基本面落库闭环，与 #05 行情落库并列，构成存储层双支柱。

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. 基本面表结构

本篇把 balance 跑通，其余表用 CREATE TABLE 一并建好，照 balance 写各自 _normalize_* 即可。

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_fundamentals_db(conn):
    conn.executescript("""
    CREATE TABLE IF NOT EXISTS balance   (code TEXT, date TEXT, total_assets REAL, total_liab REAL, equity REAL, PRIMARY KEY(code,date));
    CREATE TABLE IF NOT EXISTS income    (code TEXT, date TEXT, revenue REAL, net_profit REAL, PRIMARY KEY(code,date));
    CREATE TABLE IF NOT EXISTS cashflow  (code TEXT, date TEXT, net_cf REAL, PRIMARY KEY(code,date));
    CREATE TABLE IF NOT EXISTS holders   (code TEXT, date TEXT, holder TEXT, ratio REAL, PRIMARY KEY(code,date,holder));
    CREATE TABLE IF NOT EXISTS dividend  (code TEXT, year INT, amount REAL, PRIMARY KEY(code,year));
    CREATE TABLE IF NOT EXISTS lift_lock (code TEXT, date TEXT, volume REAL, PRIMARY KEY(code,date));
    CREATE TABLE IF NOT EXISTS capital   (code TEXT, total REAL, float REAL, PRIMARY KEY(code));
    """)


def _normalize_balance(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", "period"]) or "",
            "total_assets": _to_float(_hit_key(it, ["总资产", "total_assets"])),
            "total_liab":   _to_float(_hit_key(it, ["总负债", "total_liab"])),
            "equity":       _to_float(_hit_key(it, ["股东权益", "equity", "净资产"])),
        })
    return out


def save_balance(conn, code, rows):
    n = 0
    for r in rows:
        conn.execute(
            "INSERT OR REPLACE INTO balance(code,date,total_assets,total_liab,equity)"
            " VALUES (?,?,?,?,?)",
            (code, r["date"], r["total_assets"], r["total_liab"], r["equity"]))
        n += 1
    conn.commit()
    return n


def fetch_and_save_balance(conn, code, default=None):
    data, err = _get(f"/hs/fin/balance/{code}", default=default if default is not None else [])
    if err:
        return 0, err
    return save_balance(conn, code, _normalize_balance(data)), None


def run_check():
    # 合成数据仅逻辑校验，非真实行情
    conn = sqlite3.connect(":memory:"); init_fundamentals_db(conn)
    synth = [{"报告期": "2023-12-31", "总资产": 1000.0, "总负债": 600.0, "股东权益": 400.0}]
    rows = _normalize_balance(synth)
    assert rows[0]["equity"] == 400.0
    assert save_balance(conn, "000001.SZ", rows) == 1
    cur = conn.cursor()
    cur.execute("SELECT COUNT(*) FROM balance 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_fundamentals_db(conn2)
        n2, e2 = fetch_and_save_balance(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.fin.balance ->", _get("/hs/fin/balance/000001.SZ"))

5. 跑通示例

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

6. 坑与注意事项

1. 三表日期用报告期：财务数据是「报告期」粒度（季报 / 年报），不是交易日，落库前确认 date 是报告期。
2. 股东表主键带 holder：十大股东同名会冲突，主键加 holder 列区分。
3. 归一列名要对齐表：每张表写自己的 _normalize_*，字段候选键覆盖中英文，避免写 None 坏表。
4. 解禁 / 股本变化慢：这两类不必日更，周更或事件触发更合理。

7. 小结与下篇预告

本篇把基本面七张表建好并以资产负债表跑通「归一 + 写入 + 降级」。至此行情（#05）与基本面（#06）双支柱落库完成。

下一篇计划写 #07《资金流·板块·龙虎榜 落库》：复用同一骨架，把个股资金流向、板块分类、龙虎榜落进 money_flow / sector / lhb 表，补齐存储层最后一块。

8. 免责声明

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

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

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

把代码里的 你的智兔token 换成你拿到的真实 token，上面的脚本就能直接打印公司基本面财务数据。