protocol: freeze convolutional HFA baseline funnel

1 files changed, 107 insertions, 0 deletions

diff --git a/experiments/analyze_oral_a_hfa_short.py b/experiments/analyze_oral_a_hfa_short.py
new file mode 100644
index 0000000..cc2ee3c
--- /dev/null
+++ b/experiments/analyze_oral_a_hfa_short.py
@@ -0,0 +1,107 @@
+#!/usr/bin/env python3
+"""Audit and select the frozen convolutional-HFA short screen."""
+import argparse
+import glob
+import json
+import math
+import os
+
+
+RATES = (0.01, 0.03, 0.1)
+
+
+def read(path):
+    with open(path) as handle:
+        record = json.load(handle)
+    args = record["args"]
+    expected = {
+        "mode": "hfa", "depth": 20, "width": 16, "seed": 0,
+        "loader_seed": 0, "epochs": 20, "train_limit": 10000,
+        "val_examples": 5000, "split_seed": 2027,
+        "eval_split": "validation", "eval_every": 0,
+        "augment_train": 1, "lr_schedule": "cosine", "warmup_epochs": 0,
+        "momentum": 0.9, "weight_decay": 1e-4,
+        "normalization": "batchnorm", "output_lr": 0.1,
+        "a_scale": 1.0, "alignment_probe": 32,
+    }
+    for key, value in expected.items():
+        if args.get(key) != value:
+            raise ValueError(f"{path}: {key} drift")
+    if float(args["lr"]) not in RATES:
+        raise ValueError(f"{path}: unregistered learning rate")
+    if record["provenance"]["git_tracked_dirty"]:
+        raise ValueError(f"tracked-dirty result: {path}")
+    protocol = record["evaluation_protocol"]
+    if protocol["test_evaluations"] or protocol["test_used_for_selection"]:
+        raise ValueError(f"short screen touched test: {path}")
+    accuracy = float(record["final"]["accuracy"])
+    loss = float(record["final"]["loss"])
+    diagnostics = record.get("diagnostics") or {}
+    early = float(diagnostics.get("early_third_mean", float("nan")))
+    finite = (bool(record["final"]["finite"])
+              and math.isfinite(accuracy + loss + early))
+    return {
+        "path": path, "lr": float(args["lr"]), "accuracy": accuracy,
+        "loss": loss, "early_third_alignment": early, "finite": finite,
+        "total_macs": int(record["work"]["total_macs_estimate"]),
+        "fixed_feedback_parameters": int(
+            record["architecture"]["fixed_feedback_parameters"]),
+        "logical_batch_loss_queries": int(
+            record["work"]["logical_batch_loss_queries"]),
+        "peak_memory_allocated_bytes": int(
+            record["hardware"]["peak_memory_allocated_bytes"]),
+        "wall_s": float(record["timing"]["total_timed_wall_s"]),
+        "source_commit": record["provenance"]["git_commit"],
+    }
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--input", default="results/oral_a_hfa_short")
+    parser.add_argument(
+        "--out", default="results/oral_a_hfa_short_selection.json")
+    args = parser.parse_args()
+    rows = [read(path) for path in sorted(
+        glob.glob(os.path.join(args.input, "*.json")))]
+    if len(rows) != len(RATES):
+        raise ValueError(f"expected {len(RATES)} HFA-S1 runs, found {len(rows)}")
+    if {row["lr"] for row in rows} != set(RATES):
+        raise ValueError("HFA-S1 learning-rate grid is incomplete")
+    if len({row["source_commit"] for row in rows}) != 1:
+        raise ValueError("HFA-S1 source commits differ")
+    finite = [row for row in rows if row["finite"]]
+    if not finite:
+        selected = None
+        status = "failed_no_finite_candidate"
+    else:
+        selected = sorted(
+            finite, key=lambda row: (
+                -row["accuracy"], row["total_macs"], row["lr"]))[0]
+        status = ("selected_full_open" if selected["accuracy"] >= 0.50
+                  else "selected_full_closed")
+    output = {
+        "protocol": "convolutional_hfa_S1_v1", "status": status,
+        "rows": rows, "selected": selected,
+        "comparators": {
+            "matched_short_dfa_accuracy": 0.3716,
+            "matched_short_failed_v1_sdil_accuracy": 0.4198,
+            "matched_short_bp_accuracy": 0.7494,
+        },
+        "full_validation_threshold": 0.50,
+        "confirmation_test_seeds_touched": False,
+    }
+    os.makedirs(os.path.dirname(os.path.abspath(args.out)), exist_ok=True)
+    with open(args.out, "w") as handle:
+        json.dump(output, handle, indent=2, sort_keys=True)
+        handle.write("\n")
+    print(json.dumps({
+        "status": status, "selected": selected,
+        "rows": [{key: row[key] for key in (
+            "lr", "accuracy", "early_third_alignment", "finite")}
+                 for row in rows],
+    }, indent=2))
+
+
+if __name__ == "__main__":
+    main()
+