diff options
| author | YurenHao0426 <Blackhao0426@gmail.com> | 2026-06-09 15:28:30 -0500 |
|---|---|---|
| committer | YurenHao0426 <Blackhao0426@gmail.com> | 2026-06-09 15:28:30 -0500 |
| commit | b246cc878c3db48e6d993a6a293176ee9be97e77 (patch) | |
| tree | ed7baf048144927ba98f417e5d9c28b5a4489f31 | |
| parent | 2488deff02737330ff4f355746be2b49492ac3dc (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.md | 9 | ||||
| -rw-r--r-- | notes/38_mnist_real_data_validation.md | 99 | ||||
| -rw-r--r-- | scripts/real_data_validation.py | 515 |
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() |
