系列：《量化系统从零构建》｜连载项目 · 纯 GET 取数 · 仅依赖 requests
适用：想把这些落库函数串成「首次全量回填 + 之后每日增量」的读者；数据由智兔数服提供，用 #03 客户端循环拉取，串起 #05/#06/#07 的落库逻辑，不依赖任何行情终端。

1. 你将得到什么

• backfill_daily：全量回填，遍历股票池逐个拉日线落库，单只失败不中断整体。
• incremental_update：增量更新骨架（按日期过滤，接口侧传参即可），盘后定时跑。
• 容错调度：一只失败跳过、其余继续，避免一只接口异常拖垮整批。
• 本篇交付：调度闭环，存储层（#04–#07）到此完整可用，信号层（#09 起）在此之上计算。

2. 本篇用到的取数约定

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

• 鉴权：token 走查询参数 ?token=，不要放进请求头。
• 错误形态：非 200 常见 404 102:Licence证书(你的智兔token)不存在 —— 证书不存在，不代表路径错。
• 落库函数（daily）沿用 #05 的 _normalize_daily / save_daily，本篇内联最小版以便自跑。数据来自 智兔数服（www.zhituapi.com）。

3. 调度设计

生产可用系统 cron / 计划任务定时调用 incremental_update；本篇用标准库 time 演示循环，不引入额外依赖。

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]}"


class SimpleClient:
    """最小客户端：request 复用 _get（完整限频/重试/缓存见 #03 ZhituClient）。"""
    def request(self, path, params=None, default=None):
        return _get(path, params=params, default=default)


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):
    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 backfill_daily(client, conn, codes):
    """全量回填：逐只拉日线落库；单只失败跳过不中断。返回写入行数。"""
    done = 0
    for code in codes:
        data, err = client.request(f"/hs/history/d/{code}", default=[])
        if err:
            print(f"  skip {code}: {err}")
            continue
        done += save_daily(conn, code, _normalize_daily(data))
    return done


def incremental_update(client, conn, codes, since):
    """增量更新：按 since 过滤（演示用，真实把 since 作为接口参数传入）。"""
    return backfill_daily(client, conn, codes)  # 接口侧按 since 取数即可


def run_check():
    # 合成数据仅逻辑校验（含容错），非真实行情
    synth = [{"date": "2024-01-02", "open": 1.0, "high": 1.1, "low": 0.9,
              "close": 1.05, "volume": 100, "amount": 105}]

    class FakeClient:
        def request(self, path, params=None, default=None):
            return synth, None
    conn = sqlite3.connect(":memory:"); init_daily_db(conn)
    n = backfill_daily(FakeClient(), conn, ["000001.SZ", "600000.SH"])
    assert n == 2
    cur = conn.cursor(); cur.execute("SELECT COUNT(*) FROM daily")
    assert cur.fetchone()[0] == 2

    # 容错：一只返回错误，另一只仍写入
    class Flaky:
        def request(self, path, params=None, default=None):
            return (default, "404 102") if "000001" in path else (synth, None)
    conn2 = sqlite3.connect(":memory:"); init_daily_db(conn2)
    backfill_daily(Flaky(), conn2, ["000001.SZ", "600000.SH"])
    cur2 = conn2.cursor(); cur2.execute("SELECT COUNT(*) FROM daily")
    assert cur2.fetchone()[0] == 1
    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. 限频在客户端层做：批量回填高频触发限流，SimpleClient 仅演示；生产用 #03 的 ZhituClient（带限频/重试）。
2. 增量靠 since 过滤：incremental_update 要把 since 作为接口参数，别每次全量重拉。
3. 容错要打印跳过：一只失败 print + continue，方便事后补拉失败清单。
4. 落库幂等：用 INSERT OR REPLACE，重跑同一天会覆盖，幂等安全。

7. 小结与下篇预告

本篇把 #05–#07 的落库函数串成「全量回填 + 增量更新」调度，单只失败不中断。存储层至此完整闭环。

下一篇计划写 #09《复权与清洗：复权因子·字段归一·异常对齐》：在落库数据上做前/后复权、缺失值与异常值处理、停复牌对齐，为信号层准备好干净数据。

8. 免责声明

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

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

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

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