消融、统计显著性与有效性威胁

mkdir l7-ablation
cd l7-ablation

"""对固定资料选择器做配对消融，枚举置换与 bootstrap 重采样。"""

from itertools import product
from math import ceil
from statistics import mean

# 候选顺序模拟召回结果；正确答案标签只供评分，不传给选择器。
CASES = [
    {"id": "q1", "product": "book", "expected": 30,
     "docs": [("food", 0), ("book", 30)]},
    {"id": "q2", "product": "book", "expected": 7,
     "docs": [("food", 0), ("book", 7)]},
    {"id": "q3", "product": "hotel", "expected": 1,
     "docs": [("food", 0), ("hotel", 1)]},
    {"id": "q4", "product": "ticket", "expected": 2,
     "docs": [("food", 0), ("ticket", 2)]},
    {"id": "q5", "product": "book", "expected": 14,
     "docs": [("book", 14), ("food", 0)]},
    # 第六条的正确文档商品元数据误写，重排会选到另一份旧政策。
    {"id": "q6", "product": "book", "expected": 30,
     "docs": [("misc", 30), ("book", 7)]},
]


def choose(product_name, docs, rerank):
    if rerank:
        ranked = sorted(docs, key=lambda doc: doc[0] == product_name, reverse=True)
        return ranked[0][1]
    return docs[0][1]


def paired_scores(cases):
    if len({c["id"] for c in cases}) != len(cases):
        raise ValueError("duplicate paired sample")
    rows = []
    for case in cases:
        full = int(choose(case["product"], case["docs"], True) == case["expected"])
        removed = int(choose(case["product"], case["docs"], False) == case["expected"])
        rows.append((case["id"], full, removed, full - removed))
    return rows


def validate(differences):
    if not 1 <= len(differences) <= 6:
        raise ValueError("enumeration supports 1 to 6 pairs")


def permutation_p(differences):
    validate(differences)
    observed = abs(sum(differences))
    # 每一对交换 full/removed 等价于让差值乘 +1 或 -1。
    values = [abs(sum(sign * diff for sign, diff in zip(signs, differences)))
              for signs in product((-1, 1), repeat=len(differences))]
    return sum(value >= observed for value in values) / len(values)


def bootstrap_interval(differences):
    validate(differences)
    n = len(differences)
    # 从完整配对差值有放回抽 n 次；枚举 n^n 种索引，保留配对。
    estimates = sorted(mean(differences[i] for i in indices)
                       for indices in product(range(n), repeat=n))
    def quantile(p):
        return estimates[ceil(p * len(estimates)) - 1]
    return quantile(.025), quantile(.975)


def main():
    rows = paired_scores(CASES)
    for identifier, full, removed, diff in rows:
        print(f"{identifier} full={full} removed={removed} difference={diff:+d}")
    differences = [row[3] for row in rows]
    low, high = bootstrap_interval(differences)
    p = permutation_p(differences)
    print(f"paired_n={len(rows)} full_rate={mean(r[1] for r in rows):.4f} "
          f"removed_rate={mean(r[2] for r in rows):.4f}")
    print(f"mean_effect={mean(differences):.4f} bootstrap_95pct=[{low:.4f}, {high:.4f}]")
    print(f"two_sided_exact_p={p:.4f} alpha=0.0500 significant={p < .05}")


if __name__ == "__main__":
    main()

ls
python -X utf8 ablation_lab.py

q1 full=1 removed=0 difference=+1
q2 full=1 removed=0 difference=+1
q3 full=1 removed=0 difference=+1
q4 full=1 removed=0 difference=+1
q5 full=1 removed=1 difference=+0
q6 full=0 removed=1 difference=-1
paired_n=6 full_rate=0.8333 removed_rate=0.3333
mean_effect=0.5000 bootstrap_95pct=[-0.1667, 1.0000]
two_sided_exact_p=0.3750 alpha=0.0500 significant=False

ls
python -X utf8 -m unittest -v test_ablation_lab.py