Track simulator fidelity experiment artifacts

runs/frontier-multicase-sufficiency-v0/audit_qwen30_baseline.py

@@ -0,0 +1,110 @@
+#!/usr/bin/env python3
+"""Extract the predictive-versus-calibrated Frontier Qwen30B baseline."""
+
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+from typing import Any
+
+
+MODES = ("uncalibrated/SLO-gated", "frozen-calibrated/SLO-gated")
+
+
+def load_json(path: Path) -> dict[str, Any]:
+    with path.open(encoding="utf-8") as source:
+        value = json.load(source)
+    if not isinstance(value, dict):
+        raise ValueError(f"expected JSON object: {path}")
+    return value
+
+
+def extract(metrics: dict[str, Any], protocol: dict[str, Any]) -> dict[str, Any]:
+    analyses = metrics["analyses"]
+    rows = []
+    for mode in MODES:
+        analysis = analyses[mode]
+        values = analysis["metrics"]
+        top1 = values["top1"]
+        confusion = analysis["false_feasibility"]["overall"]
+        rows.append(
+            {
+                "mode": mode,
+                "selected_cells": top1["candidate_cells"],
+                "optimistic_real_regret": top1["optimistic_regret"],
+                "worst_case_real_regret": top1["worst_case_regret"],
+                "kendall_tau_b": values["kendall_tau_b"]["tau_b"],
+                "pairwise_exact_sign_accuracy": values["pairwise_direction"][
+                    "exact_sign_accuracy"
+                ],
+                "false_feasible": confusion["false_feasible"],
+                "false_infeasible": confusion["false_infeasible"],
+                "agreement": confusion["agreement"],
+            }
+        )
+    return {
+        "schema": "frontier-qwen30-calibration-audit-v0",
+        "rows": rows,
+        "calibration": {
+            "fitted_a_tp": protocol["fitted_a_tp"],
+            "fit_fixture": protocol["fit_fixture"],
+            "holdout_fixture": protocol["holdout_fixture"],
+            "loss": protocol["loss"],
+            "refit_on_holdout": protocol["refit_on_holdout"],
+        },
+    }
+
+
+def report(result: dict[str, Any]) -> str:
+    lines = [
+        "# Qwen30B Frontier baseline audit",
+        "",
+        "| mode | selected cells | worst real regret | Kendall tau-b | pair sign accuracy | feasibility (agree/FP/FN) |",
+        "|---|---|---:|---:|---:|---:|",
+    ]
+    for row in result["rows"]:
+        lines.append(
+            f"| {row['mode']} | {', '.join(row['selected_cells'])} "
+            f"| {row['worst_case_real_regret']:.2%} | {row['kendall_tau_b']:.4f} "
+            f"| {row['pairwise_exact_sign_accuracy']:.2%} "
+            f"| {row['agreement']}/{row['false_feasible']}/{row['false_infeasible']} |"
+        )
+    calibration = result["calibration"]
+    lines.extend(
+        [
+            "",
+            "The calibrated mode applies a distinct end-to-end execution-time scale per TP: "
+            + ", ".join(
+                f"TP{tp}={value:.6f}"
+                for tp, value in sorted(calibration["fitted_a_tp"].items())
+            )
+            + ".",
+            "",
+            f"Those scales were fitted against real total throughput on "
+            f"`{calibration['fit_fixture']}` and checked without refitting on "
+            f"`{calibration['holdout_fixture']}`. This validates within-workload transfer of "
+            "the calibration, not zero-shot Frontier prediction across TP.",
+            "",
+        ]
+    )
+    return "\n".join(lines)
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--metrics", type=Path, required=True)
+    parser.add_argument("--calibration-protocol", type=Path, required=True)
+    parser.add_argument("--output-dir", type=Path, required=True)
+    args = parser.parse_args()
+
+    result = extract(load_json(args.metrics), load_json(args.calibration_protocol))
+    args.output_dir.mkdir(parents=True, exist_ok=True)
+    (args.output_dir / "metrics.json").write_text(
+        json.dumps(result, indent=2, sort_keys=True) + "\n", encoding="utf-8"
+    )
+    (args.output_dir / "report.md").write_text(report(result), encoding="utf-8")
+
+
+if __name__ == "__main__":
+    main()