results: validate Rain layer-state mechanism

1 files changed, 108 insertions, 0 deletions

diff --git a/experiments/analyze_rain_ep_s1.py b/experiments/analyze_rain_ep_s1.py
new file mode 100644
index 0000000..72ac2f0
--- /dev/null
+++ b/experiments/analyze_rain_ep_s1.py
@@ -0,0 +1,108 @@
+#!/usr/bin/env python3
+"""Audit the Rain neuron-state development screen without promoting it."""
+
+from __future__ import annotations
+
+import json
+import math
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[1]
+RESULT_ROOT = ROOT / "results" / "ep_bias" / "s1"
+R001 = {
+    name: f"layer-{file_name}-r0p01-e3-s1988.json"
+    for name, file_name in {
+        "clean": "clean",
+        "raw": "raw",
+        "same_rms_noise": "noise",
+        "constant": "constant",
+        "innovation": "innovation",
+        "oracle": "oracle",
+    }.items()
+}
+
+
+def read(path: Path) -> dict:
+    with path.open(encoding="utf-8") as handle:
+        return json.load(handle)
+
+
+def finite(record: dict) -> bool:
+    return all(
+        metric.get("finite", all(math.isfinite(float(metric[key])) for key in (
+            "train_cost", "test_cost", "train_accuracy", "test_accuracy")))
+        for metric in record["metrics"]
+    )
+
+
+def main() -> None:
+    records = {
+        name: read(RESULT_ROOT / file_name)
+        for name, file_name in R001.items()
+    }
+    rows = {
+        name: {
+            "final_test_accuracy": record["final"]["test_accuracy"],
+            "all_finite": finite(record),
+            "wall_seconds": record["final"]["wall_seconds"],
+            "neutral_observations": record["final"].get(
+                "corrector", {}).get("neutral_observations", 0),
+            "final_corrector": record["final"].get("corrector", {}),
+        }
+        for name, record in records.items()
+    }
+    clean = rows["clean"]["final_test_accuracy"]
+    raw = rows["raw"]["final_test_accuracy"]
+    innovation = rows["innovation"]["final_test_accuracy"]
+    constant = rows["constant"]["final_test_accuracy"]
+    boundary = {}
+    for ratio_name in ("r4", "r0p1"):
+        boundary[ratio_name] = {}
+        for mode in ("innovation", "constant"):
+            record = read(
+                RESULT_ROOT / f"layer-{mode}-{ratio_name}-e3-s1988.json")
+            boundary[ratio_name][mode] = {
+                "epochs_completed": len(record["metrics"]),
+                "all_finite": finite(record),
+                "final_test_accuracy": record["final"]["test_accuracy"],
+            }
+    report = {
+        "stage": "rain_ep_layer_state_s1",
+        "status": "positive_single_seed_development_not_confirmation",
+        "ratio": 0.01,
+        "rows": rows,
+        "paired_development_effects": {
+            "innovation_minus_raw_accuracy_points": 100.0 * (
+                innovation - raw),
+            "innovation_minus_constant_accuracy_points": 100.0 * (
+                innovation - constant),
+            "clean_minus_innovation_accuracy_points": 100.0 * (
+                clean - innovation),
+            "raw_loss_recovered_fraction": (
+                innovation - raw) / (clean - raw),
+            "innovation_wall_over_raw_ratio": (
+                rows["innovation"]["wall_seconds"] / rows["raw"]["wall_seconds"]),
+        },
+        "matched_predictor_protocol": {
+            "innovation_neutral_observations": rows[
+                "innovation"]["neutral_observations"],
+            "constant_neutral_observations": rows[
+                "constant"]["neutral_observations"],
+            "extra_equilibrium_phases": 0,
+            "source": "existing first EP phase of the first training minibatch",
+            "predictor_updates_after_first_minibatch": 0,
+        },
+        "stronger_ratio_boundary": boundary,
+        "test_policy": (
+            "development test subset observed each epoch; ratio chosen here; "
+            "all accuracy claims require a new frozen confirmation"),
+    }
+    output = RESULT_ROOT.parent / "s1_summary.json"
+    output.write_text(
+        json.dumps(report, indent=2, sort_keys=True, allow_nan=False) + "\n")
+    print(json.dumps(report, indent=2, sort_keys=True, allow_nan=False))
+
+
+if __name__ == "__main__":
+    main()