1import json, math, random
  2import numpy as np
  3
  4SEED = 996
  5np.random.seed(SEED); random.seed(SEED)
  6
  7def sat(u): return np.clip(u, -1.0, 1.0)
  8
  9def xi_law(a, kappa=2.0, eps=0.05, xmin=1.0, xmax=8.0):
 10    return np.clip(kappa/(eps + np.asarray(a)), xmin, xmax)
 11
 12def midpoint_quantize(x, L):
 13    # Uniform interval quantizer on [-1,1], midpoint reconstruction.
 14    edges = np.linspace(-1, 1, int(L)+1)
 15    idx = np.clip(np.searchsorted(edges, x, side='right')-1, 0, L-1)
 16    return (edges[idx] + edges[idx+1])/2
 17
 18def math_checks():
 19    # Prediction 1: saturation begins exactly at |x|=1/xi.
 20    xis = [1.25, 2., 4., 7.]
 21    threshold_err = []
 22    for z in xis:
 23        x = np.linspace(0, 1, 200001)
 24        y = sat(z*x)
 25        first = x[np.argmax(y >= 1-1e-10)]
 26        threshold_err.append(abs(first - 1/z))
 27    # Prediction 2: unsaturated local slope is xi and decreases with influence.
 28    aa = np.linspace(0.05, 2.0, 20)
 29    zz = xi_law(aa)
 30    slopes = []
 31    for z in zz:
 32        x = np.linspace(-0.05, 0.05, 101)
 33        slopes.append(np.polyfit(x, sat(z*x), 1)[0])
 34    slope_err = float(np.max(np.abs(np.asarray(slopes)-zz)))
 35    monotone = bool(np.all(np.diff(zz) <= 1e-12))
 36    # Prediction 3: increasing strategic pressure reduces adaptive credible bins.
 37    # The proposed design maps xi to a variable number of bins, capped at binary.
 38    kappas = np.linspace(.5, 8., 16)
 39    a = .30
 40    bins = [max(2, int(np.rint(8/xi_law(a, kappa=k)))) for k in kappas]
 41    occupied = []
 42    x = np.linspace(-1, 1, 20001)
 43    for k, L in zip(kappas, bins):
 44        z = xi_law(a, kappa=k)
 45        occupied.append(len(np.unique(midpoint_quantize(sat(z*x), L))))
 46    return {
 47        'threshold_max_abs_error': float(max(threshold_err)),
 48        'threshold_observed': [float(1/z) for z in xis],
 49        'threshold_errors': [float(v) for v in threshold_err],
 50        'influence_xi_at_low_high_a': [float(zz[0]), float(zz[-1])],
 51        'influence_xi_monotone_decreasing': monotone,
 52        'slope_max_abs_error_vs_xi': slope_err,
 53        'kappas': [float(v) for v in kappas], 'adaptive_bins': bins,
 54        'occupied_bins': occupied,
 55        'binary_reached': bool(min(bins)==2 and occupied[-1] <= 2),
 56        'formula_bounds_ok': bool(np.all(np.abs(sat(zz*x[:len(zz)])) <= 1+1e-12))
 57    }
 58
 59def gnn_experiment():
 60    try:
 61        import torch
 62        torch.manual_seed(SEED)
 63        device = 'cuda' if torch.cuda.is_available() else 'cpu'
 64        try:
 65            if device == 'cuda': torch.cuda.empty_cache()
 66        except Exception: device='cpu'
 67        n=96; d=12; c=2
 68        # Homophilic SBM, undirected with self loops, row normalized.
 69        y=torch.arange(n, device=device)%2
 70        rng=np.random.default_rng(SEED)
 71        A=np.zeros((n,n),dtype=np.float32)
 72        for i in range(n):
 73            for j in range(i+1,n):
 74                p=.24 if int(y[i])==int(y[j]) else .035
 75                if rng.random()<p: A[i,j]=A[j,i]=1
 76        A += np.eye(n,dtype=np.float32)
 77        A=A/A.sum(1,keepdims=True)
 78        At=torch.tensor(A,device=device)
 79        X=torch.randn(n,d,device=device)
 80        # Class signal in first coordinate, intentionally modest.
 81        X[:,0] += (2*y.float()-1)*.75
 82        perm=torch.randperm(n,device=device); tr=perm[:60]; va=perm[60:]
 83        class Base(torch.nn.Module):
 84            def __init__(self):
 85                super().__init__(); self.w1=torch.nn.Linear(d,16); self.w2=torch.nn.Linear(16,c)
 86            def forward(self):
 87                h=torch.relu(At @ self.w1(X)); return self.w2(At @ h)
 88        # Scalar strategic channel replaces the scalar message, then broadcasts through Wm.
 89        class Strategic(torch.nn.Module):
 90            def __init__(self, quant=False):
 91                super().__init__(); self.w0=torch.nn.Linear(d,16); self.vm=torch.nn.Parameter(torch.randn(d)*.15)
 92                self.wm=torch.nn.Linear(1,16,bias=False); self.out=torch.nn.Linear(16,c); self.quant=quant
 93            def forward(self, eval_quant=False):
 94                h=torch.relu(self.w0(X)); raw=torch.tanh(X @ self.vm)
 95                # audience statistic: normalized incoming influence (column mass)
 96                influence=At.sum(0)
 97                xi=torch.clamp(2.0/(.05+influence),1.,8.)
 98                msg=torch.clamp(xi*raw,-1.,1.)
 99                if self.quant and eval_quant:
100                    # adaptive resolution; low influence -> larger xi -> fewer bins
101                    L=torch.clamp(torch.round(8./xi),min=2,max=8).long()
102                    vals=[]
103                    for i in range(n): vals.append(midpoint_quantize(msg[i].detach().cpu().numpy(),int(L[i])))
104                    msg=torch.tensor(np.asarray(vals),device=device,dtype=X.dtype)
105                h=torch.relu(h + At @ self.wm(msg[:,None]))
106                return self.out(h), raw, msg, xi
107        def train_model(model, strategic=False, quant=False):
108            opt=torch.optim.Adam(model.parameters(),lr=.025,weight_decay=1e-3)
109            for step in range(180):
110                opt.zero_grad()
111                out,*_=model() if strategic else (model(),)
112                loss=torch.nn.functional.cross_entropy(out[tr],y[tr]); loss.backward(); opt.step()
113            with torch.no_grad():
114                if strategic: out,raw,msg,xi=model(eval_quant=quant)
115                else: out=model(); raw=msg=xi=None
116                acc=(out[va].argmax(1)==y[va]).float().mean().item()
117            return acc, raw, msg, xi
118        b=Base().to(device); bacc,*_=train_model(b)
119        s=Strategic(False).to(device); sacc,sraw,smsg,sxi=train_model(s,True,False)
120        q=Strategic(True).to(device); qacc,qraw,qmsg,qxi=train_model(q,True,True)
121        with torch.no_grad():
122            satur=np.abs((sxi*sraw).cpu().numpy())>=1-1e-5
123            bits=np.log2(np.maximum(2,np.rint(8/sxi.cpu().numpy())))
124            ent=None
125            # communication summary uses adaptive assigned bits, not tensor storage.
126        return {'device':device,'nodes':n,'train_steps':180,
127          'validation_accuracy':{'baseline_gcn':bacc,'strategic_clipped':sacc,'strategic_adaptive_quantized':qacc},
128          'strategic_saturation_fraction':float(satur.mean()),
129          'mean_adaptive_bits':float(bits.mean()),'min_max_xi':[float(sxi.min()),float(sxi.max())],
130          'bounded_message_max_abs':float(torch.max(torch.abs(smsg)).item()),
131          'mean_influence':float(sxi.numel() and (At.sum(0).mean()).item())}
132    except Exception as e:
133        return {'error':repr(e),'fallback':'math checks still valid'}
134
135def main():
136    result={'seed':SEED,'math_checks':math_checks(),'gnn_experiment':gnn_experiment()}
137    print(json.dumps(result,indent=2,sort_keys=True))
138if __name__=='__main__': main()