import sys, json, math from pathlib import Path import numpy as np import torch sys.path.insert(0, '/home/maxwelhelp/all/math2nn') from bench import get_dataset, make_model, train_model, sweep_baseline, make_report SEEDS = tuple(range(8)) GRID = [ {'lr': 0.0015, 'weight_decay': 0.0, 'epochs': 12}, {'lr': 0.0030, 'weight_decay': 0.0, 'epochs': 12}, {'lr': 0.0060, 'weight_decay': 0.0, 'epochs': 12}, {'lr': 0.0015, 'weight_decay': 1e-4, 'epochs': 12}, {'lr': 0.0030, 'weight_decay': 1e-4, 'epochs': 12}, {'lr': 0.0060, 'weight_decay': 1e-4, 'epochs': 12}, ] def entropy(y, z): y = np.asarray(y, int); z = np.asarray(z, int) if z.ndim == 1: z = z[:, None] keys = np.zeros(len(y), dtype=np.int64) mult = 1 for c in range(z.shape[1]): keys += z[:, c] * mult; mult *= int(z[:, c].max() + 1) _, inv = np.unique(keys, return_inverse=True) ny = int(y.max()) + 1 tab = np.zeros((int(inv.max()) + 1, ny), dtype=np.int64) np.add.at(tab, (inv, y), 1) p = tab / np.maximum(tab.sum(1, keepdims=True), 1) h = -(np.where(p > 0, p * np.log(np.maximum(p, 1e-300)), 0)).sum(1) return float((tab.sum(1) / len(y) * h).sum()) def quantile_codes(a, q): a = np.asarray(a) out = np.empty_like(a, dtype=np.int64) for j in range(a.shape[1]): edges = np.unique(np.quantile(a[:, j], np.linspace(0, 1, q + 1)[1:-1])) out[:, j] = np.searchsorted(edges, a[:, j], side='right') return out def selector(xtr, ytr, seed): # Fixed, train-only quantile discretization for structure discovery. qx, qy = 4, 8 xd = quantile_codes(xtr, qx) ye = quantile_codes(ytr.reshape(-1, 1), qy).ravel() h0 = entropy(ye, np.zeros((len(ye), 1), dtype=np.int64)) sh = np.array([h0 - entropy(ye, xd[:, [j]]) for j in range(xtr.shape[1])]) rng = np.random.default_rng(1000 + seed) boot = [] for _ in range(24): ix = rng.integers(0, len(ye), len(ye)) hb = entropy(ye[ix], np.zeros((len(ix), 1), dtype=np.int64)) boot.append([hb - entropy(ye[ix], xd[ix, [j]]) for j in range(xtr.shape[1])]) boot = np.asarray(boot) jstar = int(np.argmax(sh)) gamma = float(sh[jstar] / (boot[:, jstar].std(ddof=1) + 1e-8)) # Held-out predictive risk: polynomial ridge captures Friedman nonlinearities # while avoiding a different neural architecture during final training. n = len(xtr); cut = max(1, int(.75*n)); a, b = xtr[:cut], xtr[cut:] ya, yb = ytr[:cut], ytr[cut:] def feat(x, j): v = x[:, j:j+1]; return np.concatenate([v, v*v, np.sin(v)], axis=1) sr = [] for j in range(xtr.shape[1]): z, zv = feat(a, j), feat(b, j) reg = 1e-2 * np.eye(z.shape[1]); w = np.linalg.solve(z.T@z + reg, z.T@ya) sr.append(float(np.mean((yb - zv@w)**2))) sr = -np.asarray(sr) # larger is better, as S_R relative to common null risk mh = np.argsort(sh)[-5:][::-1] mr = np.argsort(sr)[-5:][::-1] kappa = float(h0 / math.log(qy)) # qY=8 makes kappa operationally sensitive to whether bins resolve residual uncertainty. use_h = bool(kappa < 0.90 and gamma > 2.0) chosen = mh if use_h else mr return {'mask_h': mh.tolist(), 'mask_r': mr.tolist(), 'mask': chosen.tolist(), 'sh': sh.tolist(), 'sr': sr.tolist(), 'kappa': kappa, 'gamma': gamma, 'used_entropy': use_h, 'h0': h0} def run_one(track, seed, cfg, idea): torch.manual_seed(seed); np.random.seed(seed) d = get_dataset(track, seed, n_train=1200, n_test=500) meta = selector(d['xtr'].numpy(), d['ytr'].numpy(), seed) if idea else None if idea: cols = meta['mask'] else: cols = list(range(d['xtr'].shape[1])) ds = dict(d) ds['xtr'], ds['xte'] = d['xtr'][:, cols], d['xte'][:, cols] ds['input_shape'] = tuple(ds['xtr'].shape[1:]) net = make_model('mlp_tiny', ds['input_shape'], ds['out_dim']) _, metric, _ = train_model(net, ds, epochs=cfg['epochs'], lr=cfg['lr'], weight_decay=cfg['weight_decay'], log=lambda *_: None) return float(metric), meta def fn(track, cfg, idea): return lambda seed: run_one(track, seed, cfg, idea)[0] def main(): track, model = 'tabular', 'mlp_tiny' baseline = sweep_baseline(lambda c: fn(track, c, False), GRID, seeds=SEEDS) # Equal-sized idea sweep over exactly the baseline union; final result is best config. tried=[] for cfg in GRID: r = __import__('bench').evaluate(fn(track, cfg, True), SEEDS) tried.append({'cfg':cfg, 'mean':r['mean']}) best_cfg = min(GRID, key=lambda c: next(x['mean'] for x in tried if x['cfg']==c)) idea = __import__('bench').evaluate(fn(track, best_cfg, True), SEEDS) # Test-model signature: selector's predicted choice and the observed trained-model metric. sig=[] for s in SEEDS: _, m = run_one(track, s, best_cfg, True) full = run_one(track, s, best_cfg, False)[0] chosen = run_one(track, s, best_cfg, True)[0] sig.append({'seed':s, 'kappa':m['kappa'], 'gamma':m['gamma'], 'used_entropy':bool(m['used_entropy']), 'predicted_fallback':bool(not m['used_entropy']), 'observed_subset_minus_full_mse':chosen-full}) # Signature prediction is that high kappa/high stochasticity triggers fallback; verify on trained outcomes. high = [x for x in sig if x['kappa'] >= .90] confirmed = bool(high) and all(x['predicted_fallback'] for x in high) and np.mean([x['observed_subset_minus_full_mse'] for x in high]) <= 0.02 idea_block={'best_cfg':best_cfg, 'sweep':tried, 'full':idea} report=make_report(track, model, baseline, idea, { 'mechanism_signature': {'prediction':'kappa>=0.90 routes away from entropy mask and avoids materially worse trained-model MSE', 'observed':sig, 'high_kappa_n':len(high), 'high_kappa_observed_mean_delta':float(np.mean([x['observed_subset_minus_full_mse'] for x in high])) if high else None, 'confirmed':confirmed}, 'selection_details':'Entropy and validation-risk masks selected from train split; final systems use identical mlp_tiny/training.'}) Path('bench_report.json').write_text(json.dumps(report, indent=2, default=lambda o: o.item() if isinstance(o, np.generic) else str(o))) print(json.dumps(report, indent=2, default=lambda o: o.item() if isinstance(o, np.generic) else str(o))) if __name__=='__main__': main()