"""附录 F：实际运行打印算例，生成配置、JSONL 日志和可复核报告；仅标准库。""" from argparse import ArgumentParser from datetime import datetime, timezone import hashlib import json from math import isfinite from pathlib import Path import platform import tempfile CONFIG = { "experiment_id": "printing-grid-v1", "slopes": [i / 2 for i in range(11)], "intercepts": list(range(41)), "training_metric": "mse_seconds_squared", "selection_metric": "validation_mae_seconds", "tie_break": "first_in_declared_order", } DATA = { "train": [[5, 37], [10, 47], [15, 66], [20, 81], [25, 92], [30, 112]], "validation": [[8, 43], [18, 76], [28, 103]], "test": [[12, 58], [22, 85], [26, 101]], } HERE = Path(__file__).resolve() def encode(value): return json.dumps(value, ensure_ascii=False, sort_keys=True, indent=2, allow_nan=False) + "\n" def sha(data): return hashlib.sha256(data).hexdigest() def score(rows, a, b, squared=False): errors = [a * x + b - y for x, y in rows] return sum(e * e if squared else abs(e) for e in errors) / len(errors) def calculate(config, data): if set(config) != set(CONFIG): raise ValueError("配置字段不完整或包含未知字段") for field in ["training_metric", "selection_metric", "tie_break"]: if config[field] != CONFIG[field]: raise ValueError("本示例只实现固定的指标与并列规则") if not config["slopes"] or not config["intercepts"]: raise ValueError("候选参数不能为空") for name in ["slopes", "intercepts"]: if not isinstance(config[name], list) or any( isinstance(v, bool) or not isinstance(v, (int, float)) or not isfinite(v) for v in config[name] ): raise ValueError("候选参数必须是有限数值的列表") # 先完成训练与验证选择，再读取测试行。 train, validation = data["train"], data["validation"] logs, best, best_loss = [], None, float("inf") for a in config["slopes"]: for b in config["intercepts"]: loss = score(train, a, b, squared=True) logs.append(dict(event="candidate", step=len(logs), split="train", a=a, b=b, metric="mse_seconds_squared", value=loss)) if loss < best_loss: best, best_loss = (a, b), loss baseline = (0., sum(y for _, y in train) / len(train)) candidates = [("mean", baseline), ("linear", best)] selected, selected_score = None, float("inf") for name, (a, b) in candidates: value = score(validation, a, b) logs.append(dict(event="selection", split="validation", model=name, a=a, b=b, metric="mae_seconds", value=value)) if value < selected_score: selected, selected_score = (name, a, b), value name, a, b = selected predictions = [dict(sample_id=f"test-{i}", pages=x, target_seconds=y, prediction_seconds=a * x + b) for i, (x, y) in enumerate(data["test"])] report = dict(status="completed", data_kind="synthetic_teaching_example", search_count=len(config["slopes"]) * len(config["intercepts"]), selected_model=name, a=a, b=b, validation_mae_seconds=selected_score, test_mae_seconds=score(data["test"], a, b), limitations=["人工教学数据，不是打印机实测", "无随机训练，未提供人群泛化或区间保证", "记录与重算不等于独立研究复现"]) return logs, predictions, report def write_run(directory, config): # 所有输入验证与计算成功后才创建输出，不覆盖旧实验。 logs, predictions, report = calculate(config, DATA) directory.mkdir(parents=True, exist_ok=False) payload = {"config.json": config, "data.json": DATA, "predictions.json": predictions, "report.json": report, "environment.json": { "python": platform.python_version(), "platform": platform.platform(), "recorded_at_utc": datetime.now(timezone.utc).isoformat(), "code_sha256": sha(HERE.read_bytes()), "randomness": "none: exhaustive grid search", }} for name, value in payload.items(): (directory / name).write_text(encode(value), encoding="utf-8") # 每行一条完整 JSON，按事件顺序保存，末尾保留换行。 (directory / "metrics.jsonl").write_text( "".join(json.dumps(row, ensure_ascii=False, allow_nan=False) + "\n" for row in logs), encoding="utf-8", ) files = {p.name: sha(p.read_bytes()) for p in sorted(directory.iterdir())} (directory / "manifest.json").write_text(encode(files), encoding="utf-8") return report def verify_run(directory): manifest = json.loads((directory / "manifest.json").read_text()) expected = {"config.json", "data.json", "environment.json", "metrics.jsonl", "predictions.json", "report.json"} if set(manifest) != expected: raise ValueError("资料清单不完整") for name in expected: if sha((directory / name).read_bytes()) != manifest[name]: raise ValueError(f"文件指纹不符：{name}") env = json.loads((directory / "environment.json").read_text()) if env["code_sha256"] != sha(HERE.read_bytes()): raise ValueError("核对程序与记录中的版本不同") config = json.loads((directory / "config.json").read_text()) data = json.loads((directory / "data.json").read_text()) logs, predictions, report = calculate(config, data) saved_logs = [json.loads(line) for line in (directory / "metrics.jsonl").read_text().splitlines()] if logs != saved_logs or predictions != json.loads((directory / "predictions.json").read_text()): raise ValueError("逐条日志或预测与重算不一致") if report != json.loads((directory / "report.json").read_text()): raise ValueError("报告与重算不一致") return report def main(): parser = ArgumentParser(description=__doc__) group = parser.add_mutually_exclusive_group() group.add_argument("--out", type=Path, help="创建新的实验目录") group.add_argument("--verify", type=Path, help="检查现有目录，重新计算小实验") parser.add_argument("--config", type=Path, help="新实验的 JSON 配置") args = parser.parse_args() if args.verify and args.config: parser.error("--verify 使用记录自己的配置，不能另传 --config") config = json.loads(args.config.read_text()) if args.config else CONFIG if args.verify: report = verify_run(args.verify) elif args.out: write_run(args.out, config) report = verify_run(args.out) else: with tempfile.TemporaryDirectory() as temporary: output = Path(temporary) / "example" write_run(output, config) report = verify_run(output) print(f'完成 {report["search_count"]} 组训练参数计算；日志包含全部候选和两项验证选择。') print(f'选定 {report["selected_model"]}：a={report["a"]:.1f}，b={report["b"]:.1f}。') print(f'测试 MAE：{report["test_mae_seconds"]:.3f} 秒。') print("六份文件的指纹、逐条日志、模型选择与测试预测均已核对。") if __name__ == "__main__": main()