import sys, json, math, time, random from pathlib import Path import numpy as np import torch import torch.nn as nn sys.path.insert(0, '/home/maxwelhelp/all/math2nn') from bench import get_dataset, train_model, sweep_baseline, evaluate, make_report SEEDS = tuple(range(8)) EPOCHS = 8 NTR, NTE = 800, 300 def tail(z, n): term, partial = 1.0, 1.0 for k in range(1, n+1): term *= z/k; partial += term return math.exp(z)-partial def lie_dim(k, d): def mu(q): x=q; p=2; nd=0 while p*p <= x: if x%p==0: x//=p; nd+=1 if x%p==0: return 0 while x%p==0: x//=p p+=1 if x>1: nd+=1 return -1 if nd%2 else 1 return sum(mu(q)*d**(k//q) for q in range(1,k+1) if k%q==0)//k def sanity(): # Cheap numerical verification before any neural training. deg = [[tail(z,n) for n in range(1,6)] for z in (.5, 1., 2.)] mono_n = all(a>b for row in deg for a,b in zip(row,row[1:])) ms = [m*tail(6/m,2) for m in (1,2,4,8,16)] mono_m = all(a>b for a,b in zip(ms,ms[1:])) costs = [m*sum(lie_dim(k,2) for k in range(1,n+1)) for m,n in ((1,3),(2,2),(4,2),(8,1))] return {'tail_decreases_with_degree': mono_n, 'equal_variation_error_decreases_with_m': mono_m, 'tail_values': deg, 'equal_m_values': ms, 'lie_dims_d2': [lie_dim(k,2) for k in range(1,6)], 'cost_examples': costs} class TokenTransformer(nn.Module): def __init__(self, feat, tokens, d=48): super().__init__(); self.inp=nn.Linear(feat,d) self.pos=nn.Parameter(torch.zeros(1,tokens,d)); nn.init.normal_(self.pos,std=.02) layer=nn.TransformerEncoderLayer(d,nhead=2,dim_feedforward=96,batch_first=True,dropout=0.) self.enc=nn.TransformerEncoder(layer,2); self.head=nn.Linear(tokens*d,1) def forward(self,x): h=self.inp(x)+self.pos[:,:x.shape[1]] return self.head(self.enc(h).reshape(x.shape[0],-1)) def local_sig(x, m, adaptive=True): # Add time as a second control channel; degree-2 log signature is its signed area. n=len(x); t=np.linspace(0,1,n,dtype=np.float32) path=np.stack([t,x],1); ds=np.linalg.norm(np.diff(path,axis=0),axis=1) v=np.r_[0.,np.cumsum(ds)]; total=v[-1] if adaptive and total>1e-9: cuts=[0]+[int(np.searchsorted(v,q*total/m,'left')) for q in range(1,m)]+[n-1] else: cuts=list(np.linspace(0,n-1,m+1).astype(int)) cuts=np.maximum.accumulate(np.asarray(cuts)); cuts[-1]=n-1 out=[]; ells=[] for a,b in zip(cuts[:-1],cuts[1:]): dx=np.diff(path[a:b+1],axis=0); inc=path[b]-path[a] pref=np.zeros(2); area=0. for u in dx: area += .5*(pref[0]*u[1]-u[0]*pref[1]); pref += u ell=float(v[b]-v[a]); ells.append(ell) out.append([inc[0],inc[1],area,float(b-a)/max(1,n-1)]) return np.asarray(out,np.float32),np.asarray(ells),cuts def raw_tokens(x,m=8): # Fixed standard patch interface, same four-channel width as idea. n=len(x); out=[] for a,b in zip(np.linspace(0,n,m+1)[:-1].astype(int),np.linspace(0,n,m+1)[1:].astype(int)): b=max(b,a+1); p=x[a:b]; out.append([p.mean(),p[-1]-p[0],p.std(),float(b-a)/n]) return np.asarray(out,np.float32) def transform_ds(ds, kind, m): fn=(lambda x: raw_tokens(x,m)) if kind=='baseline' else (lambda x: local_sig(x,m,True)[0]) tr=np.stack([fn(x) for x in ds['xtr'].numpy()]); te=np.stack([fn(x) for x in ds['xte'].numpy()]) return {'xtr':torch.tensor(tr), 'ytr':ds['ytr'].float(), 'xte':torch.tensor(te), 'yte':ds['yte'].float(), 'task':'regression','metric':'mse','input_shape':tr.shape[1:],'out_dim':1} def run_one(seed, kind, lr, m): torch.manual_seed(seed); np.random.seed(seed); random.seed(seed) ds=get_dataset('sequence',seed,n_train=NTR,n_test=NTE) z=transform_ds(ds,kind,m); net=TokenTransformer(z['input_shape'][1],z['input_shape'][0]) _, metric, _=train_model(net,z,epochs=EPOCHS,lr=lr,batch=128,log=lambda *_:None) return float(metric) def main(): sanity_result=sanity(); print('SANITY',json.dumps(sanity_result)) # Union parity: every idea lr is in baseline grid; baseline method knob m is also swept. grid=[{'lr':1e-3,'m':4},{'lr':3e-3,'m':8},{'lr':6e-3,'m':8}] def maker(cfg): return lambda seed: run_one(seed,'baseline',cfg['lr'],cfg['m']) base=sweep_baseline(maker,grid,seeds=(0,1,2,3)) # Evaluate idea at all three shared settings, then select best by the same four-seed tuning split. idea_trials=[] for cfg in grid: r=evaluate(lambda seed,cfg=cfg: run_one(seed,'idea',cfg['lr'],cfg['m']),seeds=(0,1,2,3)) idea_trials.append({'cfg':cfg,'mean':r['mean']}) best=min(idea_trials,key=lambda q:q['mean'])['cfg'] idea=evaluate(lambda seed: run_one(seed,'idea',best['lr'],best['m']),seeds=SEEDS) # Trained-model behavior signature: observed residual versus the representation's predicted tail budget. vals=[] for s in SEEDS: ds=get_dataset('sequence',s,n_train=NTR,n_test=NTE); z=transform_ds(ds,'idea',best['m']) # proxy uses observed local variation, while residuals are from the trained system. _,metric,_=train_model(TokenTransformer(z['input_shape'][1],z['input_shape'][0]),z,epochs=EPOCHS,lr=best['lr'],batch=128,log=lambda *_:None) e=[] for x in ds['xte'].numpy(): _,ells,_=local_sig(x,best['m'],True); e.append(sum(tail(float(q),2) for q in ells)) vals.append((float(np.mean(e)),float(metric))) corr=float(np.corrcoef(np.asarray(vals).T)[0,1]) if len(vals)>1 else float('nan') sig={'prediction':'larger local Taylor tail should proxy larger CDE approximation error', 'predicted_tail_mean':float(np.mean([v[0] for v in vals])), 'observed_test_mse_mean':float(np.mean([v[1] for v in vals])), 'seed_level_correlation':corr,'confirmed':False, 'note':'Both quantities are measured on trained benchmark systems; eight seed points do not establish the claimed quantitative relation.'} report=make_report('sequence','transformer_tiny',base,idea,{'math_sanity':sanity_result,'idea_trials':idea_trials,'selected_cfg':best,'mechanism_signature':sig}) report['baseline']['idea_union_grid']=grid Path('bench_report.json').write_text(json.dumps(report,indent=2)) print(json.dumps(report,indent=2)) if __name__=='__main__': main()