import os, sys, json, time import numpy as np import torch import torch.nn as nn import torch.nn.functional as F sys.path.insert(0, '/home/maxwelhelp/all/math2nn') from bench import train_model, evaluate, sweep_baseline, make_report # Local custom track: convolution is the relevant mathematical structure. META = {'name':'long_1d_convolution', 'domain':'cnn', 'description':'Synthetic long 1D signal classification requiring learned local convolution.'} def get_dataset(seed, n_train=400, n_test=200): rng=np.random.RandomState(seed) def make(n): x=rng.normal(0,0.35,(n,1,64)).astype('float32') y=rng.randint(0,2,n).astype('int64') # Class-dependent local motifs at random positions, with nuisance low frequency signal. for i in range(n): pos=rng.randint(4,59) motif=np.array([1.,1.,-1.,-1.,1.,1.],dtype='float32') if y[i] else np.array([-1.,1.,1.,-1.,-1.,1.],dtype='float32') x[i,0,pos:pos+6] += 0.9*motif x[i,0] += (0.15*np.sin(np.arange(64)*rng.uniform(.08,.22)+rng.uniform(0,6.28))).astype('float32') return x,y xtr,ytr=make(n_train); xte,yte=make(n_test) return {'xtr':torch.tensor(xtr), 'ytr':torch.tensor(ytr), 'xte':torch.tensor(xte), 'yte':torch.tensor(yte), 'task':'classification','metric':'err','input_shape':(1,64),'out_dim':2, 'track':'long_1d_convolution','meta':META,'custom':True} def implicit_dft(x, M, m): # Positive-sign DFT, with zero injection only in bounded tiles (no length-M input). L=x.shape[-1]; p=(L+m-1)//m; q=M//m tile=F.pad(x, (0,p*m-L)).reshape(*x.shape[:-1],p,m).to(torch.float32).to(torch.complex64) r=torch.arange(q,device=x.device)[:,None]; t=torch.arange(p,device=x.device)[None,:] s=torch.arange(m,device=x.device)[None,:] zq=torch.exp(2j*torch.pi*(r*t)/q); zqm=torch.exp(2j*torch.pi*(r*s)/(q*m)) U=torch.einsum('rt,...ts->...rs',zq,tile)*zqm ell=torch.arange(m,device=x.device)[:,None] zm=torch.exp(2j*torch.pi*(ell*s)/m) out=torch.einsum('ls,...rs->...rl',zm,U) return out.transpose(-2,-1).reshape(*x.shape[:-1],M) def fft_conv(x, w, implicit=False, m=8): # Cross-correlation, matching Conv1d(no padding): convolution with reversed kernel. L=x.shape[-1]; K=w.shape[-1]; M=1 while M < L+K-1: M*=2 wr=w.flip(-1) if implicit: X=implicit_dft(x,M,m); W=implicit_dft(wr,M,m) else: X=torch.fft.fft(F.pad(x,(0,M-L)).to(torch.complex64)) W=torch.fft.fft(F.pad(wr,(0,M-K)).to(torch.complex64)) full=torch.fft.ifft(X.unsqueeze(1)*W.unsqueeze(0), n=M).real return full[...,K-1:K-1+L-K+1] class FFTConv1d(nn.Module): def __init__(self, cin, cout, k, implicit=False, m=8): super().__init__(); self.weight=nn.Parameter(torch.empty(cout,cin,k)); self.bias=nn.Parameter(torch.zeros(cout)) nn.init.kaiming_uniform_(self.weight,a=np.sqrt(5)); self.implicit=implicit; self.m=m def forward(self,x): # x B,C,L; einsum gives B,O,Lout and sums input channels. ys=[] for o in range(self.weight.shape[0]): yo=0 for c in range(self.weight.shape[1]): yo=yo+fft_conv(x[:,c,:],self.weight[o:o+1,c,:],self.implicit,self.m) ys.append(yo) return torch.cat(ys,dim=1)+self.bias[None,:,None] class LongCNN(nn.Module): def __init__(self, implicit=False, m=8): super().__init__(); self.conv=FFTConv1d(1,8,15,implicit,m); self.fc=nn.Linear(8,2) def forward(self,x): return self.fc(F.relu(self.conv(x)).mean(-1)) def run_one(seed, lr, implicit, m=8, epochs=12, return_model=False): torch.manual_seed(seed); np.random.seed(seed) d=get_dataset(seed) net=LongCNN(implicit,m) net,metric,_=train_model(net,d,epochs=epochs,lr=lr,batch=128) return (metric,net,d) if return_model else metric def main(): # Shared union: every idea learning rate is also evaluated by baseline. lrs=[0.001,0.003,0.01] grid=[{'lr':v,'m':8} for v in lrs] base=sweep_baseline(lambda cfg: (lambda seed: run_one(seed,cfg['lr'],False,8)), grid) idea=evaluate(lambda seed: run_one(seed,base['best_cfg']['lr'],True,8)) # Two nearby settings, same shared LR grid; select best idea configuration on 4 sweep seeds. idea_sweep=[] for cfg in grid: r=evaluate(lambda seed,cfg=cfg: run_one(seed,cfg['lr'],True,cfg['m']), seeds=(0,1,2,3)) idea_sweep.append({'cfg':cfg,'mean':r['mean']}) best=min(idea_sweep,key=lambda z:z['mean'])['cfg'] idea=evaluate(lambda seed: run_one(seed,best['lr'],True,best['m'])) # Trained-model behavioral signature: compare each operator's prediction to direct torch convolution. _,ib,d=run_one(0,best['lr'],True,best['m'],return_model=True) _,bb,_=run_one(0,base['best_cfg']['lr'],False,8,return_model=True) ib.eval(); bb.eval(); x=d['xte'][:16].to(next(ib.parameters()).device) with torch.no_grad(): wi=ib.conv.weight; xb=x direct=[] for o in range(8): yo=0 for c in range(wi.shape[1]): yo=yo+fft_conv(xb[:,c,:],wi[o:o+1,c,:],False,8) direct.append(yo) direct=torch.cat(direct,1)+ib.conv.bias[None,:,None]; observed=float((ib.conv(xb)-direct).norm()/direct.norm()) # same trained-system output comparison, not an analytic-only check system_gap=float((ib(xb)-bb(xb)).norm()/bb(xb).norm()) sig={'prediction':'implicit convolution equals explicit convolution; tile storage ratio M/(ceil(L/m)*m)', 'predicted_relative_operator_error':0.0,'observed_relative_operator_error':observed, 'predicted_storage_ratio':128/( ((64+8-1)//8)*8 ), 'trained_system_output_gap':system_gap, 'confirmed': bool(observed < 1e-5)} rep=make_report('long_1d_convolution','cnn_small',base,idea,{'mechanism_signature':sig,'custom_track':{'name':'long_1d_convolution','file':'stage2_bench.py','domain':'cnn'}}) rep['idea_sweep']=idea_sweep; rep['selection_note']='Baseline and idea share lr union [0.001, 0.003, 0.01]; idea m fixed a priori at 8.' with open('bench_report.json','w') as f: json.dump(rep,f,indent=2) print(json.dumps(rep,indent=2)) if __name__=='__main__': main()