"""CPU protocol experiment. Binary linear model; no image/LLM benchmark.""" import json import platform from pathlib import Path import numpy as np OUT = Path(__file__).resolve().parent / 'results' SEEDS = (79, 80, 81) EPS = (0., .05, .1, .2, .3) L2, LR, STEPS = .01, .2, 1000 def sigmoid(x): return np.exp(-np.logaddexp(0., -x)) def loss_grad(x, y, w, b): margin = y * (x @ w + b) a = -y * sigmoid(-margin) loss = np.logaddexp(0., -margin).mean() + L2 * (w @ w) / 2 return loss, x.T @ a / len(y) + L2 * w, a.mean() def input_grad(x, y, w, b): return (-y * sigmoid(-y * (x @ w + b)))[:, None] * w def generate(rng, n): # These are abstract sensor features, not bounded pixels. y = rng.choice([-1., 1.], n) x = rng.normal(0., .3, (n, 20)) + .15 * y[:, None] x[:, 0] = y + rng.normal(0., .6, n) return x, y def select_worse(x, candidate, y, w, b): mask = y * (candidate @ w + b) < y * (x @ w + b) return np.where(mask[:, None], candidate, x) def pgd(x, y, w, b, eps, alpha, steps, restarts, rng, random_start): # Retain best per sample across all iterates, restarts AND the clean input. best = x.copy() for _ in range(restarts): z = x + rng.uniform(-eps, eps, x.shape) if random_start else x.copy() best = select_worse(best, z, y, w, b) for _ in range(steps): z = z + alpha * np.sign(input_grad(z, y, w, b)) z = np.clip(z, x - eps, x + eps) best = select_worse(best, z, y, w, b) return best def metrics(x, y, w, b, candidate, exact_margin): clean = y * (x @ w + b) > 0 margin = y * (candidate @ w + b) assert np.min(np.abs(margin)) > 1e-10 # no ambiguous ties in this dataset survived = clean & (margin > 0) exact_ok = exact_margin > 0 assert np.all(~exact_ok | survived) return dict(clean_accuracy=float(clean.mean()), robust_accuracy=float(survived.mean()), conditional_asr=float((clean & ~survived).sum() / clean.sum()), clean_correct=int(clean.sum()), failures=int((~survived).sum()), missed_failures=int((survived & ~exact_ok).sum()), max_margin_gap=float(np.max(margin - exact_margin))) def gradient_check(x, y, w, b): _, gw, gb = loss_grad(x, y, w, b) h, errors = 1e-5, [] for j in (0, 1, 7, 19): wp, wm = w.copy(), w.copy() wp[j] += h wm[j] -= h errors.append(abs((loss_grad(x, y, wp, b)[0] - loss_grad(x, y, wm, b)[0]) / (2*h) - gw[j])) errors.append(abs((loss_grad(x, y, w, b+h)[0] - loss_grad(x, y, w, b-h)[0]) / (2*h) - gb)) gx = input_grad(x[:1], y[:1], w, b)[0] for j in (0, 1, 7, 19): xp, xm = x[:1].copy(), x[:1].copy() xp[0, j] += h xm[0, j] -= h lp = np.logaddexp(0., -y[0]*(xp[0] @ w + b)) lm = np.logaddexp(0., -y[0]*(xm[0] @ w + b)) errors.append(abs((lp-lm)/(2*h)-gx[j])) assert max(errors) < 1e-8 return max(errors) def main(): OUT.mkdir(exist_ok=True) runs, checks = [], [] for seed in SEEDS: rng = np.random.default_rng(seed) train, yt = generate(rng, 1024) dev, yd = generate(rng, 256) test, y = generate(rng, 2048) assert len(np.unique(np.vstack([train, dev, test]), axis=0)) == 3328 w, b = rng.normal(0., .01, 20), 0. trace = [] check = gradient_check(train[:32], yt[:32], w, b) for step in range(STEPS): loss, gw, gb = loss_grad(train, yt, w, b) if step % 100 == 0: trace.append(dict(step=step, objective=float(loss))) w -= LR * gw b -= LR * gb final_loss, gw, gb = loss_grad(train, yt, w, b) trace.append(dict(step=STEPS, objective=float(final_loss))) clean_margin = y * (test @ w + b) assert np.min(np.abs(clean_margin)) > 1e-10 payload = dict(train=train, train_y=yt, dev=dev, dev_y=yd, test=test, test_y=y, w=w, b=np.array(b)) frozen = w.copy(), b row = dict(seed=seed, trace=trace, dev_accuracy=float((yd*(dev@w+b)>0).mean()), final_gradient_norm=float(np.sqrt(gw@gw+gb*gb)), budgets=[]) for k, eps in enumerate(EPS): attack_rng = np.random.default_rng(seed * 1000 + k) exact_margin = clean_margin - eps*np.abs(w).sum() assert np.min(np.abs(exact_margin)) > 1e-10 random_best = test.copy() for _ in range(16): z = test + attack_rng.uniform(-eps, eps, test.shape) random_best = select_worse(random_best, z, y, w, b) candidates = dict( random16=random_best, pgd2=pgd(test, y, w, b, eps, eps/20, 2, 1, attack_rng, False), fgsm=test + eps*np.sign(input_grad(test, y, w, b)), pgd20=pgd(test, y, w, b, eps, eps/5, 20, 3, attack_rng, True), zero_gradient_fixture=test.copy()) budget_row = dict(epsilon=eps, exact_robust_accuracy=float((exact_margin>0).mean()), attacks={}) payload[f'e{k}_exact_margin'] = exact_margin union_survived = clean_margin > 0 for name, z in candidates.items(): assert np.max(np.abs(z-test)) <= eps + 1e-12 assert np.isfinite(z).all() payload[f'e{k}_{name}'] = z budget_row['attacks'][name] = metrics(test, y, w, b, z, exact_margin) previous = union_survived.copy() union_survived &= y * (z @ w + b) > 0 assert np.all(~union_survived | previous) if name in ('fgsm', 'pgd20'): assert np.allclose(y*(z@w+b), exact_margin, atol=1e-10, rtol=0) budget_row['union_robust_accuracy'] = float(union_survived.mean()) row['budgets'].append(budget_row) assert np.array_equal(w, frozen[0]) and b == frozen[1] exact_curve = [v['exact_robust_accuracy'] for v in row['budgets']] assert all(a >= b for a, b in zip(exact_curve, exact_curve[1:])) np.savez_compressed(OUT / f'seed{seed}.npz', **payload) checks.append(dict(seed=seed, gradient_max_error=check)) runs.append(row) print('seed', seed, 'train', train.shape, 'test', test.shape, 'logits', (len(y),), 'gradient_error', check, 'exact_curve', exact_curve) aggregate = [] for k, eps in enumerate(EPS): ar = dict(epsilon=eps, attacks={}) for name in runs[0]['budgets'][k]['attacks']: ar['attacks'][name] = {} for metric in ('clean_accuracy', 'robust_accuracy', 'conditional_asr', 'missed_failures'): values = [r['budgets'][k]['attacks'][name][metric] for r in runs] ar['attacks'][name][metric] = dict(mean=float(np.mean(values)), sd=float(np.std(values, ddof=1))) aggregate.append(ar) summary = dict(environment=dict(python=platform.python_version(), numpy=np.__version__, device='CPU', dtype='float64'), configuration=dict(seeds=SEEDS, epsilon=EPS, l2=L2, lr=LR, steps=STEPS, domain='R^20; fixed latent label; no pixel clipping; no feature scaling', training='ordinary regularized logistic regression; final checkpoint; dev logging only'), gradient_checks=checks, runs=runs, aggregate=aggregate) (OUT/'summary.json').write_text(json.dumps(summary, indent=2)+'\n') print(json.dumps(aggregate[3], indent=2)) if __name__ == '__main__': main()