code/reaggregate_vit_claim5.py

#!/usr/bin/env python3
"""Deterministic reaggregation of the paper's released ViT result arrays.

This script uses only official raw result files and the official schedule
definition at the paper-linked commit.  It never imports peer material,
executes author code, downloads data, or trains a model.
"""

from __future__ import annotations

import hashlib
import json
import math
from pathlib import Path


ROOT = Path(__file__).resolve().parents[1]
INPUTS = ROOT / "official-inputs"
OUTPUT = ROOT / "outputs" / "claim5_official_reaggregate.json"
COMMIT = "ce1baa41915ba1601186c604e91fecf9f13b83fe"
EXPECTED = {
    "schedules.py": (2581, "9708ccfdf27641fd4ca9618badcd0325b0134528c39d4ac640f7ec83b3717257"),
    "vit_dropout_results.json": (166043, "9301bcbd5c3404970da872bbac3966585a0ca5d20e49ce96ad8833c100ccefb6"),
    "ablation_results.json": (235141, "86e2d0df7937cdf75336f8d47fa476ed3953c4aab253a15d362c84ece0e07470"),
}
URL_BASE = (
    "https://raw.githubusercontent.com/luklacasito/"
    "dropout-universality-experiments/" + COMMIT + "/"
)


def load_checked(name: str):
    path = INPUTS / name
    data = path.read_bytes()
    expected_bytes, expected_sha = EXPECTED[name]
    actual_sha = hashlib.sha256(data).hexdigest()
    if len(data) != expected_bytes or actual_sha != expected_sha:
        raise SystemExit(f"input gate failed for {name}: {len(data)} {actual_sha}")
    if name.endswith(".json"):
        return json.loads(data)
    return data.decode()


def mean_sem(values: list[float]) -> tuple[float, float]:
    mean = sum(values) / len(values)
    if len(values) < 2:
        return mean, 0.0
    variance = sum((x - mean) ** 2 for x in values) / (len(values) - 1)
    return mean, math.sqrt(variance / len(values))


def final_metric(group: dict, metric: str) -> list[float]:
    rows = group[metric]
    if not rows or any(len(row) != 75 for row in rows):
        raise SystemExit(f"unexpected 75-epoch shape for {metric}")
    return [float(row[-1]) for row in rows]


def summarize(group: dict, baseline: dict, label: str) -> dict:
    loss = final_metric(group, "test_loss")
    acc = final_metric(group, "test_acc")
    base_loss = final_metric(baseline, "test_loss")
    base_acc = final_metric(baseline, "test_acc")
    if len(loss) != len(base_loss):
        raise SystemExit(f"paired seed count mismatch for {label}")
    loss_mean, loss_sem = mean_sem(loss)
    acc_mean, acc_sem = mean_sem(acc)
    base_loss_mean, _ = mean_sem(base_loss)
    base_acc_mean, _ = mean_sem(base_acc)
    return {
        "runs": len(loss),
        "epochs": 75,
        "final_test_loss_mean": loss_mean,
        "final_test_loss_sem": loss_sem,
        "final_test_accuracy_mean_percent": acc_mean,
        "final_test_accuracy_sem_percent": acc_sem,
        "loss_reduction_vs_constant_percent": 100.0 * (base_loss_mean - loss_mean) / base_loss_mean,
        "accuracy_gain_vs_constant_pp": acc_mean - base_acc_mean,
        "paired_final_loss_wins": sum(a < b for a, b in zip(loss, base_loss)),
        "paired_final_accuracy_wins": sum(a > b for a, b in zip(acc, base_acc)),
        "final_loss_values": loss,
        "final_accuracy_values_percent": acc,
    }


def main() -> None:
    schedules = load_checked("schedules.py")
    vit = load_checked("vit_dropout_results.json")
    ablation = load_checked("ablation_results.json")
    if "reverse_step" not in schedules or "reverse_linear" not in schedules:
        raise SystemExit("official schedule identity gate failed")
    if "return [h_adj] * n_drop + [0.0] * (depth - n_drop)" not in schedules or "reverse_linear" not in schedules:
        raise SystemExit("front-loaded schedule gate failed")
    for key in ("constant", "reverse_step", "reverse_linear"):
        if key not in vit:
            raise SystemExit(f"missing CIFAR-100 schedule {key}")
    for key in ("both_constant", "both_reverse_step"):
        if key not in ablation:
            raise SystemExit(f"missing CIFAR-10 ablation {key}")
    result = {
        "route": "official_released_raw_artifact_reaggregation",
        "orid": "FoDU47u2jk",
        "paper": "arXiv:2605.21648v2",
        "official_repository_commit": COMMIT,
        "inputs": {
            name: {
                "url": URL_BASE + ("src/dropout_mft/schedules.py" if name == "schedules.py" else "results/transformer/" + name),
                "bytes": EXPECTED[name][0],
                "sha256": EXPECTED[name][1],
            }
            for name in EXPECTED
        },
        "protocol_controls": {
            "paper_viT_cifar100": "10 runs, 75 epochs; Table 10/11; full CIFAR-100; A100",
            "paper_vit_cifar10_ablation": "5 runs, 75 epochs; Table 12/13; full CIFAR-10; A100",
            "metric": "last recorded test_loss and test_acc value per run",
            "denominator": "mean across 10 or 5 official runs; SEM uses sample standard deviation / sqrt(n)",
            "fixed_budget": "official reverse_step/reverse_linear versus constant schedule definitions at mean dropout 0.1 and maximum 0.2",
        },
        "cifar100": {
            "baseline": "constant",
            "comparisons": {
                "front_loaded_step": summarize(vit["reverse_step"], vit["constant"], "cifar100_reverse_step"),
                "front_loaded_linear": summarize(vit["reverse_linear"], vit["constant"], "cifar100_reverse_linear"),
            },
        },
        "cifar10": {
            "baseline": "both_constant",
            "comparison": summarize(ablation["both_reverse_step"], ablation["both_constant"], "cifar10_both_reverse_step"),
        },
    }
    c100_linear = result["cifar100"]["comparisons"]["front_loaded_linear"]
    c10_step = result["cifar10"]["comparison"]
    result["claim5_numeric_gate"] = {
        "cifar100_linear_loss_reduction_percent": c100_linear["loss_reduction_vs_constant_percent"],
        "cifar10_step_loss_reduction_percent": c10_step["loss_reduction_vs_constant_percent"],
        "maximum_accuracy_gain_pp": c100_linear["accuracy_gain_vs_constant_pp"],
        "maximum_accuracy_gain_rounded_2dp": round(c100_linear["accuracy_gain_vs_constant_pp"], 2),
        "all_loss_comparisons_win": (
            c100_linear["paired_final_loss_wins"] == c100_linear["runs"]
            and c10_step["paired_final_loss_wins"] == c10_step["runs"]
        ),
        "within_claimed_approximately_4_to_6_percent_band": (
            4.0 <= c100_linear["loss_reduction_vs_constant_percent"] <= 4.5
            and 6.0 <= c10_step["loss_reduction_vs_constant_percent"] <= 6.5
        ),
        "within_claimed_approximately_0_66pp_accuracy_bound": round(c100_linear["accuracy_gain_vs_constant_pp"], 2) == 0.66,
    }
    if not all(result["claim5_numeric_gate"].values()):
        raise SystemExit("claim 5 numeric gate failed")
    OUTPUT.write_text(json.dumps(result, indent=2) + "\n")
    print(json.dumps(result, indent=2))


if __name__ == "__main__":
    main()