"""第 46 章：重新分析真实预测记录，不重新训练分类器。""" from argparse import ArgumentParser import json from pathlib import Path import platform import time import numpy as np import scipy from scipy.special import logsumexp import reproduce_calibration as method HERE = Path(__file__).resolve().parent SOURCE = HERE / "data/calibration-reproduction.json" PLAN = HERE / "data/calibration-evidence-plan.json" def losses(logits, labels, temperature=1.0): z, y = method.checked(logits, labels, temperature) return logsumexp(z, axis=1) - z[np.arange(len(y)), y] def paired_bootstrap(before, after, indices): """同一个索引同时抽取一对结果；保留逐条差值，不把两组拆开。""" a, b = np.asarray(before), np.asarray(after) if a.ndim != 1 or a.shape != b.shape or len(a) == 0: raise ValueError("两种方法需要同一批、按相同 ID 排列的逐条损失") delta = b - a means = delta[indices].mean(axis=1) low, high = np.quantile(means, [.025, .975], method="linear") return {"mean": float(delta.mean()), "interval": [float(low), float(high)], "per_item_delta": delta.tolist(), "resampled_means": means.tolist()} def subset_orders(ids, repeats, seed): rng = np.random.default_rng(seed) return [rng.permutation(len(ids)) for _ in range(repeats)] def brief(values): a = np.asarray(values) return {"median": float(np.median(a)), "min": float(a.min()), "max": float(a.max())} def source_checks(source): if method.digest(source["numeric"]) != source["numeric_sha256"]: raise ValueError("第 45 章数值记录与指纹不一致") if method.file_hash(method.__file__) != source["source_sha256"]: raise ValueError("第 45 章方法代码已变化，请先核对实验版本") groups = list(source["numeric"]["splits"].values()) ids = [i for group in groups for i in group] if len(ids) != len(set(ids)): raise ValueError("数据划分中出现重复 ID") def run(source, plan): source_checks(source) start = time.perf_counter() numeric = source["numeric"] cy = np.asarray(numeric["labels"]["calibration"], dtype=int) ty = np.asarray(numeric["labels"]["test"], dtype=int) cal_ids, test_ids = numeric["splits"]["calibration"], numeric["splits"]["test"] if (sorted(set(plan["sizes"])) != plan["sizes"] or plan["sizes"][0] < 1 or plan["sizes"][-1] != len(cy)): raise ValueError("样本量必须递增且最后包含完整校准集") orders = subset_orders(cal_ids, plan["subset_repeats"], plan["subset_seed"]) bootstrap_ids = np.random.default_rng(plan["bootstrap_seed"]).integers( len(ty), size=(plan["bootstrap_repeats"], len(ty))) runs, full, sensitivity = [], [], [] for original in numeric["results"]: seed = original["seed"] cz = np.asarray(original["logits"]["calibration"]) tz = np.asarray(original["logits"]["test"]) base_loss = losses(tz, ty) for size in plan["sizes"]: # 完整 269 条只有同一份集合，不冒充 20 次不同的抽样。 chosen = [np.arange(len(cy))] if size == len(cy) else [p[:size] for p in orders] for repeat, positions in enumerate(chosen): t = method.fit_temperature(cz[positions], cy[positions], plan["temperature_bounds"]) after = method.metrics(tz, ty, t, bins=15) runs.append({ "seed": seed, "size": size, "repeat": repeat, "calibration_ids": [cal_ids[i] for i in positions], "class_counts": np.bincount(cy[positions], minlength=10).tolist(), "calibration_errors": int((cz[positions].argmax(1) != cy[positions]).sum()), "temperature": t, "at_boundary": t in plan["temperature_bounds"], "fit_nll_before": method.nll(cz[positions], cy[positions]), "fit_nll_after": method.nll(cz[positions], cy[positions], t), "test_nll": after["nll"], "test_ece": after["ece"], "test_accuracy": after["accuracy"], "delta_nll": float(after["nll"] - base_loss.mean()), }) if size == len(cy): # 独立重新拟合；应与上一章保存的完整校准温度一致。 if abs(t - original["temperature"]) > 1e-8: raise ValueError("完整校准温度与第 45 章不一致") pair = paired_bootstrap(base_loss, losses(tz, ty, t), bootstrap_ids) full.append({"seed": seed, "temperature": t, "bootstrap": pair}) for bins in plan["ece_bins"]: sensitivity.append({ "seed": seed, "bins": bins, "before": method.metrics(tz, ty, 1, bins), "after": method.metrics(tz, ty, t, bins)}) summaries = [] for seed in [r["seed"] for r in numeric["results"]]: for size in plan["sizes"]: group = [r for r in runs if r["seed"] == seed and r["size"] == size] summaries.append({ "seed": seed, "size": size, "count": len(group), "temperature": brief([r["temperature"] for r in group]), "delta_nll": brief([r["delta_nll"] for r in group]), "worse_count": sum(r["delta_nll"] > 0 for r in group), "boundary_count": sum(r["at_boundary"] for r in group), "zero_error_count": sum(r["calibration_errors"] == 0 for r in group)}) report = { "plan": plan, "plan_sha256": method.digest(plan), "source_file_sha256": method.file_hash(SOURCE), "source_numeric_sha256": source["numeric_sha256"], "script_sha256": method.file_hash(__file__), "method_sha256": method.file_hash(method.__file__), "environment": {"python": platform.python_version(), "numpy": np.__version__, "scipy": scipy.__version__, "platform": platform.platform()}, "test_ids": test_ids, "subset_orders": [[cal_ids[i] for i in order] for order in orders], "bootstrap_indices_sha256": method.digest(bootstrap_ids.tolist()), "runs": runs, "summaries": summaries, "full": full, "bin_sensitivity": sensitivity, "fit_count": len(runs), "seconds": time.perf_counter() - start} return report def show(report): print(f"固定三份分类器；重新拟合温度 {report['fit_count']} 次；不增加分类器训练。") print("完整校准集：测试 NLL 差值与配对百分位区间（负值表示下降）") for row in report["full"]: b = row["bootstrap"] print(f"seed={row['seed']}：差值 {b['mean']:.5f}，" f"区间 [{b['interval'][0]:.5f}, {b['interval'][1]:.5f}]") print("每个小样本量抽取 20 份；下列为测试 NLL 变差次数：") for seed in [r["seed"] for r in report["full"]]: groups = [r for r in report["summaries"] if r["seed"] == seed and r["count"] > 1] print(f"seed={seed}：" + "；".join( f"n={r['size']} 时 {r['worse_count']}/{r['count']}" for r in groups)) print("分箱敏感性：种子 22 的测试 ECE（温度不重新拟合）") for r in report["bin_sensitivity"]: if r["seed"] == 22: print(f"bins={r['bins']}：{r['before']['ece']:.4f} → {r['after']['ece']:.4f}") print("这是已看过第 45 章测试结果后的探索；抽样区间不覆盖所有研究不确定性。") def main(): parser = ArgumentParser(description=__doc__) parser.add_argument("--out", type=Path) args = parser.parse_args() report = run(json.loads(SOURCE.read_text()), json.loads(PLAN.read_text())) if args.out: args.out.parent.mkdir(parents=True, exist_ok=True) # 留下旧结果：输出文件若存在，要求另选路径。 with args.out.open("x") as stream: stream.write(json.dumps(report, indent=2) + "\n") show(report) if __name__ == "__main__": main()