Uniform Stochastic Barrier Critic / barrier_experiment.py

Failed on benchmark

Raw ⬇ ZIP
  1import json
  2import numpy as np
  3from scipy.optimize import linprog
  4
  5SEED=11
  6rng=np.random.default_rng(SEED)
  7mu,sigma=0.055,0.20
  8U_EDGE,T_EDGE=-0.80,0.80
  9x=np.linspace(-1.,1.,201); N=len(x)
 10U=x<=U_EDGE; T=x>=T_EDGE; C=~(U|T); X0=(x>=-.2)&(x<=.2)
 11z,qw=np.polynomial.hermite.hermgauss(61); z*=np.sqrt(2.); qw/=np.sqrt(np.pi)
 12uix,tix=np.flatnonzero(U)[0],np.flatnonzero(T)[-1]
 13P=np.zeros((N,N))
 14for i,xi in enumerate(x):
 15    if U[i]: P[i,uix]=1.; continue
 16    if T[i]: P[i,tix]=1.; continue
 17    for y,wt in zip(xi+mu+sigma*z,qw):
 18        if y<=U_EDGE: P[i,uix]+=wt
 19        elif y>=T_EDGE: P[i,tix]+=wt
 20        else:
 21            j=np.clip(np.searchsorted(x,y)-1,0,N-2); a=(y-x[j])/(x[j+1]-x[j])
 22            P[i,j]+=wt*(1-a); P[i,j+1]+=wt*a
 23
 24def oracle(delta):
 25    tv=N; c=np.zeros(N+1); c[tv]=1.; A=[]; b=[]
 26    for i in np.flatnonzero(C):
 27        r=np.zeros(N+1); r[:N]=P[i]; r[i]-=1.; A.append(r); b.append(-delta)
 28    for i in np.flatnonzero(X0):
 29        r=np.zeros(N+1); r[i]=1.; r[tv]=-1.; A.append(r); b.append(0.)
 30    bounds=[(1,1) if U[i] else ((0,0) if T[i] else (0,1)) for i in range(N)]+[(0,1)]
 31    r=linprog(c,A_ub=np.asarray(A),b_ub=np.asarray(b),bounds=bounds,method='highs')
 32    return (None,None) if not r.success else (float(r.x[tv]),r.x[:N])
 33
 34def exact_failure():
 35    ci=np.flatnonzero(C); h=np.zeros(N); h[U]=1
 36    h[ci]=np.linalg.solve(np.eye(len(ci))-P[np.ix_(ci,ci)],P[np.ix_(ci,[uix])].ravel())
 37    return h
 38
 39def margin_sweep():
 40    out=[]
 41    for d in [0,.002,.005,.01,.02,.03,.04,.05]:
 42        t,B=oracle(d); out.append({'delta':d,'min_initial_B':t,'p_cert':None if t is None else 1-t,'feasible':t is not None})
 43    return out
 44
 45def mc_scaling():
 46    def bf(y): return np.clip((y+1)/2,0,1)
 47    xx=.1; truth=float(np.sum(qw*bf(xx+mu+sigma*z))); out=[]
 48    for M in [8,16,32,64,128,256,512,1024]:
 49        ee=[]
 50        for _ in range(400): ee.append(abs(bf(xx+mu+sigma*rng.normal(size=M)).mean()-truth))
 51        out.append({'M':M,'mae':float(np.mean(ee))})
 52    slope=float(np.polyfit(np.log([a['M'] for a in out]),np.log([a['mae'] for a in out]),1)[0])
 53    return truth,out,slope
 54
 55def neural_comparison(h,delta=.02,steps=2500):
 56    try:
 57        import torch
 58        torch.manual_seed(SEED); torch.set_num_threads(4)
 59        dev='cuda' if torch.cuda.is_available() else 'cpu'
 60        # Small scalar MLP; P is a deterministic quadrature estimate of the stochastic expectation.
 61        xx=torch.tensor(x[:,None],dtype=torch.float32,device=dev); PP=torch.tensor(P,dtype=torch.float32,device=dev)
 62        UU=torch.tensor(U,device=dev); TT=torch.tensor(T,device=dev); CC=torch.tensor(C,device=dev); XX=torch.tensor(X0,device=dev)
 63        def run(kind):
 64            net=torch.nn.Sequential(torch.nn.Linear(1,32),torch.nn.Tanh(),torch.nn.Linear(32,32),torch.nn.Tanh(),torch.nn.Linear(32,1),torch.nn.Sigmoid()).to(dev)
 65            opt=torch.optim.Adam(net.parameters(),lr=3e-3)
 66            target=torch.tensor(h[:,None],dtype=torch.float32,device=dev)
 67            for k in range(steps):
 68                B=net(xx).squeeze(1); PB=PP@B
 69                boundary=((B[UU]-1)**2).mean()+(B[TT]**2).mean()
 70                if kind=='baseline': loss=boundary+2*((B-target)**2).mean()
 71                else:
 72                    sp=torch.nn.functional.softplus
 73                    loss=8*boundary+8*sp(B[XX]-(1-delta)).mean()+8*sp(PB[CC]-B[CC]+delta).mean()
 74                opt.zero_grad(); loss.backward(); opt.step()
 75            with torch.no_grad():
 76                B=net(xx).squeeze(1); PB=PP@B
 77                return {'loss':float(loss),'max_init_B':float(B[XX].max()),'max_init_limit_violation':float((B[XX]-(1-delta)).max()),'max_drift_residual':float((PB[CC]-B[CC]+delta).max()),'boundary_rmse':float(torch.sqrt(((B[UU]-1)**2).mean()+(B[TT]**2).mean()))}
 78        return {'device':dev,'baseline':run('baseline'),'idea':run('idea')}
 79    except Exception as e: return {'error':repr(e),'fallback':'neural comparison unavailable'}
 80
 81def rollout(n=30000):
 82    ps=[]
 83    for s in np.linspace(-.2,.2,9):
 84        ok=0
 85        for _ in range(n//9):
 86            y=s
 87            for _ in range(300):
 88                if y<=U_EDGE: break
 89                if y>=T_EDGE: ok+=1; break
 90                y+=mu+sigma*rng.normal()
 91        ps.append(ok/(n//9))
 92    return ps
 93
 94def main():
 95    h=exact_failure(); sweep=margin_sweep(); truth,mc,slope=mc_scaling(); nn=neural_comparison(h)
 96    d1,d2=sweep[1],sweep[2]
 97    slope_delta=(d2['min_initial_B']-d1['min_initial_B'])/(d2['delta']-d1['delta'])
 98    feas=[r for r in sweep if r['feasible']]; _,B=oracle(feas[-1]['delta'])
 99    result={'seed':SEED,'system':{'mu':mu,'sigma':sigma,'U':U_EDGE,'T':T_EDGE},'predictions':{'margin_slope_observed':slope_delta,'margin_prediction':'positive approximately linear until infeasible','mc_slope_observed':slope,'mc_prediction':-.5,'feasibility_cliff_between_delta': [.03,.04]},'margin_sweep':sweep,'mc_true_expectation':truth,'mc_scaling':mc,'exact_failure_max_X0':float(h[X0].max()),'rollout_success_by_start':rollout(),'certificate_check':{'delta':feas[-1]['delta'],'max_B_X0':float(B[X0].max()),'max_exact_failure_X0':float(h[X0].max()),'conservative_gap':float(B[X0].max()-h[X0].max())},'neural_comparison':nn}
100    with open('results.json','w') as f: json.dump(result,f,indent=2)
101    print(json.dumps(result,indent=2))
102if __name__=='__main__': main()