08 · 数据质量怎么量

   前五个维度：数据可以自己证明自己
   ─────────────────────────────────────────────
   「金额是数字吗」「主键重复吗」「时间戳新吗」
    → 只看这份数据本身就能判定，全部可自动化

   第六个维度：数据永远不能自己证明自己  ⭐⭐
   ─────────────────────────────────────────────
   「这个用户的年龄真的是 34 岁吗」
    → 表里写着 34，格式合法、范围合法、和其他字段自洽
      但他可能今年 51 岁
    → 需要外部参照：人工抽检 / 对账 / 回访 / 另一个系统

def check_completeness(df, required, min_rows, max_rows, group_col=None):
    """required: {列名: 允许的最大缺失率}"""
    out = []
    n = len(df)
    out.append(("行数在区间内", min_rows <= n <= max_rows, {"rows": n}))
    for col, max_null in required.items():
        r = float(df[col].isna().mean())
        out.append((f"{col} 缺失率<={max_null}", r <= max_null, {"null_rate": round(r, 5)}))
        # ⭐ 分组看：整体 0.4% 可能是某个渠道 100% 缺 + 其他渠道 0%
        if group_col is not None:
            g = df.groupby(group_col)[col].apply(lambda s: s.isna().mean())
            worst = g.idxmax(), round(float(g.max()), 5)
            out.append((f"{col} 分组最差缺失率<={max_null*3}",
                        g.max() <= max_null * 3, {"worst_group": worst}))
    return out

def check_uniqueness(df, pk_cols, sample=5):
    d = df.duplicated(subset=pk_cols, keep=False)
    bad = df.loc[d, pk_cols].drop_duplicates().head(sample)
    return [("主键唯一", not d.any(),
             {"dup_rows": int(d.sum()),
              "dup_rate": round(float(d.mean()), 6),
              # ⭐ 报告里必须带失败样例的主键，否则没人能去查
              "samples": bad.to_dict("records")})]

import re
import pandas as pd

def check_validity(df, ranges=None, enums=None, regexes=None, fk=None):
    """fk: {列名: 允许取值的集合}  —— 引用完整性"""
    out = []
    for col, (lo, hi) in (ranges or {}).items():
        s = pd.to_numeric(df[col], errors="coerce")
        bad = ((s < lo) | (s > hi)).sum()
        out.append((f"{col} in [{lo},{hi}]", bad == 0, {"bad": int(bad)}))
    for col, allowed in (enums or {}).items():
        bad = (~df[col].isin(allowed)) & df[col].notna()
        # ⭐ 一定要把"新出现的枚举值"打出来：多半是上游加了新类型没通知你
        unseen = sorted(set(df.loc[bad, col].astype(str)))[:10]
        out.append((f"{col} 枚举合法", not bad.any(),
                    {"bad": int(bad.sum()), "unseen": unseen}))
    for col, pat in (regexes or {}).items():
        c = re.compile(pat)
        bad = df[col].dropna().astype(str).map(lambda v: c.fullmatch(v) is None).sum()
        out.append((f"{col} 格式合法", bad == 0, {"bad": int(bad)}))
    for col, allowed in (fk or {}).items():
        bad = (~df[col].isin(allowed)).sum()
        out.append((f"{col} 外键可对上", bad == 0, {"orphan": int(bad)}))
    return out

def check_consistency(df, rules, recon=None, tol=1e-6):
    """rules: {规则名: 返回布尔 Series 的函数}
       recon: (明细聚合后的 Series, 汇总表的 Series) —— 对账"""
    out = []
    for name, fn in rules.items():
        ok = fn(df)
        out.append((name, bool(ok.all()), {"violations": int((~ok).sum())}))
    if recon is not None:
        a, b = recon
        diff = (a.reindex(b.index).fillna(0) - b).abs()
        # ⭐ 对账是唯一能发现"口径悄悄变了"的检查
        out.append(("明细汇总对账", bool((diff <= tol).all()),
                    {"max_diff": float(diff.max()), "bad_keys": list(diff[diff > tol].index[:5])}))
    return out

RULES = {
    "支付时间不早于下单": lambda d: d["pay_time"].isna() | (d["pay_time"] >= d["create_time"]),
    "已支付必须有支付时间": lambda d: (d["status"] != "paid") | d["pay_time"].notna(),
    "实付不超过应付": lambda d: d["paid_amt"] <= d["total_amt"] + 0.01,
}

import hashlib
import pandas as pd

def check_timeliness(df, ts_col, partition_ts, max_lag_h=3, prev_fingerprint=None):
    ts = pd.to_datetime(df[ts_col])
    lag_h = (partition_ts - ts.max()).total_seconds() / 3600
    out = [("数据新鲜度", lag_h <= max_lag_h, {"lag_hours": round(float(lag_h), 2)}),
           ("到达延迟P99可接受",
            float((partition_ts - ts).dt.total_seconds().quantile(0.99)) <= max_lag_h * 3600 * 2,
            {})]
    # ⭐⭐ 一行防「今天的分区是昨天的拷贝」：内容指纹不能和上一个分区相同
    fp = hashlib.md5(pd.util.hash_pandas_object(df, index=False).values).hexdigest()
    if prev_fingerprint is not None:
        out.append(("与上一分区内容不同", fp != prev_fingerprint, {"fingerprint": fp[:12]}))
    return out

import numpy as np

def sample_for_review(df, n=200, strata=None, seed=0):
    """分层抽样出待人工复核的样本；⭐ 别用简单随机抽，稀有类会抽不到"""
    if strata is None:
        return df.sample(min(n, len(df)), random_state=seed)
    g = df.groupby(strata, group_keys=False)
    per = max(1, n // max(1, g.ngroups))
    return g.apply(lambda s: s.sample(min(per, len(s)), random_state=seed))

def accuracy_from_review(n_checked, n_wrong, conf=0.95):
    """人工复核完之后：错误率点估计 + 单侧上限（全对时也别写 0%）"""
    p = n_wrong / n_checked
    upper = 1 - (1 - conf) ** (1 / n_checked) if n_wrong == 0 else p + 1.96 * np.sqrt(p * (1 - p) / n_checked)
    return {"error_rate": round(p, 4), "upper_bound": round(float(upper), 4)}

# ⭐ 抽 200 条全对 → 只能说"95% 置信下错误率 < 1.49%"，不能说"错误率 0%"

from dataclasses import dataclass, field
from typing import Callable, Any

@dataclass
class Check:
    id: str
    dim: str                    # completeness / uniqueness / validity / consistency / timeliness
    level: str                  # BLOCK / WARN / INFO   ⭐ 级别是断言的一部分，不是事后决定的
    fn: Callable[[Any], list]

def run_suite(df, checks, ctx=None):
    report, blocked = [], False
    for c in checks:
        for name, ok, detail in c.fn(df):
            report.append({"check_id": c.id, "dim": c.dim, "level": c.level,
                           "name": name, "passed": bool(ok), "detail": detail,
                           "n_rows": len(df)})
            if not ok and c.level == "BLOCK":
                blocked = True          # ⭐ 只有 BLOCK 级失败才拦流水线
    return {"blocked": blocked, "results": report}

   ❌ 反模式一：所有断言都是 BLOCK
      → 第一次大促当天全线飘红 → 有人加了个"强制通过"按钮
      → 三周后所有人默认先点强制通过再看报告
      💀 门禁就此死亡，但报表上它还是"100% 覆盖"

   ❌ 反模式二：BLOCK 了但没有回退目标
      → 拦住了脏数据，下游拿到的是"没有数据"
      → 有些场景比脏数据更糟（特征服务直接返回默认值）
      ⭐ BLOCK 的正确语义是"回退到上一个已知良好版本"，不是"什么都不给"

   ❌ 反模式三：断言写在数据加工代码里
      → 改一次口径要动两处，很快就不同步
      ⭐ 断言应该和数据契约（第 3 章）放在一起，独立于加工逻辑

   机器可读（JSON，落库）              人可读（Markdown / HTML）
   ────────────────────────           ──────────────────────────
   check_id / dim / level             ✅ 32 通过  🟡 3 警告  🔴 0 阻断
   passed / n_rows / fail_count       每条失败：名称 + 失败率 + 5 个样例主键
   fail_rate / sample_keys  ⭐        和上一批次的对比箭头
   run_id / commit_sha / 数据版本     趋势小图（近 30 批次）

   系统：消费金融的申请评分卡（A 卡）
   规模：日均申请 4.2 万笔，46 个特征来自离线数仓的用户行为宽表
   门禁：已有 38 条质量断言，覆盖完整性 / 唯一性 / 有效性 / 一致性
        ⚠️ 时效性：0 条         准确性：0 条

   那六天里，38 条断言的运行结果：

     行数           200,143 行   ← 和昨天完全一样，波动 0.00%   ✅
     主键唯一       0 重复                                      ✅
     非空率         99.71%      ← 和昨天完全一样                ✅
     枚举合法       全部命中白名单                              ✅
     跨字段一致性   0 违规                                      ✅
     分区存在性     dt=2024-03-16 存在                          ✅

   ⭐ 38/38 全绿。而 max(event_time) 从 3 月 11 日起就再没动过。

   ① 六个维度只实现了四个 —— 时效性一条都没有
      团队认为"分区存在性检查"就是时效性检查
      💀 分区存在只证明【任务跑完了】，不证明【数据是新的】
         这两件事之间隔着整个事故

   ② 行数波动断言设的是 ±20%
      而"复制昨天"这种故障，行数波动恰好是 0.00%
      💀 它是所有断言里【最绿】的那一条 ⭐
         异常检测的盲区不是"偏差太大"，是"偏差为零"

   ③ 模型侧的漂移监控（PSI）纹丝不动：PSI = 0.02
      因为特征分布确实没变 —— 那就是昨天的分布
      ⭐ 漂移监控问的是"分布变没变"，而陈旧数据的分布恰恰【没变】
         PSI 对"数据停止更新"这类故障完全免疫

   3 月 17 日风控日报看出通过率异常，此时已连续 6 天

   通过率        31.4%  →  38.9%   (+7.5 pp)
   多通过        ≈ 1.89 万笔（4.2万 × 6天 × 7.5%）
   实际放款      1.14 万笔，7,760 万元
   首逾(M1)率    这批 8.7%  vs  正常批次 2.9%   (+5.8 pp)
   ⭐ 最终多核销  约 430 万元

   排查耗时也是代价的一部分：
     定位用了 2 天 —— 因为质量报告 38/38 全绿，
     所有人的第一反应是查模型和策略，没人怀疑数据 💀
     回溯重跑 6 天分区 + 重训模型，又用了 11 天

   Day 1  列出【进模型的前 20 个字段】，只管这 20 个
          ⚠️ 不要从"给所有表加断言"开始，那个项目永远做不完

   Day 2  每个字段写 3 条：非空率、范围/枚举、时效性
          全部先设成 WARN，一条 BLOCK 都不要有  ⭐

   Day 3-7  只跑不拦，收集 7 天的实际分布
          ⭐ 阈值从这 7 天的 P99 × 1.5 算出来，不要在 Day 2 就拍数字

   第 2 周  把其中"确定性错误"那几条升级为 BLOCK
          （主键重复、类型变更、数据年龄超 24h）
          其余永远留在 WARN

   之后    每次线上事故复盘，问一句：
          "哪一条断言本可以拦住它？" → 加上去
          ⭐ 断言集应该由事故驱动生长，不是由想象力驱动