import numpy as np

rng = np.random.default_rng(0)

def make_clean(n=50_000):
    """一份干净数据。⭐ 关键不是数据本身，是你【知道每一列该长什么样】"""
    return {
        "user_id": np.arange(1, n + 1),
        "age":     rng.integers(18, 80, n),
        "amount":  np.round(rng.lognormal(6, .8, n), 2),
        "city":    rng.choice(["北京", "上海", "广州", "深圳", "杭州"], n),
        "ts":      np.datetime64("2026-01-01") +
                   rng.integers(0, 180 * 86400, n).astype("timedelta64[s]"),
    }

def contaminate(d, kinds, rate=0.02, seed=1):
    """污染器：注入若干种脏，同时返回【真值台账】—— 这本账是整个项目的判卷依据"""
    r = np.random.default_rng(seed)
    d = {k: v.copy() for k, v in d.items()}
    n, truth = len(d["user_id"]), {}
    pick = lambda: r.choice(n, int(n * rate), replace=False)
    if "单位不一致" in kinds:                       # ⭐ 一部分金额从"元"变成"分"
        i = pick(); d["amount"] = d["amount"].astype(float); d["amount"][i] *= 100
        truth["单位不一致"] = set(i.tolist())
    if "不可能值" in kinds:
        i = pick(); d["age"] = d["age"].astype(float); d["age"][i] = 200
        truth["不可能值"] = set(i.tolist())
    if "空值伪装" in kinds:                         # ⭐ 不是 None，是字符串"未知"
        i = pick(); d["city"] = d["city"].astype(object); d["city"][i] = "未知"
        truth["空值伪装"] = set(i.tolist())
    return d, truth

import numpy as np

SENTINEL = {"", "NULL", "null", "N/A", "-", "未知", "无", "-999", "0000-00-00"}

def detect(d):
    hits = {}
    a = np.asarray(d["age"], float)
    hits["不可能值"] = set(np.where((a < 0) | (a > 120))[0].tolist())      # 业务范围断言
    c = np.asarray(d["city"], object)
    hits["空值伪装"] = set(np.where(np.isin(c, list(SENTINEL)))[0].tolist())
    m = np.asarray(d["amount"], float)
    q1, q3 = np.percentile(m, [25, 75])
    hits["单位不一致"] = set(np.where(m > q3 + 50 * (q3 - q1))[0].tolist())  # ⭐ 量级跳变
    return hits

def score(truth, hits):
    for k, t in truth.items():                    # ⭐⭐ 召回和精确必须一起报
        h = hits.get(k, set()); tp = len(t & h)
        print(k, f"召回={tp/len(t):.2f} 精确={tp/max(len(h), 1):.2f}")

   💀 「我的检测器召回全是 1.00」
      → 你八成在检测器里 import 了污染器的常量。分开两个文件重写。

   💀 「误报率没测」
      → 只在被污染的数据上评估，等于只算召回不算精确。
        ⭐ 必须留一份【完全没污染】的对照集，跑一遍看误报。

   💀 「用 describe() 当检测器」
      → 均值/分位数抓得到数值类的脏，抓不到编码错乱、类别漂移、跨列不一致。
        ⭐ 十种形态需要至少四类不同的检查手段。

import sys, numpy as np

CHECKS = []                                    # ⭐ 断言注册表：每条都带维度和等级
def check(dim, level="hard"):
    def deco(fn): CHECKS.append((dim, level, fn.__name__, fn)); return fn
    return deco

@check("完整性")
def 关键列非空(d):
    bad = int(np.sum([x is None or x == "" for x in d["city"]]))
    return bad / len(d["city"]) <= 0.001, f"空值率 {bad/len(d['city']):.4f}"

@check("唯一性")
def 主键唯一(d):
    return len(set(d["user_id"].tolist())) == len(d["user_id"]), "主键重复"

@check("分布稳定性", level="warn")             # ⭐ 分布类默认只 warn，别一上来就 hard
def 金额PSI(d, base_q=None):
    q = np.percentile(d["amount"], [10, 50, 90])
    return True, f"分位数 {q.round(1).tolist()}"

def run_gate(d):
    hard_fail = 0
    for dim, level, name, fn in CHECKS:
        ok, msg = fn(d)
        if not ok:
            print(f"[{'FAIL' if level=='hard' else 'WARN'}] {dim}/{name}: {msg}")
            hard_fail += (level == "hard")      # ⭐⭐ 只有 hard 才决定退出码
    sys.exit(1 if hard_fail else 0)             # ⭐ CI 靠这个退出码变红

   💀 「阈值是拍脑袋定的」
      → 必须用历史数据回测：把过去 14 天逐日跑一遍，
        看每条断言的取值分布，阈值放在【历史 p99 之外】而不是"感觉"上。

   💀 「门禁只跑在抽样上，线上还是炸了」
      → 唯一性、跨列一致性这两类【不能抽样】——
        抽样恰好会把重复和边缘不一致漏掉。 ⭐

   💀 「红了以后没人知道该干嘛」
      → 每条断言必须带三样东西：负责人、样例坏行、修复动作。
        只报"完整性检查失败"的门禁等于没有。

import numpy as np

def kappa(M):
    """M 是混淆矩阵（两遍标注的交叉表）。⭐ p0 和 pe 必须一起报"""
    M = np.asarray(M, float); n = M.sum()
    p0 = np.trace(M) / n                                  # 观察一致率
    pe = (M.sum(0) * M.sum(1)).sum() / n ** 2             # ⭐ 瞎猜也能达到的一致率
    return round(p0, 3), round(pe, 3), round((p0 - pe) / (1 - pe), 3)

print(kappa([[26, 7, 1], [9, 20, 8], [1, 7, 21]]))    # 第一版指南 → (0.67, 0.335, 0.504)
print(kappa([[31, 3, 0], [4, 29, 4], [0, 3, 26]]))    # 修订后     → (0.86, 0.336, 0.789)
print(kappa([[92, 3], [3, 2]]))                       # ⭐⭐ 陷阱   → (0.94, 0.905, 0.368)

   💀 「两遍间隔两小时，κ = 0.93」
      → 你在背答案。⭐ 至少隔 48 小时，并且【打乱顺序 + 隐藏第一遍结果】。

   💀 「只报一致率不报 Kappa」
      → 见上面那个 0.94 / 0.368 的例子。

   💀 「指南越写越长，κ 反而掉了」
      → 常见且真实。⭐ 修订的正确方向是【把模糊的定义换成可判定的规则 + 边界样例】，
        不是【加更多描述性形容词】。规则要能让人回答"是/否"，而不是"感觉像"。

   目标：做出一个【离线 AUC 0.99、去掉泄漏后掉到 0.72】的完整例子，
        并把"怎么发现的"写成一份复盘。

import numpy as np
from sklearn.ensemble import HistGradientBoostingClassifier
from sklearn.metrics import roc_auc_score

rng, N = np.random.default_rng(11), 60_000
tenure  = rng.exponential(180, N)          # 在网天数
mau     = rng.poisson(12, N)               # 月活跃天数
tickets = rng.poisson(0.4, N)              # 工单数
plan    = rng.choice([0, 1, 2], N, p=[.6, .3, .1])
logit   = -1.4 - .005 * tenure - .115 * mau + .45 * tickets + .5 * plan
y = (rng.random(N) < 1 / (1 + np.exp(-logit))).astype(int)      # 是否流失，约 5.6%

# ⭐ 泄漏一：后果特征 —— 退款工单，只有【已经流失的人】才会开
refund = rng.poisson(np.where(y == 1, 1.4, 0.02))
# ⭐ 泄漏二：时间穿越 —— 账户状态最后更新距今天数，在流失【之后】被系统改写
gap    = np.where(y == 1, rng.gamma(2, 3, N), rng.gamma(2, 30, N))

base = np.c_[tenure, mau, tickets, plan]
tr   = np.arange(N) < 45_000
def auc(X, name):
    m = HistGradientBoostingClassifier(max_iter=150, random_state=0).fit(X[tr], y[tr])
    a = roc_auc_score(y[~tr], m.predict_proba(X[~tr])[:, 1])
    print(f"{name:16s} AUC={a:.4f}"); return a

auc(np.c_[base, refund, gap], "两个泄漏都在")   # 0.9899
auc(np.c_[base, gap],         "只有时间穿越")   # 0.9789
auc(np.c_[base, refund],      "只有后果特征")   # 0.9237
auc(base,                     "清洁版")        # 0.7168   ⭐⭐ 这才是真实能力

   💀 「我只用随机切分，指标一直很漂亮」
      → 时间穿越型泄漏在随机切分下【也是高分】。
        ⭐ 时间切分是发现它的最低成本手段，也是这个项目的第一道检查。

   💀 「删了泄漏特征，AUC 还是 0.95」
      → 你漏了它的衍生特征（refund 的对数、gap 的分箱、
        以及任何用它算出来的交叉特征）。⭐ 按【血缘】删，不是按列名删。

   💀 「用特征重要性找泄漏」
      → 重要性高 ≠ 泄漏，重要性低 ≠ 干净。
        ⭐⭐ 重要性只能帮你【排查顺序】，判定必须回到"生成时刻"这一条。

   💀 「0.72 太难看了，我想把它调上去」
      → ⭐ 这正是这个项目要治的病。0.72 是真的，0.99 是假的。
        一个项目组愿不愿意接受这一点，比任何技术手段都重要。

# 数据事故 / 实验：____
## 我注入了什么（真值台账）
## 第一个发现它的信号是什么，来自哪个检查
## 我用了哪几种手段，各花了多久
## 没能发现的部分 —— 为什么？需要什么信息才能发现？   ← 最有价值的一节
## 修复后的真实指标 vs 修复前的假指标
## 要加什么，才能让下次在【进模型之前】就被拦住