04 · 脏数据的十种形态

   会报错（好脏数据）              静默通过（坏脏数据）💀
   ─────────────────              ────────────────────
   ① 类型不一致   （有时会）        ③ 时区
   ② 字符编码     （有时会）        ④ 单位
                                   ⑤ 字段截断
                                   ⑥ 默认值伪装成真值
                                   ⑦ 枚举漂移
                                   ⑧ 前后空格与全半角
                                   ⑨ 嵌套 JSON 结构变了
                                   ⑩ 同名不同义

   ⭐ 大多数人的时间分配是反的：
      90% 的排查精力花在 ①②（因为它们会叫、会红、会有堆栈）
      而 90% 的实际损失来自 ③–⑩（因为它们不叫）

import unicodedata
import pandas as pd

SENT_NUM = {0, -1, -99, -999, -9999, 9999, 99999, 999999, 2147483647, 32767, 65535}
SENT_STR = {"", "-", "null", "none", "nan", "n/a", "未知", "无", "其他", "默认", "test"}
SENT_DAY = {"1970-01-01", "1900-01-01", "0000-00-00", "2099-12-31", "9999-12-31"}
TRUNC_LEN = {16, 32, 50, 64, 100, 128, 200, 255, 256, 512, 1000, 1024}


def scan_dirty(df: pd.DataFrame, enum_spec=None, top_ratio=0.01):
    """十种形态里能被代码发现的那几种，一次扫完。返回 (级别, 列, 形态, 详情)。"""
    v = []
    for c in df.columns:
        s = df[c].dropna()
        if len(s) == 0:
            v.append(("P1", c, "⑥ 整列为空", "0 行非空")); continue

        # ⑥ 默认值伪装成真值：最高频单值占比 ⭐ 一条规则抓住绝大多数哨兵值
        vc = s.value_counts()
        top, share = vc.index[0], vc.iloc[0] / len(s)
        looks_sent = (top in SENT_NUM) or (str(top).strip().lower() in SENT_STR) \
                     or (str(top)[:10] in SENT_DAY)
        if share > 0.30 or (looks_sent and share > top_ratio):
            v.append(("P1" if looks_sent else "P2", c, "⑥ 疑似默认值/哨兵值",
                      f"{top!r} 占 {share:.2%}"))

        if s.map(type).nunique() > 1:      # ① 类型不一致
            kinds = s.map(lambda x: type(x).__name__).value_counts().to_dict()
            v.append(("P1", c, "① 类型不一致", str(kinds)))

        if s.dtype == object:
            t = s.astype(str)
            # ② 编码：替换符 / BOM / 零宽空格
            bad = t.str.contains("�||​", regex=True).mean()
            if bad > 0:
                v.append(("P0", c, "② 编码损坏或不可见字符", f"{bad:.2%} 的行"))
            # ⑧ 空格与全半角：归一化后类别数掉了多少 ⭐
            norm = t.map(lambda x: unicodedata.normalize("NFKC", x).strip())
            if norm.nunique() < t.nunique():
                v.append(("P1", c, "⑧ 空格/全半角未归一",
                          f"类别数 {t.nunique()} → {norm.nunique()}"))
            # ⑤ 截断：长度直方图在整数边界上出现尖峰
            L = t.str.len()
            peak = L.value_counts(normalize=True)
            for cand in TRUNC_LEN:
                if peak.get(cand, 0) > 0.005 and cand >= L.median():
                    v.append(("P0", c, "⑤ 疑似字段截断",
                              f"长度恰好={cand} 的行占 {peak[cand]:.2%}"))
            # ① 半数能转数值 = 混合列
            num_ok = pd.to_numeric(t, errors="coerce").notna().mean()
            if 0.02 < num_ok < 0.98:
                v.append(("P2", c, "① 数值/文本混合列", f"可转数值 {num_ok:.1%}"))
            # ⑦ 枚举漂移
            if enum_spec and c in enum_spec:
                unknown = set(norm.unique()) - set(enum_spec[c])
                if unknown:
                    v.append(("P1", c, "⑦ 未知枚举值",
                              f"{sorted(unknown)[:5]} 占 {norm.isin(unknown).mean():.2%}"))
        elif pd.api.types.is_numeric_dtype(s):
            # ④ 单位：分位数比均值稳，作为跨天对比的基准量
            q = s.quantile([0.5, 0.99]).to_dict()
            v.append(("P3", c, "④ 单位基准（存起来跨天比）", f"p50={q[0.5]:.4g} p99={q[0.99]:.4g}"))
    return pd.DataFrame(v, columns=["级别", "列", "形态", "详情"])

# 用法：report = scan_dirty(df, enum_spec={"channel": ["app_home", "h5"]})
#      把今天的 P3 基准行存下来，明天和今天比 —— 单位跳变就是这么抓的 ⭐

import numpy as np

def hour_shift(hours_now, hours_ref):
    """按小时直方图找整点位移。hours_*: 0-23 的整数数组（事件发生的小时）。
    返回 (最可能的位移小时数, 该位移下的相关系数, PSI)。"""
    a = np.bincount(np.asarray(hours_ref) % 24, minlength=24).astype(float)
    b = np.bincount(np.asarray(hours_now) % 24, minlength=24).astype(float)
    a, b = a / a.sum() + 1e-9, b / b.sum() + 1e-9
    psi = float(np.sum((b - a) * np.log(b / a)))          # ⭐ PSI > 0.25 = 重大偏移
    cor = [float(np.corrcoef(a, np.roll(b, -s))[0, 1]) for s in range(24)]
    k = int(np.argmax(cor))
    return k, cor[k], psi

# 实跑（200 万条事件，人为把时间戳整体 +8 小时）：
#   按【天】看：总量一模一样，日活曲线毫无异常
#   按【小时】看：PSI = 1.457，位移 0 的相关系数 = -0.465，
#                位移 8 的相关系数 = 1.000   ⭐⭐ 一眼就指出「差了 8 小时」

   系统：某内容 App 的埋点主表 dwd_event_di，下游 40+ 个消费者
   字段：event_ts   bigint   not null   （契约里就写了这三样）

   埋点 SDK v4.2 上线，作者做了一个"体验优化"：
       把 event_ts 从【UTC 毫秒】改成【本地时间毫秒】（东八区，+8h）
       理由是"这样在日志文件里肉眼可读，排查方便"
   字段名没变、类型没变、非空率没变、数量级没变  ✅✅✅✅

   ⚠️ 而且它是【灰度】上线的：3 周内从 5% 放量到 100%

   ✅ 契约校验全绿：bigint、not null、行数在区间内
   ✅ 时间戳范围检查也全绿 —— 规则是"必须落在过去 30 天内"，
      差 8 小时在这个规则里根本不算什么 💀
   ✅ 日活曲线正常、总事件量正常、缺失率 0

   💀 灰度把断崖变成了缓坡：
      3 周从 5% 到 100%，指标是【每天挪一点点】
      拐点检测按天跑，形状是"缓慢上升" → 被判定成【漂移】而不是【故障】
      于是走了"重训模型"的流程，而不是"排查上游"的流程  ⚠️

   受影响的是"跨午夜"那一段：22:00–23:59 的事件占全天的 11.3%，
   它们被整体划进了第二天

   · 「次日留存」从 31.2% 虚高到 35.3%   （+4.1pp）
   · 运营据此把 push 时间从 20:00 挪到 22:30 —— 因为"数据显示晚间用户更活跃"
     实际结果：7 日留存掉了 0.9pp，两个月后才回滚
   · 流失预测模型的标签（次日是否回访）有 6.8% 直接翻转
     线上 recall 从 0.62 掉到 0.51

   发现延迟：5 周
   发现方式：财务对账时发现按小时的 GMV 曲线有【两个午夜峰】⭐
             —— 因为新旧 SDK 版本在灰度期同时在跑
   修复：回溯 38 天日志重算分区 + 重训 = 11 人日 + 4.2 万核时的全量回刷

   ① 先跑 scan_dirty，只看 P0/P1                       5 分钟
      —— 截断、编码、类型混合、哨兵值，这几类几乎一定有

   ② 每个字符串列打印【长度直方图】和【Top-20 取值】     10 分钟 ⭐
      —— 肉眼扫一遍。这一步的产出率高得离谱，
         因为人对"这个值出现 8% 不对劲"的直觉比任何规则都准

   ③ 每个数值列打印 p1 / p50 / p99 和最高频单值          5 分钟
      —— 单位问题和哨兵值都在这三个数字里

   ④ 时间列：画按小时的分布 + 和上周做循环互相关          5 分钟 ⭐⭐

   ⑤ 每个 join：先算交集率，低于 95% 就停下来问人         5 分钟
      —— ⑩ 同名不同义只能在这一步被抓到