06 · 重复与实体解析

   ① 精确重复      两行的每一个字节都一样
      来源：任务重跑、回填、多次导出、上游 at-least-once 投递
      检测：哈希，一行代码，O(n)                          ⭐ 简单

   ② 近重复        同一条内容的不同版本
      文本：改了标点 / 加了后缀 / 换了排版
      图片：改了尺寸 / 压缩过一次 / 加了水印
      数值：浮点尾数差一点、时间戳差 1 秒
      检测：SimHash / MinHash / 感知哈希，O(n) ~ O(n log n)  ⚠️ 要定阈值

   ③ 同一实体多条记录   "张三" 和 "张 三" 和 "ZHANG SAN"
      来源：拼写差异、别名、ID 变更、多系统合并、用户重新注册
      检测：阻塞 + 相似度 + 判定 + 传递闭包                 💀 最难
      —— 这一层没有"正确答案"，只有"你定义的正确答案"

   随机切分之后：验证集里有 41.5% 的行，在训练集里存在一个孪生副本  ⭐

   验证集里【有孪生副本】的那部分：AUC = 1.0000   ⭐⭐ 完美
   验证集里【没有孪生副本】的那部分：AUC = 0.8788

   —— 前者不是"预测"，是"检索"。
      模型只是在训练集里找到了那条几乎一模一样的记录，然后把它的标签抄了下来。

import hashlib
import pandas as pd

def exact_dup_report(df: pd.DataFrame, key_cols=None, round_float=6):
    """精确重复检测。返回 (重复行数, 重复组数, 最大组的大小)。"""
    cols = key_cols or list(df.columns)
    x = df[cols].copy()
    for c in x.columns:                      # ⭐ 浮点必须先定精度，否则 1e-16 的差就不"相同"
        if pd.api.types.is_float_dtype(x[c]):
            x[c] = x[c].round(round_float)
    # ⚠️ NaN != NaN，必须【先】填占位串【再】转字符串，否则全空行永远不算重复
    s = x.fillna("<NA>").astype(str).agg("\x1f".join, axis=1)
    h = s.map(lambda v: hashlib.blake2b(v.encode(), digest_size=16).hexdigest())
    g = h.value_counts()
    dup_groups = g[g > 1]
    return int((dup_groups - 1).sum()), int(len(dup_groups)), int(g.max())

# 三个必查的对象，缺一不可：
#   1) 全字段重复      —— 纯粹的技术性重复
#   2) 业务主键重复    —— (dt, user_id, item_id) 重了 = 上游违约（第 3 章）
#   3) 去掉时间戳之后重复 ⭐ —— 「同一件事被记录了两次，只是时间差 1 秒」

   精确 MD5：找到 0 对  💀  —— 连"只多了一个尾部空格"的那对都抓不到

   MinHash（字符 3-gram，128 个排列）的相似度：
     1.000  (0,3)  同一条，只差一个尾空格            真重复
     0.812  (0,1)  "iPhone 15" vs "iPhone15"        真重复
     0.594  (0,4)  256G 原色  vs  512G 蓝色         ⚠️ 不是重复！
     0.492  (0,2)  加了「（）」和「全网通」          真重复
     0.445  (5,6)  "小米 14 Ultra" vs "小米14Ultra"  真重复

import hashlib
import numpy as np

def shingles(s, k=3):
    """中文用【字符 n-gram】，不要分词——分词器本身会引入不一致。"""
    s = "".join(s.split()).lower()
    return {s[i:i + k] for i in range(max(1, len(s) - k + 1))}

def minhash(text, num_perm=128, seed=1):
    rng = np.random.default_rng(seed)
    A = rng.integers(1, 2 ** 31 - 1, num_perm)
    B = rng.integers(0, 2 ** 31 - 1, num_perm)
    P = (1 << 31) - 1
    h = np.array([int(hashlib.md5(g.encode()).hexdigest()[:8], 16)
                  for g in shingles(text)])
    return ((A[None, :] * h[:, None] + B[None, :]) % P).min(axis=0)  # ⭐ 每个排列取最小

def jaccard_est(sig_a, sig_b):
    return float((sig_a == sig_b).mean())     # 签名相等的比例 ≈ Jaccard 相似度

# 图片用感知哈希（pHash/dHash）：缩到 8x8 灰度 → DCT → 比中位数 → 64 bit
#   汉明距离 <= 5 基本可判近重复；改尺寸、轻度压缩、加水印都扛得住

   全量两两            2.000e+10 对     （两百亿次比较，跑不完）
   按【城市】阻塞       5.883e+07 对     降到 1/339
   按【城市+姓氏首字母】 2.262e+06 对     降到 1/8842      ⭐

   ⚠️ 阻塞键的代价：任何一个阻塞键写错的记录，
      永远不会和它的真配对相遇 —— 阻塞是【只降召回，不降精确】的一步
   ⭐ 所以工程上用【多个阻塞键取并集】：城市 / 手机号后4位 / 姓名首字母 / 地址前6字
      任何一个键对上，就进候选池

import collections

class DSU:
    """并查集：把两两判定的结果合并成实体簇。"""
    def __init__(self, n): self.p = list(range(n))
    def find(self, x):
        while self.p[x] != x:
            self.p[x] = self.p[self.p[x]]; x = self.p[x]
        return x
    def union(self, a, b):
        a, b = self.find(a), self.find(b)
        if a != b: self.p[a] = b

def blocked_pairs(records, key_funcs):
    """多个阻塞键取并集，产出候选对。records: [dict]，key_funcs: [callable]"""
    cand = set()
    for kf in key_funcs:
        buckets = collections.defaultdict(list)
        for i, r in enumerate(records):
            k = kf(r)
            if k is not None:
                buckets[k].append(i)
        for idx in buckets.values():
            if len(idx) > 200:      # ⭐ 超大桶直接丢弃：它一定是个无信息的键（比如"未知城市"）
                continue
            for a in range(len(idx)):
                for b in range(a + 1, len(idx)):
                    cand.add((idx[a], idx[b]))
    return cand

# 用法：
#   cand = blocked_pairs(recs, [lambda r: r["city"], lambda r: r["phone"][-4:]])
#   dsu = DSU(len(recs))
#   for i, j in cand:
#       if score(recs[i], recs[j]) >= THRESHOLD: dsu.union(i, j)

   ① 归一化    trim / NFKC / 大小写 / 去标点 / 统一编码（第 4 章）
               ⚠️ 不做这步，"北京" 和 "北京 " 永远是两条

   ② 去重      精确 → 近重复 → 实体解析

   ③ 切分      训练 / 验证 / 测试
               ⭐⭐ ②和③的顺序绝对不能反。先切再去重 = 等于没做

   ④ 采样/加权  如果重复反映了真实的业务分布，用【权重】表达，不要用【复制】

import numpy as np
from sklearn.model_selection import GroupKFold

def entity_split(entity_ids, test_size=0.3, seed=0):
    """按实体切分：同一实体的所有记录只能落在同一侧。⭐ 这是本章最重要的一段代码"""
    ids = np.asarray(entity_ids)
    uniq = np.unique(ids)
    rng = np.random.default_rng(seed)
    rng.shuffle(uniq)
    test_ids = set(uniq[:int(len(uniq) * test_size)].tolist())
    is_test = np.array([i in test_ids for i in ids])
    return ~is_test, is_test

def leak_probe(g_train, g_valid):
    """上线前必跑：验证集里有多少行，在训练集里存在同实体的副本。期望是 0。"""
    overlap = np.isin(np.asarray(g_valid), np.unique(g_train))
    return float(overlap.mean())    # ⭐ 实跑里这个数是 41.5%，对应 +0.16 的虚高 AUC

# 交叉验证同理：GroupKFold(n_splits=5).split(X, y, groups=entity_ids)

   系统：某电商的商品图文违规审核模型（图片 + 标题）
   数据：运营历史处置记录，约 240 万条，人工标注过

   ⚠️ 数据来源有两条：
      · 主动巡检导出        （每周一次全量导出）
      · 用户举报工单        （实时流）
      同一个违规商品，常常两边都有 —— 而且标题被商家反复改过，
      每改一次就是一条新记录：只差一个 emoji、一个空格、一个"正品"后缀

   去重代码只有一行：df.drop_duplicates(subset=["item_id", "title", "img_md5"])
   —— 精确匹配。标题差一个字符、图片重新压缩过一次，就抓不到。

   ✅ 精确重复率显示 0.4%，看起来干干净净
   ✅ 数据集大小 240 万，行数监控正常
   ✅ 离线指标一路上涨：新模型 recall@P90 = 0.871，旧模型 0.803
   ✅ A/B 实验也"赢"了 —— 因为实验组和对照组用的是同一份污染的评测集 💀

   💀 真实情况：
      精确重复 0.4%，但【近重复】达到 3.1%（图片 pHash 汉明距离 <= 5
      且标题 MinHash >= 0.75），实体级（同一个商品的多次改标题）达到 11.6%
      随机切分之后，验证集里 9.8% 的样本在训练集有近重复孪生

   模型上线全量之后：
      线上 recall（人工抽检 5000 条复核）  0.871 → 0.642
      而旧模型在同一批抽检上是            0.658
      —— 新模型实际上比旧模型【略差】，却在离线和 A/B 上赢了两个月 ⭐

   期间：
      · 漏放违规商品约 1.9 万件，其中 340 件触发平台处罚，罚款 88 万元
      · 团队按"新模型更好"的结论砍掉了两条规则兜底策略，事后全部回滚
      · 最贵的其实不是钱：两个月里所有基于这份评测集做的 17 次实验结论
        全部作废，需要重跑  💀

   发现方式：一位新同事在看 bad case 时发现，
            某条验证集样本的标题在训练集里出现过 6 次 ⭐
   修复：pHash + MinHash 重建实体 id → 按实体重切 → 重跑全部实验 = 4 周