Exponential-Map Stochastic Residual Layer / stage2_sphere_bench.py
Beats tuned baseline
1import json
2import math
3import sys
4from pathlib import Path
5
6import numpy as np
7import torch
8import torch.nn as nn
9
10sys.path.insert(0, "/home/maxwelhelp/all/math2nn")
11from bench import make_model, train_model, evaluate, sweep_baseline, make_report
12from sphere_track import get_dataset, META
13
14OUT = Path("bench_report.json")
15
16
17def exp_sphere(x, v):
18 r = torch.linalg.vector_norm(v, dim=-1, keepdim=True)
19 return torch.cos(r) * x + torch.sinc(r / math.pi) * v
20
21
22def tangent(x, u):
23 return u - (u * x).sum(dim=-1, keepdim=True) * x
24
25
26class ResidualSystem(nn.Module):
27 """Same MLP backbone; only the residual integration rule differs."""
28 def __init__(self, idea, h):
29 super().__init__()
30 self.backbone = make_model("mlp_tiny", (3,), 3)
31 self.idea = bool(idea)
32 self.h = float(h)
33
34 def forward(self, x):
35 raw = self.backbone(x)
36 if self.idea:
37 return exp_sphere(x, self.h * tangent(x, raw))
38 return x + self.h * raw
39
40
41def dataset(seed):
42 d = get_dataset(seed, 400, 400)
43 out = {k: torch.as_tensor(v, dtype=torch.float32) for k, v in d.items()
44 if k in ("xtr", "ytr", "xte", "yte")}
45 out.update({"task": "regression", "metric": "mse", "input_shape": (3,), "out_dim": 3})
46 return out
47
48
49def run_one(idea, cfg, seed, return_model=False):
50 torch.manual_seed(10000 + int(seed))
51 np.random.seed(10000 + int(seed))
52 net, metric, history = train_model(
53 ResidualSystem(idea, cfg["h"]), dataset(seed),
54 epochs=20, lr=cfg["lr"], batch=128, weight_decay=0.0,
55 log=lambda *_args, **_kwargs: None)
56 if metric is None:
57 raise RuntimeError("benchmark training failed")
58 return (float(metric), net) if return_model else float(metric)
59
60
61def main():
62 # Cheap numerical verification is performed before any training.
63 torch.manual_seed(3)
64 x = torch.randn(1000, 3)
65 x = x / torch.linalg.vector_norm(x, dim=-1, keepdim=True)
66 u = torch.randn_like(x)
67 v = 0.12 * tangent(x, u)
68 exp_norm_error = float((torch.linalg.vector_norm(exp_sphere(x, v), dim=-1) - 1).abs().max())
69 add_norm_error = float((torch.linalg.vector_norm(x + v, dim=-1) - 1).abs().mean())
70 if not (exp_norm_error < 1e-6 and add_norm_error > 1e-5):
71 raise RuntimeError("core exponential-map norm sanity check failed")
72
73 lrs = [1e-3, 3e-3, 1e-2]
74 hs = [0.06, 0.12, 0.24]
75 grid = [{"lr": lr, "h": h} for lr in lrs for h in hs]
76 # Baseline sees the complete union of all idea hyperparameters.
77 base = sweep_baseline(
78 lambda cfg: lambda seed: run_one(False, cfg, seed), grid)
79 best = base["best_cfg"]
80 idea_grid = [best]
81 for cfg in grid:
82 if cfg != best and len(idea_grid) < 3:
83 idea_grid.append(cfg)
84 idea_trials = []
85 for cfg in idea_grid:
86 r = evaluate(lambda seed, c=cfg: run_one(True, c, seed))
87 idea_trials.append({"cfg": cfg, "result": r})
88 best_idea_trial = min(idea_trials, key=lambda z: z["result"]["mean"])
89 idea = best_idea_trial["result"]
90 rep = make_report("sphere_one_step_dynamics", "mlp_tiny", base, idea, extra={
91 "sanity_check": {
92 "prediction": "Exp preserves unit norm while additive Euler drifts",
93 "exp_max_norm_error": exp_norm_error,
94 "additive_mean_norm_error": add_norm_error,
95 "confirmed": bool(exp_norm_error < 1e-6 and add_norm_error > 1e-5),
96 },
97 "idea_trials": idea_trials,
98 })
99 rep["custom_track"] = {
100 "name": META["name"], "file": "sphere_track.py", "domain": META["domain"]}
101 # Signature is measured from independently trained systems at their selected configs.
102 cfgb = best
103 mb_metric, mb = run_one(False, cfgb, 0, return_model=True)
104 mi_metric, mi = run_one(True, best_idea_trial["cfg"], 0, return_model=True)
105 with torch.no_grad():
106 xx = dataset(0)["xte"]
107 xx_b = xx.to(next(mb.parameters()).device)
108 xx_i = xx.to(next(mi.parameters()).device)
109 pb = mb(xx_b); pi = mi(xx_i)
110 rep["mechanism_signature"] = {
111 "trained_model_seed": 0,
112 "baseline_mean_abs_norm_error": float((pb.norm(dim=1) - 1).abs().mean()),
113 "idea_mean_abs_norm_error": float((pi.norm(dim=1) - 1).abs().mean()),
114 "predicted_idea_lower_constraint_error": True,
115 "confirmed": bool((pi.norm(dim=1) - 1).abs().mean() < (pb.norm(dim=1) - 1).abs().mean()),
116 }
117 OUT.write_text(json.dumps(rep, indent=2))
118 print(json.dumps(rep, indent=2))
119
120
121if __name__ == "__main__":
122 main()