08 · 数据质量怎么量

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}