1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
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()
|
