"""第 47 章：从既有实验记录复核表格，生成研究报告和可复制资料包。只用标准库。""" from argparse import ArgumentParser import hashlib import json import math from pathlib import Path import shutil import calibration_reading as reference HERE = Path(__file__).resolve().parent INPUTS = [ "calibration-reading.json", "calibration-reproduction.json", "calibration-repeat.json", "calibration-evidence.json", "calibration-config.json", "calibration-evidence-plan.json", "calibration-literature.json", ] SCRIPTS = ["calibration_reading.py", "reproduce_calibration.py", "calibration_evidence.py", "research_project.py"] def digest(value): return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(",", ":"), allow_nan=False).encode()).hexdigest() def file_hash(path): return hashlib.sha256(path.read_bytes()).hexdigest() def require(condition, reason): if not condition: raise ValueError(reason) def close(a, b, reason): require(math.isclose(a, b, rel_tol=1e-10, abs_tol=1e-10), reason) def quantile(values, p): ordered = sorted(values) position = (len(ordered) - 1) * p lo = math.floor(position) hi = math.ceil(position) return ordered[lo] + (position-lo) * (ordered[hi]-ordered[lo]) def load(): return {name: json.loads((HERE / "data" / name).read_text()) for name in INPUTS} def audit(records): training = records["calibration-reproduction.json"] evidence = records["calibration-evidence.json"] repeat = records["calibration-repeat.json"] reading = records["calibration-reading.json"] require(training["config"] == records["calibration-config.json"], "训练配置不一致") require(digest(training["config"]) == training["config_sha256"], "训练配置指纹不一致") require(digest(training["numeric"]) == training["numeric_sha256"], "训练数值记录已变化") require(evidence["plan"] == records["calibration-evidence-plan.json"], "扩展方案不一致") require(digest(evidence["plan"]) == evidence["plan_sha256"], "扩展方案指纹不一致") for name, expected in [ ("calibration_reading.py", reading["source_sha256"]), ("reproduce_calibration.py", training["source_sha256"]), ("calibration_evidence.py", evidence["script_sha256"]), ("reproduce_calibration.py", evidence["method_sha256"]), ]: require(file_hash(HERE / name) == expected, f"代码版本不一致：{name}") require(file_hash(HERE / "data/calibration-reproduction.json") == evidence["source_file_sha256"], "扩展分析引用的原始文件已变化") require(evidence["source_numeric_sha256"] == training["numeric_sha256"], "实验关联不一致") require(repeat["same_numeric_sha256"] == training["numeric_sha256"], "重复实验关联不一致") numeric = training["numeric"] ids = [i for group in numeric["splits"].values() for i in group] require(len(ids) == 1797 and set(ids) == set(range(1797)), "四份划分的覆盖或隔离不符") a, b = training["config"]["hidden"] require(training["parameters"] == (64+1)*a + (a+1)*b + (b+1)*10, "参数量不一致") require(evidence["test_ids"] == numeric["splits"]["test"], "扩展测试 ID 不一致") rows = numeric["results"] seeds = training["config"]["seeds"] require([r["seed"] for r in rows] == seeds, "遗漏或重排训练种子") require(len(repeat["runs"]) == 2, "缺少两次独立运行的记录") for run in repeat["runs"]: require(run["numeric_sha256"] == training["numeric_sha256"], "重复运行结果不一致") require(run["weights"] == [r["weights_sha256"] for r in rows], "重复权重记录不一致") labels = numeric["labels"]["test"] for row in rows: chosen = min(row["curves"], key=lambda r: r["selection_nll"])["epoch"] require(row["selected_epoch"] == chosen, "选中轮次与规则不一致") for stage, temperature in [("before", 1), ("after", row["temperature"])]: metric = reference.metrics(row["logits"]["test"], labels, temperature) for key in ["nll", "accuracy", "ece"]: close(metric[key], row["metrics"]["test"][stage][key], f"测试 {key} 不一致") bootstrap = next(r for r in evidence["full"] if r["seed"] == row["seed"])["bootstrap"] delta = [reference.nll([z], [y], row["temperature"]) - reference.nll([z], [y]) for z, y in zip(row["logits"]["test"], labels)] require(len(delta) == len(bootstrap["per_item_delta"]), "逐条差值缺失") for a, b in zip(delta, bootstrap["per_item_delta"]): close(a, b, "逐条差值不一致") close(sum(delta)/len(delta), bootstrap["mean"], "平均差值不一致") require(len(bootstrap["resampled_means"]) == evidence["plan"]["bootstrap_repeats"], "重采样次数不一致") for p, end in zip([.025, .975], bootstrap["interval"]): close(quantile(bootstrap["resampled_means"], p), end, "分位数区间不一致") cal = numeric["splits"]["calibration"] lookup = {value: i for i, value in enumerate(cal)} plan = evidence["plan"] expected_count = len(seeds) * (plan["subset_repeats"]*(len(plan["sizes"])-1)+1) require(len(evidence["runs"]) == evidence["fit_count"] == expected_count, "扩展记录缺失") combinations = set() for run in evidence["runs"]: key = run["seed"], run["size"], run["repeat"] require(key not in combinations, "扩展记录重复") combinations.add(key) use = run["calibration_ids"] require(len(use) == len(set(use)) == run["size"], "子集大小或唯一性不符") require(set(use) <= set(cal), "子集出现非校准 ID") row = next(r for r in rows if r["seed"] == run["seed"]) logits = [row["logits"]["calibration"][lookup[i]] for i in use] y = [numeric["labels"]["calibration"][lookup[i]] for i in use] errors = sum(max(range(len(z)), key=z.__getitem__) != label for z, label in zip(logits, y)) require(errors == run["calibration_errors"], "子集错误数不一致") t = run["temperature"] lo, hi = plan["temperature_bounds"] require(lo <= t <= hi, "温度超出预定范围") require(run["at_boundary"] == (t in [lo, hi]), "边界标记不一致") close(reference.nll(row["logits"]["test"], labels, t), run["test_nll"], "扩展 NLL 不一致") close(run["test_nll"] - row["metrics"]["test"]["before"]["nll"], run["delta_nll"], "扩展差值不一致") for seed in seeds: for size in plan["sizes"]: repeats = 1 if size == len(cal) else plan["subset_repeats"] require(all((seed, size, r) in combinations for r in range(repeats)), "组合覆盖不足") return {"training_runs": len(rows), "fits": expected_count, "test_pairs": len(labels)*len(rows)} def report_text(records): training = records["calibration-reproduction.json"] evidence = records["calibration-evidence.json"] literature = records["calibration-literature.json"] config, numeric = training["config"], training["numeric"] env = training["environment"] rows = numeric["results"] lines = [ "### 一项小型温度校准研究：方法复现与样本量探索", "", f"资料与实验日期：{literature['checked_on']}。研究类型：教学方法复现与探索性扩展。", "", "#### 摘要", "", "我们在真实手写数字数据上训练三个小型分类器，独立实现训练后的温度缩放。" "完整校准集下，三个模型的测试 NLL 均下降，准确率不变；" "条件性配对区间均跨零，不能据此宣称总体改善已得到确认。" "减少校准样本后，多次零错误子集把温度推到搜索下界，反而使测试损失大幅上升。" "本项目没有复制原论文的大规模基准，也没有提出已验证的新校准算法。", "", "#### 研究问题与方法来源", "", "问题一：在固定小型分类器上，能否正确实现并重复运行温度缩放？" "问题二：在固定模型和测试池下，减少校准样本后，拟合温度与测试 NLL 怎样变化？", "", "方法来自 Guo 等人的 ICML 2017 论文。核心计算是在冻结 logits 后除以共同正温度，" "只按独立校准标签最小化平均 NLL。第 44 章另用手工例子核对解析温度与反例；" "手工例子不参与下面的真实数据成绩。", "", "#### 实验设置", "", "数据为 E. Alpaydin 与 C. Kaynak（1998）的 " "[Optical Recognition of Handwritten Digits](https://doi.org/10.24432/C50P49)，" "使用 scikit-learn 1.9.1 随包 1797 条子集，UCI 数据页许可为 CC BY 4.0。" "像素除以 16；重新划分用于教学，不沿用原始 UCI 基准划分。", "", "样本数：" + "、".join(f"{name} {len(numeric['splits'][key])}" for key, name in [("train", "训练"), ("selection", "选模"), ("calibration", "校准"), ("test", "测试")]) + "。" f"先后分出测试、校准、选模的分层种子为 {config['split_seeds']}。", "", f"MLP 为 64→{config['hidden'][0]}→{config['hidden'][1]}→10，ReLU，" f"共 {training['parameters']:,} 个参数。Adam 学习率 {config['learning_rate']}，" f"权重衰减 0，批量 {config['batch_size']}，训练 {config['epochs']} 轮，种子 {config['seeds']}。" "按选模 NLL 最低的轮次保存模型，并列时保留较早轮次。", "", "校准采用正温度对数参数化和 SciPy 有界一维搜索，范围 " f"{config['temperature_bounds']}，并比较端点与 T=1。" "训练为 CPU FP32、单线程、确定性算法；离线指标为 float64。" f"环境：Python {env['python']}、PyTorch {env['torch']}、NumPy {env['numpy']}、" f"SciPy {env['scipy']}、scikit-learn {env['sklearn']}；{env['platform']}。" "完整构建信息在原始报告中。", "", "#### 核心实验结果", "", "| 种子 | 选中轮次 | 温度 |", "| --- | ---: | ---: |", ] for r in rows: lines.append(f"| {r['seed']} | {r['selected_epoch']} | {r['temperature']:.4f} |") lines += ["", "测试平均 NLL：", "", "| 种子 | 原始概率 | 校准后 |", "| --- | ---: | ---: |"] for r in rows: a, b = r["metrics"]["test"]["before"], r["metrics"]["test"]["after"] lines.append(f"| {r['seed']} | {a['nll']:.4f} | {b['nll']:.4f} |") lines += ["", "准确率与 ECE（15 个等宽区间，均为比例）：", ""] for r in rows: a, b = r["metrics"]["test"]["before"], r["metrics"]["test"]["after"] lines.append(f"- 种子 {r['seed']}：准确率 {a['accuracy']:.4f} → {b['accuracy']:.4f}；" f"ECE {a['ece']:.4f} → {b['ece']:.4f}。") lines += ["", "两个独立训练进程的数值记录与权重内容一致；" "第 45 章还在另一进程加载自己的权重，重新计算了逐条分数和指标。" "这是固定环境中的执行证据，不是跨硬件一致性或总体有效性的证明。", "", "固定模型与完整校准温度，逐张配对重采样 2000 次。测试 NLL 的“校准后减校准前”差值及名义 95% 百分位区间：", ""] for r in evidence["full"]: b = r["bootstrap"] lines.append(f"- 种子 {r['seed']}：{b['mean']:.5f}，" f"[{b['interval'][0]:.5f}, {b['interval'][1]:.5f}]。") lines += ["", "区间均跨零；这不证明两个方案等价。逐张独立仅为近似，" "没有按书写者分组，也未严格复现原始分层抽样设计；" "区间不包含重新训练、校准选择或任务分布变化。", "", "#### 样本量扩展与负面结果", "", "从同一 269 条校准池抽取 20/50/100 条，各 20 份。每次排列取前缀，" "三个模型共用相同 ID；完整 269 条每模型只拟合一次。" f"共 {evidence['fit_count']} 次拟合，所有结果保留。" "这项方案是在看过核心实验测试结果之后制定的探索，不是新独立确认。", "", "测试 NLL 比原始概率更高的次数：", "", "| 种子 | 20 条 | 50 条 | 100 条 |", "| --- | ---: | ---: | ---: |"] for seed in config["seeds"]: values = [] for size in [20, 50, 100]: group = [r for r in evidence["runs"] if r["seed"] == seed and r["size"] == size] values.append(f"{sum(r['delta_nll'] > 0 for r in group)}/{len(group)}") lines.append(f"| {seed} | " + " | ".join(values) + " |") lines += ["", "20 条子集中的边界结果：", ""] for seed in config["seeds"]: group = [r for r in evidence["runs"] if r["seed"] == seed and r["size"] == 20] zero = sum(r["calibration_errors"] == 0 for r in group) boundary = sum(r["at_boundary"] for r in group) lines.append(f"- 种子 {seed}：零错误子集 {zero}/20，边界解 {boundary}/20；" f"该组最大测试 NLL 为 {max(r['test_nll'] for r in group):.4f}。") lines += ["", "小集合没有错误时，降低温度会继续改善该集合的 NLL；" "测试集中的错误却可能因此变得更自信。这个解释与记录相符，" "但本项目没有把它当作所有失败的唯一机制。" "固定温度后的 5/10/15/30 箱 ECE 敏感性结果也全部保留；" "种子 22 在四种设置中均恶化。", "", "#### 限制、开放问题与结论", "", "仅一个重新划分的数据子集、三个 MLP、共享测试池；校准子集嵌套且重叠。" "没有比较其他校准方法，没有检查未见书写者或真实照片，" "也没有将同一个池里的重复抽样当作独立任务。" "小样本失败不能推翻原论文在其他设置下的实测结果。", "", "本项目完成了可重复的方法实现，并找到了值得后续验证的小校准集失败条件。" "对数温度正则化、不同数据划分与按书写者评价，均可作为下一项研究，" "但尚未在本项目得到独立验证，也不宣称这些想法新颖。", "", "#### 资料与复现入口", "", "资料包的 examples/data 保存原始报告、配置、扩展方案和文献记录；" "examples 保存完整脚本与固定依赖。manifest.json 记录各文件的 SHA-256。" "报告生成器复核已有记录和数值，不会重新训练。", "", "~~~sh", "python3 examples/research_project.py", "uv run --python 3.12 --with-requirements examples/requirements.txt \\", " python examples/reproduce_calibration.py --out runs/retrained", "uv run --python 3.12 --with-requirements examples/requirements.txt \\", " python examples/reproduce_calibration.py --verify runs/retrained", "~~~", "", "第二、三条命令用于另行训练与加载；默认研究报告仍引用随附的冻结记录，" "不会把新训练输出自动冒充同一份历史数据。", "", "参考文献：", ""] for p in literature["papers"]: lines.append(f"- {p['authors']}（{p['year']}），[{p['title']}]({p['url']})。{p['version']}。") return "\n".join(lines) + "\n" def export(folder, records): folder.mkdir(parents=True, exist_ok=False) examples = folder / "examples" (examples / "data").mkdir(parents=True) paths = [Path(name) for name in SCRIPTS + ["requirements.txt"]] paths += [Path("data") / name for name in INPUTS] for relative in paths: shutil.copyfile(HERE / relative, examples / relative) (folder / "report.md").write_text(report_text(records)) files = {str(p.relative_to(folder)): file_hash(p) for p in sorted(folder.rglob("*")) if p.is_file()} manifest = {"kind": "frozen teaching research bundle; not a new training run", "files": files} (folder / "manifest.json").write_text(json.dumps(manifest, indent=2) + "\n") return manifest def verify_bundle(folder): manifest = json.loads((folder / "manifest.json").read_text()) expected = {"report.md"} expected.update("examples/" + name for name in SCRIPTS + ["requirements.txt"]) expected.update("examples/data/" + name for name in INPUTS) require(set(manifest["files"]) == expected, "资料包清单不完整") for name, sha in manifest["files"].items(): require(file_hash(folder / name) == sha, f"资料包文件变化：{name}") return len(expected) def main(): parser = ArgumentParser(description=__doc__) group = parser.add_mutually_exclusive_group() group.add_argument("--out", type=Path, help="导出到尚不存在的资料包目录") group.add_argument("--verify-bundle", type=Path, help="核对资料包文件是否与清单一致") args = parser.parse_args() if args.verify_bundle: count = verify_bundle(args.verify_bundle) print(f"资料包文件核对通过：{count} 个文件与 SHA-256 清单一致。") return records = load() result = audit(records) if args.out: export(args.out, records) print(f"核对通过：{result['training_runs']} 次分类器训练记录，" f"{result['fits']} 次温度拟合，{result['test_pairs']} 对测试损失。") print("报告中的核心成绩与子集计数均从原始记录生成；未重新训练或拟合温度。") print("项目结论：完成方法复现与探索性扩展；不宣称原论文基准复制或新方法有效。") if __name__ == "__main__": main()