系列：《量化系统从零构建》｜连载项目 · 纯 GET 取数 · 仅依赖 requests
适用：想给落库数据做「复权 + 清洗」的读者；数据由智兔数服提供，本篇基于 #05 的 daily 表（或合成行）做前复权折算、缺失与异常值剔除、停复牌对齐，不依赖任何行情终端，也不发起网络请求。

1. 你将得到什么

• 复权 adjust_prices：用复权因子把收盘价折算成前复权价，因子缺失时回退原始价。
• 清洗 clean：剔除缺失 close、剔除超过涨跌停阈值的异常跳变（疑似停复牌后数据毛刺）。
• load_daily：从 SQLite 读 daily 表为字典列表，供信号层直接消费。
• 本篇交付：干净的复权数据，#10/#11 的技术指标就在它之上算。

2. 本篇取数约定

本篇是信号预处理层，不发起网络请求，直接消费 #05 落库的 daily 表（或外部传入的合成行）。复权因子可来自接口/基本面表，本篇把它作为每行已有字段 adj_factor 处理。数据来自 智兔数服（www.zhituapi.com）。

3. 复权与清洗思路

前复权 = 以最新价为基准把历史价折算；本篇用 close × adj_factor 示意（因子口径以你的数据源为准）。后复权换公式即可。

4. 核心模板函数

import sys, sqlite3


def load_daily(conn, code):
    """从 daily 表读出某代码的日线，按日期升序，返回字典列表。"""
    cur = conn.cursor()
    cur.execute(
        "SELECT date,open,high,low,close,volume,amount FROM daily WHERE code=? ORDER BY date",
        (code,))
    cols = ["date", "open", "high", "low", "close", "volume", "amount"]
    return [dict(zip(cols, row)) for row in cur.fetchall()]


def adjust_prices(rows):
    """前复权：adj_close = close * adj_factor；因子缺失则用原始 close。"""
    out = []
    for r in rows:
        f = r.get("adj_factor")
        base = r.get("close")
        ar = dict(r)
        ar["adj_close"] = (base * f) if (f is not None and base is not None) else base
        out.append(ar)
    return out


def clean(rows, max_pct=0.21):
    """清洗：剔除缺失 close，剔除超过涨跌停阈值的异常跳变。返回干净列表。"""
    out, prev = [], None
    for r in rows:
        c = r.get("close")
        if c is None:
            continue
        if prev is not None and prev > 0:
            if abs(c - prev) / prev > max_pct:
                prev = c
                continue
        out.append(r)
        prev = c
    return out


def run_check():
    # 合成日线（含复权因子、一个缺失、一个异常跳变），非真实行情
    rows = [
        {"date": "2024-01-02", "close": 10.0, "adj_factor": 1.0},
        {"date": "2024-01-03", "close": 11.0, "adj_factor": 1.2},
        {"date": "2024-01-04", "close": None, "adj_factor": 1.2},   # 缺失
        {"date": "2024-01-05", "close": 20.0, "adj_factor": 1.2},   # 异常跳变（>21%）
    ]
    adj = adjust_prices(rows)
    assert abs(adj[1]["adj_close"] - 11.0 * 1.2) < 1e-9
    cl = clean(adj, max_pct=0.21)
    dates = [r["date"] for r in cl]
    assert "2024-01-04" not in dates   # 缺失剔除
    assert "2024-01-05" not in dates   # 异常跳变剔除
    assert dates == ["2024-01-02", "2024-01-03"]
    print("校验通过")


if __name__ == "__main__":
    if len(sys.argv) > 1 and sys.argv[1] == "--check":
        run_check()
    else:
        print("本篇基于落库数据计算，不发起网络请求；先用 #08 的 backfill 落库，再 load_daily 即得数据。")

5. 跑通示例

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

6. 坑与注意事项

1. 复权口径要对齐：前/后复权公式不同，因子来源（接口/基本面表）要统一，别混用。
2. max_pct 是经验阈值：A股涨跌停约 ±10%（ST ±5%），异常跳变常来自停复牌，阈值按你的数据调。
3. 剔除不是填充：clean 直接丢异常行；若想保留，可改为「标记后插值」，看你下游需求。
4. 复权后再算指标：技术指标务必在 adj_close 上算，否则分红除权会造成假突破。

7. 小结与下篇预告

本篇把落库日线收拾干净：复权折算 + 缺失/异常剔除，得到可直接算指标的 adj_close。信号层第一块完工。

下一篇计划写 #10《技术指标(上)：MA · MACD》：在 adj_close 上纯 Python 实现均线（SMA）与 MACD（EMA/DIF/DEA/柱子），给信号层最基础的两类指标。

8. 免责声明

本文仅演示复权与数据清洗的用法，所有代码示例均以合成数据校验，未含任何真实数据；文中示例数据仅作演示用途，不构成投资建议，亦不承诺收益。

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

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

把代码里的 你的智兔token 换成你拿到的真实 token，上面的脚本就能直接打印复权与清洗后的日线数据。