import json, os, random from collections import deque import numpy as np import torch import sys sys.path.insert(0, '/home/maxwelhelp/all/math2nn') import bench SEEDS = tuple(range(8)) # Shared union: every idea learning rate is also evaluated by baseline. GRID = [ {'lr': 0.001}, {'lr': 0.003}, {'lr': 0.006}, ] EPOCHS = 8 NTRAIN, NTEST = 400, 400 class HankelResidualObserver: def __init__(self, t_ini=8, horizon=1, max_cols=40, ridge=1e-2, tau=1.0): self.t_ini, self.horizon = t_ini, horizon self.max_cols, self.ridge, self.tau = max_cols, ridge, tau self.r = deque(maxlen=t_ini + horizon + max_cols + 8) self.last = None self.last_q = None def update(self, residual): self.r.append(float(residual)) def predict(self): if len(self.r) < self.t_ini + self.horizon: return 0.0, 0.0, None a = np.asarray(self.r, dtype=np.float64) n = len(a) - self.t_ini - self.horizon + 1 starts = np.arange(max(0, n-self.max_cols), n) dp = np.stack([a[j:j+self.t_ini] for j in starts], axis=1) df = np.stack([a[j+self.t_ini:j+self.t_ini+self.horizon] for j in starts], axis=1) d = a[-self.t_ini:] gram = dp.T @ dp + self.ridge*np.eye(dp.shape[1]) g = np.linalg.solve(gram, dp.T @ d) recon = dp @ g q = np.linalg.norm(recon-d)/(np.linalg.norm(d)+1e-8) gamma = float(np.clip(1.0-q/self.tau, 0.0, 1.0)) pred = float((df @ g)[0]) * gamma return pred, gamma, float(q) def seed_all(seed): random.seed(seed); np.random.seed(seed); torch.manual_seed(seed) if torch.cuda.is_available(): torch.cuda.manual_seed_all(seed) def fit(seed, lr): seed_all(seed) ds = bench.get_dataset('sequence', seed, n_train=NTRAIN, n_test=NTEST) model = bench.make_model('transformer_tiny', ds['input_shape'], ds['out_dim']) net, metric, hist = bench.train_model(model, ds, epochs=EPOCHS, lr=lr, batch=128, weight_decay=0.0, log=lambda _: None) if net is None: raise RuntimeError('bench training failed') return net, ds def score(seed, lr, observer_on=False, collect_signature=False): net, ds = fit(seed, lr) net.eval() model_device = next(net.parameters()).device x, y = ds['xte'], ds['yte'] with torch.no_grad(): # Keep batches small and use the harness-produced trained model. pred = [] for i in range(0, len(x), 128): pred.append(net(x[i:i+128].to(model_device)).detach().cpu().numpy().reshape(-1)) base = np.concatenate(pred).astype(np.float64) actual = y.detach().cpu().numpy().reshape(-1).astype(np.float64) if not observer_on: return float(np.mean((base-actual)**2)), None obs = HankelResidualObserver() corrected, gammas, qs, residual_forecasts, next_residuals = [], [], [], [], [] # Strictly causal: residual at sample i is appended only after predicting i. for i in range(len(actual)): dh, gamma, q = obs.predict() corrected.append(base[i] + dh) gammas.append(gamma); qs.append(q if q is not None else np.nan) if q is not None: residual_forecasts.append(dh/gamma if gamma > 1e-12 else 0.0) next_residuals.append(actual[i]-base[i]) obs.update(actual[i]-base[i]) corr = np.asarray(corrected) result = float(np.mean((corr-actual)**2)) sig = { 'residual_forecast_mse': float(np.mean((np.asarray(residual_forecasts)-np.asarray(next_residuals))**2)) if next_residuals else None, 'residual_naive_mse': float(np.mean(np.asarray(next_residuals)**2)) if next_residuals else None, 'correction_mse': result, 'base_mse': float(np.mean((base-actual)**2)), 'mean_gamma': float(np.mean(gammas)), 'active_fraction': float(np.mean(np.asarray(gammas)>0)), 'mean_mismatch_q': float(np.nanmean(qs)), } # Quantitative confirmation requires both causal residual predictability and task improvement. sig['confirmed'] = bool(sig['residual_forecast_mse'] < sig['residual_naive_mse'] and result < sig['base_mse']) return result, sig def eval_cfg(cfg, idea=False, signature=False): vals, sigs = [], [] for s in SEEDS: v, sig = score(s, cfg['lr'], observer_on=idea, collect_signature=signature) vals.append(v) if sig is not None: sigs.append(sig) out = {'mean': float(np.mean(vals)), 'std': float(np.std(vals)), 'per_seed': [float(v) for v in vals], 'n': len(vals)} if sigs: keys = ['residual_forecast_mse','residual_naive_mse','correction_mse','base_mse','mean_gamma','active_fraction','mean_mismatch_q'] out['signature_summary'] = {k: float(np.mean([z[k] for z in sigs])) for k in keys} return out def baseline_factory(cfg): return lambda seed: score(seed, cfg['lr'], observer_on=False)[0] def main(): # sweep_baseline itself is the required baseline selection mechanism; use all 8 seeds. base = bench.sweep_baseline(baseline_factory, GRID, seeds=SEEDS) idea_trials = [] for cfg in GRID: r = eval_cfg(cfg, idea=True) idea_trials.append({'cfg': cfg, **r}) best = min(idea_trials, key=lambda z: z['mean']) idea = {k: best[k] for k in ('mean','std','per_seed','n')} sig = eval_cfg(best['cfg'], idea=True, signature=True)['signature_summary'] # make_report computes paired deltas and the mandated sign-permutation p-value. report = bench.make_report('sequence', 'transformer_tiny', base, idea, {'mechanism_signature': {**sig, 'confirmed': bool(sig['residual_forecast_mse'] < sig['residual_naive_mse'] and sig['correction_mse'] < sig['base_mse'])}, 'idea_sweep': [{'cfg': z['cfg'], 'mean': z['mean'], 'std': z['std']} for z in idea_trials], 'protocol': {'seeds': list(SEEDS), 'epochs': EPOCHS, 'n_train': NTRAIN, 'n_test': NTEST, 'shared_lr_grid': GRID, 'observer': {'t_ini':8,'horizon':1,'max_cols':40,'ridge':1e-2,'tau':1.0}}}) with open('bench_report.json','w') as f: json.dump(report, f, indent=2) print(json.dumps(report, indent=2)) if __name__ == '__main__': main()