summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorYurenHao0426 <Blackhao0426@gmail.com>2026-06-09 15:28:30 -0500
committerYurenHao0426 <Blackhao0426@gmail.com>2026-06-09 15:28:30 -0500
commitb246cc878c3db48e6d993a6a293176ee9be97e77 (patch)
treeed7baf048144927ba98f417e5d9c28b5a4489f31
parent2488deff02737330ff4f355746be2b49492ac3dc (diff)
Add MNIST real-data validation of e0 theorem and estimator
New script validates both no-fit claims on MNIST-subset MLPs. e0 theorem: 16 configs, hidden share 0.39-0.82, measured E_B[e0] within 0.0001-0.0089 of prediction (512 real-backward draws each). Estimator: 96 trajectories, velocity MAE 0.00196 (3.9% of mean gap), bias -0.0006, beating fixed K(0) everywhere; kernels computed via a layerwise gram identity self-tested against autograd at machine precision. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
-rw-r--r--notes/36_evidence_ledger.md9
-rw-r--r--notes/38_mnist_real_data_validation.md99
-rw-r--r--scripts/real_data_validation.py515
3 files changed, 621 insertions, 2 deletions
diff --git a/notes/36_evidence_ledger.md b/notes/36_evidence_ledger.md
index 63e797d..14af7df 100644
--- a/notes/36_evidence_ledger.md
+++ b/notes/36_evidence_ledger.md
@@ -100,8 +100,13 @@ standard lazy approximation whose breakdown is *measured* (item 3c below).
overprediction (+0.0026), FA near-unbiased (+0.0005); cause: drift velocity
decays after s (alpha_t < t/s; e.g. true alpha_50 = 1.46 (BP) / 1.17 (FA)
vs linear 2.5). Honest error story, no fitted correction.
-5. **MNIST replication (note 38)**: same protocol, width 256, depth 2,
- N in {128,256,512}, 48 traj: see `real_data_validation_mnist/`.
+5. **MNIST replication (note 38)**: same protocol, widths {128,256} x
+ N {128,256,512} x 2 inits x 8 feedbacks = 96 traj, lazy step
+ (`eta lam_max/N = 0.05`): velocity MAE **0.00196** (= 3.9% of mean gap),
+ bias -0.0006, corr 0.911 (gap range only 0.040-0.063, so MAE/bias are the
+ operative stats); fixed K(0) MAE 0.0061 biased high — same bracket as
+ synthetic. At eta lam_max/N = 0.3 the lazy chain breaks as expected
+ (scope boundary, recorded in note 38).
### (d) The soft ramp + closed form (T7)
diff --git a/notes/38_mnist_real_data_validation.md b/notes/38_mnist_real_data_validation.md
new file mode 100644
index 0000000..011f335
--- /dev/null
+++ b/notes/38_mnist_real_data_validation.md
@@ -0,0 +1,99 @@
+# MNIST Real-Data Validation (e0 Theorem + Estimator)
+
+Script: `scripts/real_data_validation.py`. Outputs:
+`outputs/real_data_validation_mnist/`. Purpose: remove the "synthetic-only"
+objection for the two core no-fit claims by replicating them on MNIST-subset
+MLPs. Run 2026-06-09.
+
+Setup: flattened standardized MNIST (rows scaled by `1/sqrt(784)` so kernel
+scales match the synthetic runs), one-hot regression targets, squared loss
+`L = ||r||^2/(2N)`, float64, full-batch SGD, project init/feedback
+conventions (`downstream_capacity_sweep` gradients are reused directly).
+
+Tangent operators are computed by the layerwise gram identity instead of an
+explicit `(NC) x P` Jacobian; `--self-test` checks the identity against
+autograd Jacobians on a tiny net: max errors `<= 5.3e-15` for
+`K_BP = J J^T`, `K_FA = J Jtilde^T`, and `J^T r` vs the production gradient
+code (machine precision).
+
+## Part A — initial-erosion theorem on real data
+
+`E_B[e_0 | W, r] = hidden BP speed share` (note 29 / T5). For each config the
+prediction is one BP backward pass; the measurement is 512 independent
+feedback draws, each through the **real FA backward pass**.
+
+16 configs: width {64, 256} x hidden depth {1, 2} x 4 init seeds, N=256.
+
+| depth | width | hidden_share range (4 inits) | max abs diff | typical stderr |
+|---|---|---|---|---|
+| 1 | 64 | 0.727 - 0.819 | 0.0067 | 0.005 |
+| 1 | 256 | 0.414 - 0.451 | 0.0018 | 0.0014 |
+| 2 | 64 | 0.684 - 0.792 | 0.0089 | 0.004 |
+| 2 | 256 | 0.395 - 0.531 | 0.0023 | 0.0012 |
+
+All 16 panels: `|E_B[e_0] - hidden_share|` in **0.0001-0.0089**, within ~2
+Monte-Carlo stderr. The predicted share spans **0.39-0.82** across configs and
+the measured mean tracks the diagonal across that whole range
+(`e0_mnist_calibration.png`, `e0_mnist_rows.csv`).
+
+Real-data reading: on 784-dim inputs the hidden share — the predicted initial
+FA burden — is large at small width (~0.8) and decreases as the output layer
+(fan-in = width) grabs more BP energy. The theorem prices this with no fit.
+
+## Part B — early operator-velocity estimator on real data
+
+Protocol mirrors notes 15-17: full-batch SGD to `T=50`, early snapshot at
+`s=20`, predictors rolled through `r_{t+1} = (I - eta K_hat_t/N) r_t` with
+`K_hat_t = K_0` (fixed) or `K_0 + (t/s)(K_s - K_0)` (linear velocity);
+empirical gap = trained `L_FA(T) - L_BP(T)`. No fitted scalar anywhere.
+
+96 trajectories: width {128, 256} x N {128, 256, 512} x 2 inits x 8 feedback
+seeds. Step size set per config as `eta = 0.05 N / lam_max(K_BP_0)` (lazy
+regime; see sensitivity below).
+
+| predictor | MAE | bias | corr |
+|---|---|---|---|
+| fixed K(0) | 0.00613 | +0.00529 | 0.606 |
+| **linear velocity** | **0.00196** | **-0.00062** | **0.911** |
+
+Per-config (16 traj each):
+
+| w | N | gap_emp mean +/- std | velocity MAE | fixed MAE |
+|---|---|---|---|---|
+| 128 | 128 | 0.0570 +/- 0.0038 | 0.0014 | 0.0121 |
+| 128 | 256 | 0.0517 +/- 0.0043 | 0.0016 | 0.0092 |
+| 128 | 512 | 0.0482 +/- 0.0040 | 0.0021 | 0.0075 |
+| 256 | 128 | 0.0533 +/- 0.0026 | 0.0026 | 0.0040 |
+| 256 | 256 | 0.0495 +/- 0.0023 | 0.0020 | 0.0020 |
+| 256 | 512 | 0.0455 +/- 0.0025 | 0.0021 | 0.0020 |
+
+Readings:
+
+- Relative accuracy: velocity MAE is **3.9% of the mean gap** with near-zero
+ bias; it beats fixed K(0) everywhere (4-9x at width 128, parity-to-better at
+ width 256 where the kernel drifts less — wider = lazier, as theory says).
+- Bias signs reproduce the synthetic bracket (fixed biased high +0.0053,
+ velocity slightly low -0.0006; notes 16-17).
+- corr 0.911 vs synthetic 0.999: the MNIST gap only spans 0.040-0.0625 across
+ all 96 trajectories (the sweep moves the gap ~10-20%, vs orders of magnitude
+ in the synthetic margin sweeps), so correlation is a weaker statistic here;
+ MAE/bias are the meaningful numbers. State it this way in the paper.
+
+## Step-size sensitivity (kept as a scope datum)
+
+A first run used `eta = 0.3 N / lam_max` (per-step top-mode decay 30%):
+fixed MAE 0.069 corr 0.37; velocity MAE 0.030 corr 0.68 — both overpredict,
+velocity still 2.3x better. At `0.05` (matching the synthetic regime's
+`eta lam_max/N ~ 0.04`) the chain holds. This is the expected lazy-regime
+boundary of A1, not a new failure mode: the estimator's stated scope is the
+regime where one-step linearization is accurate.
+
+## Paper use
+
+- Part A => extends Fig "e0 calibration" with a real-data panel (F5 in the
+ ledger): theorem holds on MNIST across a 0.39-0.82 share range.
+- Part B => real-data estimator scatter (`estimator_mnist_scatter.png`):
+ no-fit finite-time prediction at ~4% relative error.
+- Limitations sentence upgrade: "synthetic only" becomes "synthetic +
+ MNIST-subset MLPs; conv/transformers and stochastic optimization remain
+ future work."
diff --git a/scripts/real_data_validation.py b/scripts/real_data_validation.py
new file mode 100644
index 0000000..07c0de1
--- /dev/null
+++ b/scripts/real_data_validation.py
@@ -0,0 +1,515 @@
+#!/usr/bin/env python3
+"""Real-data (MNIST) validation of the two core no-fit claims.
+
+Part A — initial-erosion theorem. For any MLP trained with squared loss and
+independent zero-mean feedback, E_B[e_0 | W, r] = hidden BP speed share
+(notes 29/35). The theorem is architecture- and data-agnostic; this script
+checks it on MNIST-subset MLPs by running the *real FA backward pass* for
+every feedback draw.
+
+Part B — early operator-velocity estimator. The no-fit finite-time predictor
+K_hat_t = K_0 + (t/s)(K_s - K_0), rolled through
+r_{t+1} = (I - eta K_hat_t / N) r_t, predicts the T-step FA/BP train gap
+(notes 15-17). This script repeats the synthetic protocol on MNIST subsets:
+empirical full-batch SGD trajectories vs fixed-K(0) and linear-velocity
+predictions, across subset sizes and feedback seeds.
+
+Tangent operators are computed with a layerwise gram identity
+ K[(n,c),(m,c')] = sum_l <delta_l(n,c), delta'_l(m,c')> <h_{l-1}(n), h_{l-1}(m)>
+(delta' = BP deltas for K_BP, FA deltas for K_FA = J J_tilde^T), which avoids
+materializing the (N C) x P Jacobian; `--self-test` verifies it against an
+explicit autograd Jacobian on a tiny network.
+"""
+
+from __future__ import annotations
+
+import argparse
+import csv
+import json
+import math
+import sys
+import time
+from dataclasses import asdict, dataclass
+from pathlib import Path
+
+import matplotlib
+
+matplotlib.use("Agg")
+import matplotlib.pyplot as plt
+import numpy as np
+import torch
+
+SCRIPT_DIR = Path(__file__).resolve().parent
+if str(SCRIPT_DIR) not in sys.path:
+ sys.path.insert(0, str(SCRIPT_DIR))
+
+import downstream_capacity_sweep as dcs # noqa: E402
+
+Tensor = torch.Tensor
+
+
+# ---------------------------------------------------------------------------
+# data and model utilities
+# ---------------------------------------------------------------------------
+
+def parse_args() -> argparse.Namespace:
+ p = argparse.ArgumentParser(description="MNIST validation of e0 theorem and estimator.")
+ p.add_argument("--part", choices=["a", "b", "both"], default="both")
+ p.add_argument("--self-test", action="store_true", help="verify kernel code and exit")
+ p.add_argument("--data-root", type=Path, default=Path("data"))
+ # part A
+ p.add_argument("--e0-widths", type=int, nargs="+", default=[64, 256])
+ p.add_argument("--e0-depths", type=int, nargs="+", default=[1, 2])
+ p.add_argument("--e0-init-seeds", type=int, default=4)
+ p.add_argument("--e0-train-samples", type=int, default=256)
+ p.add_argument("--feedback-draws", type=int, default=512)
+ # part B
+ p.add_argument("--est-widths", type=int, nargs="+", default=[128, 256])
+ p.add_argument("--est-depth", type=int, default=2)
+ p.add_argument("--est-train-sizes", type=int, nargs="+", default=[128, 256, 512])
+ p.add_argument("--est-init-seeds", type=int, default=2)
+ p.add_argument("--est-feedback-seeds", type=int, default=8)
+ p.add_argument("--horizon", type=int, default=50)
+ p.add_argument("--early-step", type=int, default=20)
+ p.add_argument("--lr-safety", type=float, default=0.3,
+ help="lr = safety * N / lam_max(K_BP_0)")
+ p.add_argument("--feedback-scale", choices=["relu", "fan-in", "unit"], default="relu")
+ p.add_argument("--torch-threads", type=int, default=16)
+ p.add_argument("--outdir", type=Path, default=Path("outputs/real_data_validation_mnist"))
+ return p.parse_args()
+
+
+def load_mnist_subset(root: Path, samples: int, seed: int = 0) -> tuple[Tensor, Tensor]:
+ """Flattened, standardized MNIST images with one-hot regression targets."""
+ from torchvision import datasets
+
+ train = datasets.MNIST(root=str(root), train=True, download=True)
+ images = train.data.to(torch.float64) / 255.0
+ labels = train.targets
+ generator = torch.Generator().manual_seed(seed)
+ index = torch.randperm(images.shape[0], generator=generator)[:samples]
+ x = images[index].reshape(samples, -1)
+ x = (x - 0.1307) / 0.3081
+ x = x / math.sqrt(x.shape[1]) # unit-scale rows so kernel scales match synthetic runs
+ y = torch.zeros(samples, 10, dtype=torch.float64)
+ y[torch.arange(samples), labels[index]] = 1.0
+ return x, y
+
+
+def init_weights_dims(dims: list[int], seed: int) -> list[Tensor]:
+ """He init for hidden layers, 1/sqrt(fan_in) for output (project convention)."""
+ generator = torch.Generator(device="cpu")
+ generator.manual_seed(seed)
+ weights: list[Tensor] = []
+ for layer, (fan_in, fan_out) in enumerate(zip(dims[:-1], dims[1:])):
+ scale = math.sqrt(2.0 / fan_in) if layer < len(dims) - 2 else 1.0 / math.sqrt(fan_in)
+ weights.append(torch.randn(fan_out, fan_in, generator=generator,
+ dtype=torch.float64) * scale)
+ return weights
+
+
+def init_feedback_dims(dims: list[int], seed: int, mode: str) -> list[Tensor]:
+ """feedback[l] has shape (dims[l+1], dims[l+2]) and replaces W_{l+1}^T."""
+ generator = torch.Generator(device="cpu")
+ generator.manual_seed(seed)
+ feedback: list[Tensor] = []
+ for l in range(len(dims) - 2):
+ rows, cols = dims[l + 1], dims[l + 2]
+ feedback.append(torch.randn(rows, cols, generator=generator, dtype=torch.float64)
+ * dcs.feedback_scale(rows, mode))
+ return feedback
+
+
+def speeds(weights: list[Tensor], x: Tensor, y: Tensor,
+ feedback: list[Tensor] | None) -> tuple[float, float, float]:
+ """(speed_BP, speed_FA, output_speed); speed_FA = sum_l <g_l^BP, g_l^FA>."""
+ bp = dcs.gradients(weights, x, y, None)
+ speed_bp = float(sum((g * g).sum() for g in bp))
+ output_speed = float((bp[-1] * bp[-1]).sum())
+ if feedback is None:
+ return speed_bp, speed_bp, output_speed
+ fa = dcs.gradients(weights, x, y, feedback)
+ speed_fa = float(sum((gb * gf).sum() for gb, gf in zip(bp, fa)))
+ return speed_bp, speed_fa, output_speed
+
+
+# ---------------------------------------------------------------------------
+# tangent operators via the layerwise gram identity
+# ---------------------------------------------------------------------------
+
+def collect_states(weights: list[Tensor], x: Tensor) -> tuple[list[Tensor], list[Tensor]]:
+ activations, preacts = dcs.forward(weights, x)
+ gates = [(preacts[l] > 0).to(torch.float64) for l in range(len(weights) - 1)]
+ return activations, gates
+
+
+def output_deltas(weights: list[Tensor], gates: list[Tensor],
+ feedback: list[Tensor] | None) -> list[Tensor]:
+ """D[l]: (N, C, dims[l+1]) per-output-coordinate deltas at layer l."""
+ n = gates[0].shape[0]
+ c = weights[-1].shape[0]
+ layers = len(weights)
+ deltas: list[Tensor] = [torch.empty(0)] * layers
+ deltas[layers - 1] = torch.eye(c, dtype=torch.float64).unsqueeze(0).expand(n, c, c).clone()
+ for l in range(layers - 2, -1, -1):
+ if feedback is None:
+ back = deltas[l + 1] @ weights[l + 1]
+ else:
+ back = deltas[l + 1] @ feedback[l].T
+ deltas[l] = back * gates[l].unsqueeze(1)
+ return deltas
+
+
+def tangent_kernel(weights: list[Tensor], x: Tensor,
+ feedback: list[Tensor] | None) -> Tensor:
+ """K_BP = J J^T (feedback=None) or K_FA = J J_tilde^T, shape (NC, NC)."""
+ activations, gates = collect_states(weights, x)
+ d_bp = output_deltas(weights, gates, None)
+ d_rule = d_bp if feedback is None else output_deltas(weights, gates, feedback)
+ n, c = x.shape[0], weights[-1].shape[0]
+ kernel = torch.zeros(n * c, n * c, dtype=torch.float64)
+ for l in range(len(weights)):
+ left = d_bp[l].reshape(n * c, -1)
+ right = d_rule[l].reshape(n * c, -1)
+ gram = activations[l] @ activations[l].T
+ gram_x = gram.repeat_interleave(c, dim=0).repeat_interleave(c, dim=1)
+ kernel += (left @ right.T) * gram_x
+ return kernel
+
+
+def explicit_jacobians(weights: list[Tensor], x: Tensor,
+ feedback: list[Tensor] | None) -> Tensor:
+ """Explicit (NC, P) pseudo-Jacobian via per-coordinate deltas (reference)."""
+ activations, gates = collect_states(weights, x)
+ deltas = output_deltas(weights, gates, feedback)
+ rows = []
+ n, c = x.shape[0], weights[-1].shape[0]
+ for sample in range(n):
+ for coord in range(c):
+ parts = [torch.outer(deltas[l][sample, coord], activations[l][sample]).reshape(-1)
+ for l in range(len(weights))]
+ rows.append(torch.cat(parts))
+ return torch.stack(rows, dim=0)
+
+
+def autograd_jacobian(weights: list[Tensor], x: Tensor) -> Tensor:
+ leaves = [w.clone().requires_grad_(True) for w in weights]
+ pred = dcs.predict(leaves, x).reshape(-1)
+ rows = []
+ for i in range(pred.numel()):
+ grads = torch.autograd.grad(pred[i], leaves, retain_graph=True)
+ rows.append(torch.cat([g.reshape(-1) for g in grads]))
+ return torch.stack(rows, dim=0)
+
+
+def self_test() -> None:
+ torch.manual_seed(0)
+ dims = [5, 7, 6, 3]
+ weights = init_weights_dims(dims, seed=11)
+ feedback = init_feedback_dims(dims, seed=21, mode="relu")
+ x = torch.randn(4, dims[0], dtype=torch.float64)
+
+ jac_auto = autograd_jacobian(weights, x)
+ jac_manual = explicit_jacobians(weights, x, None)
+ err_j = float((jac_auto - jac_manual).abs().max())
+
+ k_bp_gram = tangent_kernel(weights, x, None)
+ k_bp_ref = jac_auto @ jac_auto.T
+ err_bp = float((k_bp_gram - k_bp_ref).abs().max())
+
+ jac_fa = explicit_jacobians(weights, x, feedback)
+ k_fa_gram = tangent_kernel(weights, x, feedback)
+ k_fa_ref = jac_auto @ jac_fa.T
+ err_fa = float((k_fa_gram - k_fa_ref).abs().max())
+
+ # gradient consistency: J^T r and J_tilde^T r reproduce dcs.gradients
+ y = torch.randn(4, dims[-1], dtype=torch.float64)
+ r = (dcs.predict(weights, x) - y).reshape(-1)
+ batch = x.shape[0]
+ g_bp_vec = jac_auto.T @ (r / batch)
+ g_bp = torch.cat([g.reshape(-1) for g in dcs.gradients(weights, x, y, None)])
+ err_g_bp = float((g_bp_vec - g_bp).abs().max())
+ g_fa_vec = jac_fa.T @ (r / batch)
+ g_fa = torch.cat([g.reshape(-1) for g in dcs.gradients(weights, x, y, feedback)])
+ err_g_fa = float((g_fa_vec - g_fa).abs().max())
+
+ print(f"self-test: |J_auto - J_manual|_max = {err_j:.3e}")
+ print(f"self-test: |K_BP_gram - J J^T|_max = {err_bp:.3e}")
+ print(f"self-test: |K_FA_gram - J J_tilde^T|_max = {err_fa:.3e}")
+ print(f"self-test: |J^T r - g_BP|_max = {err_g_bp:.3e}")
+ print(f"self-test: |J_tilde^T r - g_FA|_max = {err_g_fa:.3e}")
+ tol = 1e-9
+ assert max(err_j, err_bp, err_fa, err_g_bp, err_g_fa) < tol, "self-test FAILED"
+ print("self-test PASSED")
+
+
+# ---------------------------------------------------------------------------
+# part A: e0 theorem on MNIST
+# ---------------------------------------------------------------------------
+
+@dataclass(frozen=True)
+class E0Row:
+ width: int
+ depth: int
+ init_seed: int
+ feedback_draws: int
+ hidden_share: float
+ empirical_mean: float
+ empirical_stderr: float
+ empirical_std: float
+ mean_abs_error: float
+
+
+def run_part_a(args: argparse.Namespace, outdir: Path) -> list[E0Row]:
+ x, y = load_mnist_subset(args.data_root, args.e0_train_samples)
+ rows: list[E0Row] = []
+ for depth in args.e0_depths:
+ for width in args.e0_widths:
+ dims = [x.shape[1], *([width] * depth), y.shape[1]]
+ for init_seed in range(args.e0_init_seeds):
+ weights = init_weights_dims(dims, seed=10_000 + init_seed)
+ speed_bp, _, output_speed = speeds(weights, x, y, None)
+ hidden_share = 1.0 - output_speed / speed_bp
+ erosions = []
+ for draw in range(args.feedback_draws):
+ feedback = init_feedback_dims(
+ dims, seed=500_000 + 1_000 * init_seed + draw,
+ mode=args.feedback_scale)
+ _, speed_fa, _ = speeds(weights, x, y, feedback)
+ erosions.append(1.0 - speed_fa / speed_bp)
+ erosions_arr = np.array(erosions)
+ row = E0Row(
+ width=width, depth=depth, init_seed=init_seed,
+ feedback_draws=args.feedback_draws,
+ hidden_share=hidden_share,
+ empirical_mean=float(erosions_arr.mean()),
+ empirical_stderr=float(erosions_arr.std(ddof=1) / math.sqrt(len(erosions))),
+ empirical_std=float(erosions_arr.std(ddof=1)),
+ mean_abs_error=float(abs(erosions_arr.mean() - hidden_share)),
+ )
+ rows.append(row)
+ print(f"[A] depth={depth} width={width} init={init_seed}: "
+ f"hidden_share={row.hidden_share:.4f} "
+ f"E_B[e0]={row.empirical_mean:.4f} +/- {row.empirical_stderr:.4f} "
+ f"|diff|={row.mean_abs_error:.4f}", flush=True)
+
+ with (outdir / "e0_mnist_rows.csv").open("w", newline="") as fh:
+ writer = csv.DictWriter(fh, fieldnames=list(E0Row.__annotations__.keys()))
+ writer.writeheader()
+ for row in rows:
+ writer.writerow(asdict(row))
+
+ fig, ax = plt.subplots(figsize=(5.6, 5.2), dpi=180)
+ markers = {1: "o", 2: "s", 3: "^"}
+ for depth in sorted({row.depth for row in rows}):
+ sub = [row for row in rows if row.depth == depth]
+ ax.errorbar([row.hidden_share for row in sub],
+ [row.empirical_mean for row in sub],
+ yerr=[2 * row.empirical_stderr for row in sub],
+ fmt=markers.get(depth, "o"), ms=5.5, capsize=2.0, lw=0, elinewidth=1.0,
+ label=f"{depth} hidden layer{'s' if depth > 1 else ''}")
+ lo = min(min(row.hidden_share for row in rows), min(row.empirical_mean for row in rows))
+ hi = max(max(row.hidden_share for row in rows), max(row.empirical_mean for row in rows))
+ pad = 0.04 * (hi - lo + 1e-9)
+ ax.plot([lo - pad, hi + pad], [lo - pad, hi + pad], color="black", lw=1.0)
+ ax.set_xlabel("predicted: hidden BP speed share (one BP pass, no fit)")
+ ax.set_ylabel("measured: E_B[e0] over real FA backward draws")
+ ax.set_title("MNIST: initial FA erosion = hidden speed share")
+ ax.legend(fontsize=8)
+ ax.grid(alpha=0.16)
+ fig.tight_layout()
+ fig.savefig(outdir / "e0_mnist_calibration.png", bbox_inches="tight")
+ plt.close(fig)
+ return rows
+
+
+# ---------------------------------------------------------------------------
+# part B: early operator-velocity estimator on MNIST
+# ---------------------------------------------------------------------------
+
+@dataclass(frozen=True)
+class EstimatorRow:
+ width: int
+ depth: int
+ train_samples: int
+ init_seed: int
+ feedback_seed: int
+ lr: float
+ lam_max: float
+ horizon: int
+ early_step: int
+ hidden_share: float
+ loss_bp_emp: float
+ loss_fa_emp: float
+ gap_emp: float
+ gap_fixed: float
+ gap_velocity: float
+
+
+def power_lam_max(kernel: Tensor, iters: int = 60) -> float:
+ vec = torch.randn(kernel.shape[0], dtype=torch.float64)
+ vec /= vec.norm()
+ lam = 0.0
+ for _ in range(iters):
+ nxt = kernel @ vec
+ lam = float(nxt.norm())
+ vec = nxt / (lam + 1e-30)
+ return lam
+
+
+def train_with_snapshots(weights0: list[Tensor], x: Tensor, y: Tensor, lr: float,
+ steps: int, feedback: list[Tensor] | None,
+ snap_steps: set[int]) -> tuple[list[Tensor], dict[int, list[Tensor]]]:
+ weights = dcs.clone_weights(weights0)
+ snaps: dict[int, list[Tensor]] = {}
+ for t in range(steps + 1):
+ if t in snap_steps:
+ snaps[t] = dcs.clone_weights(weights)
+ if t == steps:
+ break
+ grads = dcs.gradients(weights, x, y, feedback)
+ weights = [w - lr * g for w, g in zip(weights, grads)]
+ return weights, snaps
+
+
+def rollout_loss(k_zero: Tensor, k_slope: Tensor | None, r0: Tensor, lr: float,
+ n: int, steps: int, early_step: int) -> float:
+ """Roll r_{t+1} = r_t - (lr/N) K_hat_t r_t with K_hat_t = K0 + (t/s) dK."""
+ r = r0.clone()
+ for t in range(steps):
+ kr = k_zero @ r
+ if k_slope is not None:
+ kr = kr + (t / early_step) * (k_slope @ r)
+ r = r - (lr / n) * kr
+ return float(r.pow(2).sum() / (2 * n))
+
+
+def run_part_b(args: argparse.Namespace, outdir: Path) -> list[EstimatorRow]:
+ rows: list[EstimatorRow] = []
+ t_total, s_step = args.horizon, args.early_step
+ for est_width, n_train in [(w, n) for w in args.est_widths
+ for n in args.est_train_sizes]:
+ x, y = load_mnist_subset(args.data_root, n_train)
+ dims = [x.shape[1], *([est_width] * args.est_depth), y.shape[1]]
+ n = x.shape[0]
+ for init_seed in range(args.est_init_seeds):
+ weights0 = init_weights_dims(dims, seed=10_000 + init_seed)
+ r0 = (dcs.predict(weights0, x) - y).reshape(-1)
+ k_bp_0 = tangent_kernel(weights0, x, None)
+ lam_max = power_lam_max(k_bp_0)
+ lr = args.lr_safety * n / lam_max
+ speed_bp, _, output_speed = speeds(weights0, x, y, None)
+ hidden_share = 1.0 - output_speed / speed_bp
+
+ t0 = time.time()
+ bp_final, bp_snaps = train_with_snapshots(
+ weights0, x, y, lr, t_total, None, {0, s_step})
+ loss_bp_emp = dcs.mse(bp_final, x, y)
+ k_bp_s = tangent_kernel(bp_snaps[s_step], x, None)
+ bp_loss_fixed = rollout_loss(k_bp_0, None, r0, lr, n, t_total, s_step)
+ bp_loss_vel = rollout_loss(k_bp_0, k_bp_s - k_bp_0, r0, lr, n, t_total, s_step)
+
+ for feedback_index in range(args.est_feedback_seeds):
+ feedback_seed = 100_000 + init_seed * 1_000 + feedback_index
+ feedback = init_feedback_dims(dims, feedback_seed, args.feedback_scale)
+ fa_final, fa_snaps = train_with_snapshots(
+ weights0, x, y, lr, t_total, feedback, {0, s_step})
+ loss_fa_emp = dcs.mse(fa_final, x, y)
+ k_fa_0 = tangent_kernel(weights0, x, feedback)
+ k_fa_s = tangent_kernel(fa_snaps[s_step], x, feedback)
+ fa_loss_fixed = rollout_loss(k_fa_0, None, r0, lr, n, t_total, s_step)
+ fa_loss_vel = rollout_loss(k_fa_0, k_fa_s - k_fa_0, r0, lr, n, t_total, s_step)
+ row = EstimatorRow(
+ width=est_width, depth=args.est_depth, train_samples=n_train,
+ init_seed=init_seed, feedback_seed=feedback_seed, lr=lr,
+ lam_max=lam_max, horizon=t_total, early_step=s_step,
+ hidden_share=hidden_share,
+ loss_bp_emp=loss_bp_emp, loss_fa_emp=loss_fa_emp,
+ gap_emp=loss_fa_emp - loss_bp_emp,
+ gap_fixed=fa_loss_fixed - bp_loss_fixed,
+ gap_velocity=fa_loss_vel - bp_loss_vel,
+ )
+ rows.append(row)
+ print(f"[B] w={est_width} N={n_train} init={init_seed} fb={feedback_index}: "
+ f"gap_emp={row.gap_emp:.5f} fixed={row.gap_fixed:.5f} "
+ f"velocity={row.gap_velocity:.5f}", flush=True)
+ print(f"[B] w={est_width} N={n_train} init={init_seed} block done "
+ f"in {time.time() - t0:.1f}s", flush=True)
+
+ with (outdir / "estimator_mnist_rows.csv").open("w", newline="") as fh:
+ writer = csv.DictWriter(fh, fieldnames=list(EstimatorRow.__annotations__.keys()))
+ writer.writeheader()
+ for row in rows:
+ writer.writerow(asdict(row))
+
+ emp = np.array([row.gap_emp for row in rows])
+ metrics = []
+ for name, pred in [("fixed K(0)", np.array([row.gap_fixed for row in rows])),
+ ("linear velocity", np.array([row.gap_velocity for row in rows]))]:
+ metrics.append({
+ "predictor": name,
+ "rows": len(rows),
+ "mae": float(np.mean(np.abs(pred - emp))),
+ "bias": float(np.mean(pred - emp)),
+ "corr": float(np.corrcoef(pred, emp)[0, 1]),
+ })
+ with (outdir / "estimator_mnist_metrics.csv").open("w", newline="") as fh:
+ writer = csv.DictWriter(fh, fieldnames=list(metrics[0].keys()))
+ writer.writeheader()
+ writer.writerows(metrics)
+ for metric in metrics:
+ print(f"[B] {metric['predictor']}: MAE={metric['mae']:.5f} "
+ f"bias={metric['bias']:+.5f} corr={metric['corr']:.5f}", flush=True)
+
+ fig, ax = plt.subplots(figsize=(5.8, 5.4), dpi=180)
+ lims = [min(emp.min(), 0.0), max(emp.max(), 0.0)]
+ pad = 0.06 * (lims[1] - lims[0] + 1e-9)
+ ax.plot([lims[0] - pad, lims[1] + pad], [lims[0] - pad, lims[1] + pad],
+ color="black", lw=1.0)
+ ax.scatter([row.gap_fixed for row in rows], emp, s=22, alpha=0.65,
+ color="#9aa5b1", marker="^",
+ label=f"fixed K(0) (MAE {metrics[0]['mae']:.4f})")
+ ax.scatter([row.gap_velocity for row in rows], emp, s=22, alpha=0.8,
+ color="#c65f16", marker="o",
+ label=f"early velocity (MAE {metrics[1]['mae']:.4f}, corr {metrics[1]['corr']:.3f})")
+ ax.set_xlabel("predicted FA/BP train gap at T (no fit)")
+ ax.set_ylabel("empirical FA/BP train gap at T")
+ ax.set_title(f"MNIST: finite-time gap prediction\n"
+ f"widths {args.est_widths}, depth {args.est_depth}, "
+ f"T={t_total}, s={s_step}, N in {args.est_train_sizes}")
+ ax.legend(fontsize=8)
+ ax.grid(alpha=0.16)
+ fig.tight_layout()
+ fig.savefig(outdir / "estimator_mnist_scatter.png", bbox_inches="tight")
+ plt.close(fig)
+ return rows
+
+
+def main() -> None:
+ args = parse_args()
+ torch.set_num_threads(args.torch_threads)
+ if args.self_test:
+ self_test()
+ return
+ args.outdir.mkdir(parents=True, exist_ok=True)
+ payload: dict = {"config": {k: (str(v) if isinstance(v, Path) else v)
+ for k, v in vars(args).items()}}
+ if args.part in ("a", "both"):
+ rows_a = run_part_a(args, args.outdir)
+ payload["e0_max_abs_error"] = max(row.mean_abs_error for row in rows_a)
+ payload["e0_rows"] = len(rows_a)
+ if args.part in ("b", "both"):
+ rows_b = run_part_b(args, args.outdir)
+ emp = np.array([row.gap_emp for row in rows_b])
+ vel = np.array([row.gap_velocity for row in rows_b])
+ fixed = np.array([row.gap_fixed for row in rows_b])
+ payload["estimator_velocity_corr"] = float(np.corrcoef(vel, emp)[0, 1])
+ payload["estimator_velocity_mae"] = float(np.mean(np.abs(vel - emp)))
+ payload["estimator_fixed_mae"] = float(np.mean(np.abs(fixed - emp)))
+ payload["estimator_rows"] = len(rows_b)
+ (args.outdir / "summary.json").write_text(json.dumps(payload, indent=2) + "\n")
+ print(f"summary: {args.outdir / 'summary.json'}")
+
+
+if __name__ == "__main__":
+ main()