"""MVP for MI-guided latent protection. Run with the configured Python interpreter. The toy checks are deliberately first and print predicted versus observed quantities before the learning test. """ import json, math, random from pathlib import Path import numpy as np SEED = 2392 def seed_all(seed=SEED): random.seed(seed); np.random.seed(seed) try: import torch torch.manual_seed(seed) if torch.cuda.is_available(): torch.cuda.manual_seed_all(seed) except Exception: pass def allocate_variance(scores, mean_var=0.25, delta=1e-8): scores = np.asarray(scores, dtype=np.float64) q = (scores + delta) / (scores.mean() + delta) raw = 1.0 / q var = mean_var * len(scores) * raw / raw.sum() return q, var def math_checks(): # Prediction 1: normalization preserves exactly K * average variance. rng = np.random.default_rng(SEED) budget_errors = [] for _ in range(200): s = np.exp(rng.normal(0, 2, 16)) _, v = allocate_variance(s, .37) budget_errors.append(abs(v.mean() - .37)) p1 = {"predicted_mean_variance": .37, "observed_mean_variance": float(np.mean([np.mean(allocate_variance(np.exp(rng.normal(0,2,16)), .37)[1]) for _ in range(200)])), "max_abs_sweep_error": float(max(budget_errors))} # Prediction 2: a high-score coordinate gets inverse score-ratio variance. ratios = np.array([1, 2, 4, 8, 16, 32], dtype=float) observed = [] for r in ratios: _, v = allocate_variance([1, r], .5) observed.append(v[1] / v[0]) predicted = 1.0 / ratios p2 = {"score_ratios": ratios.tolist(), "predicted_high_to_low_variance": predicted.tolist(), "observed_high_to_low_variance": observed, "max_abs_error": float(np.max(np.abs(np.asarray(observed)-predicted)))} # Prediction 3: if local task-noise sensitivity equals s, the quadratic # penalty ratio is K^2/(sum(s)*sum(1/s)); it decreases as heterogeneity grows. penalty_ratios=[] predicted_penalty=[] for r in ratios: s=np.array([1., r]) _, v=allocate_variance(s, .5) adaptive=float(np.sum(s*v)); uniform=float(np.sum(s*.5)) penalty_ratios.append(adaptive/uniform) predicted_penalty.append(4/(np.sum(s)*np.sum(1/s))) p3={"score_ratios":ratios.tolist(), "predicted_adaptive_over_uniform_penalty":predicted_penalty, "observed_adaptive_over_uniform_penalty":penalty_ratios, "max_abs_error":float(np.max(np.abs(np.asarray(penalty_ratios)-predicted_penalty)))} return {"budget_conservation":p1,"inverse_scaling":p2,"quadratic_penalty":p3} def learning_experiment(steps=700): try: import torch from torch import nn device=torch.device("cuda" if torch.cuda.is_available() else "cpu") torch.set_num_threads(4) seed_all(SEED) n_train,n_test=6000,2500; k=8; batch=128 # First two coordinates contain the target; remaining dimensions are nuisance. g=torch.Generator().manual_seed(SEED) xtr=torch.randn(n_train,k,generator=g); xte=torch.randn(n_test,k,generator=g) ytr=((xtr[:,0]+0.8*xtr[:,1]+0.35*torch.randn(n_train,generator=g))>0).float() yte=((xte[:,0]+0.8*xte[:,1]+0.35*torch.randn(n_test,generator=g))>0).float() def run(mode): seed_all(SEED+({"uniform":0,"mi":1,"random":2,"magnitude":3}[mode])) model=nn.Linear(k,1).to(device); opt=torch.optim.Adam(model.parameters(),lr=.01) score_ema=torch.ones(k,device=device); random_scores=torch.rand(k,device=device); mean_var=.45; beta=.90 perm=torch.arange(n_train) for step in range(steps): if step% (n_train//batch)==0: perm=torch.randperm(n_train) ix=perm[(step*batch)%n_train:((step+1)*batch)%n_train] z=xtr[ix].to(device); y=ytr[ix].to(device) if mode=="uniform": var=torch.full((k,),mean_var,device=device) else: # Sensitivity proxy: gradient of a detached supervised MI-style # critic loss with respect to z; scores do not train the model. zz=z.detach().requires_grad_(True) critic=nn.functional.binary_cross_entropy_with_logits(model(zz).squeeze(1),y) grad=torch.autograd.grad(critic,zz)[0].abs().mean(0).detach() if mode=="mi": s=beta*score_ema+(1-beta)*grad; score_ema=s elif mode=="random": s=random_scores else: s=z.abs().mean(0) # Stabilize noisy minibatch sensitivities: retain a 10% floor. s=torch.nan_to_num(s, nan=1.0, posinf=1.0, neginf=0.0) s=s + 0.10*s.mean() s=torch.nan_to_num(s, nan=1.0, posinf=1.0, neginf=1.0).clamp_min(1e-6) q=(s+1e-6)/(s.mean()+1e-6); raw=1/q var=mean_var*k*raw/raw.sum() noisy=z+torch.randn_like(z)*var.sqrt() loss=nn.functional.binary_cross_entropy_with_logits(model(noisy).squeeze(1),y) opt.zero_grad(); loss.backward(); torch.nn.utils.clip_grad_norm_(model.parameters(), 5.0); opt.step() z=xte.to(device) if mode=="uniform": var=torch.full((k,),mean_var,device=device) else: # Sensitivity needs autograd; only this small block is enabled. zz=z[:batch].detach().requires_grad_(True) yy=yte[:batch].to(device) cr=nn.functional.binary_cross_entropy_with_logits(model(zz).squeeze(1),yy) gr=torch.autograd.grad(cr,zz)[0].abs().mean(0).detach() if mode=="mi": s=score_ema elif mode=="random": s=random_scores else: s=z.abs().mean(0) s=torch.nan_to_num(s, nan=1.0, posinf=1.0, neginf=0.0) s=s + 0.10*s.mean() s=torch.nan_to_num(s, nan=1.0, posinf=1.0, neginf=1.0).clamp_min(1e-6) q=(s+1e-6)/(s.mean()+1e-6); var=mean_var*k*(1/q)/(1/q).sum() with torch.no_grad(): logits=model(z+torch.randn_like(z)*var.sqrt()).squeeze(1) acc=((logits>0)==(yte.to(device)>0.5)).float().mean().item() clean=((model(z).squeeze(1)>0)==(yte.to(device)>0.5)).float().mean().item() return acc,clean,float(var.mean().item()) result={m:run(m) for m in ["uniform","mi","random","magnitude"]} return {"device":str(device),"steps":steps,"results":result} except Exception as e: # Required safe fallback if CUDA/runtime setup fails. return {"error":repr(e),"fallback_note":"analytic checks completed; learning run unavailable"} def main(): seed_all() out={"math_checks":math_checks(),"learning":learning_experiment()} Path("results.json").write_text(json.dumps(out,indent=2)) print(json.dumps(out,indent=2)) if __name__ == '__main__': main()