import sys, json, 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 train_model, evaluate, sweep_baseline, make_report META = {'name': 'gecc_loop_graph', 'domain': 'graph-nn', 'description': 'Classify stochastic graphs with triangle-rich and inconsistent local overlaps using node features.'} N = 18 D = 5 def _adj(seed): rng = np.random.RandomState(seed + 173) y = np.arange(N) % 2 A = np.zeros((N, N), dtype=np.float32) # Two regimes: same-community triangles and cross-community noisy closures. for i in range(N): for j in range(i + 1, N): p = 0.42 if y[i] == y[j] else 0.13 if rng.rand() < p: A[i, j] = A[j, i] = 1 for c in range(0, N, 3): ids = [c, c + 1, c + 2] for i in ids: for j in ids: if i < j: A[i, j] = A[j, i] = 1 np.fill_diagonal(A, 0) return A def _stats(A): sets = [set(np.flatnonzero(A[i]).tolist()) | {i} for i in range(N)] common, C, r = {}, {}, {} for u in range(N): for v in np.flatnonzero(A[u]).astype(int): I = sorted(sets[u] & sets[v]) common[u, v] = I C[u, v] = 2.0 * len(I) / (len(sets[u]) + len(sets[v]) + 1e-12) r[u, v] = len(I) / (len(sets[u]) * len(sets[v]) + 1e-12) return common, C, r def get_dataset(seed, n_train, n_test): # Every seed has one deterministic graph; samples differ in node evidence. A = _adj(seed) rng = np.random.RandomState(seed + 991) def make(n): labels = rng.randint(0, 2, size=n).astype(np.int64) root = (2 * labels - 1).astype(np.float32) x = rng.normal(0, 0.95, (n, N, D)).astype(np.float32) # Evidence is locally coherent on one class and deliberately noisy on the other. x[:, :, 0] += root[:, None] * 0.28 x[:, ::2, 1] += root[:, None] * 0.23 x[:, 1::2, 2] += root[:, None] * 0.10 return x, labels xtr, ytr = make(n_train); xte, yte = make(n_test) common, C, r = _stats(A) return {'xtr': xtr, 'ytr': ytr, 'xte': xte, 'yte': yte, 'task': 'classification', 'metric': 'error', 'input_shape': (N, D), 'out_dim': 2, 'adjacency': A, 'common': common, 'closure': C, 'density': r} class GraphLayer(nn.Module): def __init__(self, d, idea=False): super().__init__(); self.idea = idea self.selfp = nn.Linear(d, d); self.msg = nn.Linear(d, d) self.loop = nn.Linear(d, d) if idea else None self.gate = nn.Parameter(torch.tensor([-1.0, 2.0, 0.2])) if idea else None def forward(self, h, A, common, closure, density, force_gate=None): # Algebraically identical vectorization of the directed-edge sum. dev = h.device; n = A.shape[0] at = torch.as_tensor(A, dtype=h.dtype, device=dev) if not self.idea: return torch.relu(self.selfp(h) + torch.einsum('uv,bvd->bud', at, self.msg(h))) c = torch.as_tensor(closure, dtype=h.dtype, device=dev) r = torch.as_tensor(density, dtype=h.dtype, device=dev) if force_gate is None: alpha = torch.sigmoid(self.gate[0] + self.gate[1]*c + self.gate[2]*r) * at else: alpha = torch.full_like(c, float(force_gate)) * at # Q[u,w] counts gated generalized-edge occurrences of w in intersections. q = torch.zeros((n,n), dtype=h.dtype, device=dev) for u in range(n): for v in np.flatnonzero(A[u]).astype(int): if common[u,v]: q[u, common[u,v]] += alpha[u,v] / len(common[u,v]) ordinary = torch.einsum('uv,bvd->bud', at-alpha, self.msg(h)) overlap = torch.einsum('uw,bwd->bud', q, self.loop(h)) return torch.relu(self.selfp(h) + ordinary + overlap) class GraphNet(nn.Module): def __init__(self, d=D, hidden=24, idea=False): super().__init__(); self.idea = idea; self.inp = nn.Linear(d, hidden) self.layer = GraphLayer(hidden, idea); self.head = nn.Linear(hidden, 2) def forward(self, x): h = torch.relu(self.inp(x)) h = self.layer(h, self._A, self._common, self._closure, self._density) return self.head(h.mean(1)) def bind(self, ds): self._A = ds['adjacency']; self._common = ds['common'] self._closure = np.zeros((N,N), dtype=np.float32); self._density = np.zeros((N,N), dtype=np.float32) for (u,v), val in ds['closure'].items(): self._closure[u,v] = val for (u,v), val in ds['density'].items(): self._density[u,v] = val return self def train_one(seed, lr, idea): torch.manual_seed(seed); np.random.seed(seed); random.seed(seed) ds = get_dataset(seed, 400, 400) for k in ('xtr','ytr','xte','yte'): ds[k] = torch.from_numpy(ds[k]) net = GraphNet(idea=idea).bind(ds) net, metric, _ = train_model(net, ds, epochs=10, lr=lr, batch=64, weight_decay=1e-3, log=lambda *_: None) return float(metric) def main(): seeds = tuple(range(8)); lrs = [1e-3, 3e-3, 1e-2] grid = [{'lr': lr} for lr in lrs] base = sweep_baseline(lambda cfg: (lambda s: train_one(s, cfg['lr'], False)), grid, seeds=seeds) candidates = [] for cfg in grid: candidates.append((cfg, evaluate(lambda s, c=cfg: train_one(s, c['lr'], True), seeds=seeds))) best_cfg, idea = min(candidates, key=lambda z: z[1]['mean']) # Signature from trained GECC systems: observed gate-vs-closure behavior and prediction effect. ds = get_dataset(700, 128, 128) for k in ('xtr','ytr','xte','yte'): ds[k] = torch.from_numpy(ds[k]) net = GraphNet(idea=True).bind(ds) net, _, _ = train_model(net, ds, epochs=10, lr=best_cfg['lr'], batch=64, weight_decay=1e-3, log=lambda *_: None) net = net.cpu(); net.eval(); bins = [[], [], []] with torch.no_grad(): h = torch.relu(net.inp(ds['xte'])) for u in range(N): for v in np.flatnonzero(ds['adjacency'][u]).astype(int): c = ds['closure'][u,v]; b = 0 if c < .25 else (1 if c < .5 else 2) alpha = float(torch.sigmoid(net.layer.gate[0] + net.layer.gate[1]*c + net.layer.gate[2]*ds['density'][u,v])) ordinary = net.layer.msg(h[:,v]); I=ds['common'][u,v] q=h[:,I].mean(1) if I else torch.zeros_like(ordinary) corr=net.layer.loop(q) bins[b].append((alpha, float((corr-ordinary).abs().mean()))) observed = [{'mean_C_bin': [0.125,0.375,0.75][i], 'mean_alpha': float(np.mean([z[0] for z in b])) if b else 0., 'mean_correction_gap': float(np.mean([z[1] for z in b])) if b else 0.} for i,b in enumerate(bins)] monotone = observed[0]['mean_alpha'] < observed[1]['mean_alpha'] < observed[2]['mean_alpha'] extra = {'prediction': 'trained gate activates more on higher closure; correction is selectively weighted', 'observed_model_behavior': observed, 'predicted_alpha_order': 'low < medium < high', 'confirmed': bool(monotone), 'custom_track': {'name': META['name'], 'file': 'gecc_bench.py', 'domain': META['domain']}} report = make_report('custom:gecc_loop_graph', 'GraphNet', base, idea, extra) report['idea_grid'] = [{'cfg': c, 'mean': r['mean'], 'std': r['std'], 'per_seed': r['per_seed']} for c,r in candidates] Path('bench_report.json').write_text(json.dumps(report, indent=2)) print(json.dumps(report, indent=2)) if __name__ == '__main__': main()