Spaces:
Running
Running
| #!/usr/bin/env python3 | |
| """Anytime Detection of Strategic Deviations in Multi-Agent Systems (arXiv:2601.05427). Reproduces: | |
| [2] Theorem 2.7: the expected detection time of an eta-deviation is O(log(sum|A_i|/alpha)/eta^2), with | |
| a sequential e-process test that controls the false-alarm rate at alpha (Ville's inequality). | |
| [4] Theorem 3.4: (stochastic-game analogue) detection time scales inversely with the deviation's KL | |
| signal -- reproduced here as the ~1/eta^2 scaling of the mean detection time. | |
| Test: for each candidate deviating action a', an e-process M_t(a') = int_0^1 prod_s (1 + lam X_s(a')) dnu(lam), | |
| X_s(a') = u_i(a', a_{-i,s}) - u_i(a_{i,s}, a_{-i,s}) (gain from switching). Reject when max_a' M_t > b = |A|/alpha. | |
| Deterministic seeds; 2x2 game. | |
| """ | |
| import numpy as np, json, hashlib | |
| U1 = np.array([[1.0, 0.0], [0.0, 1.0]]) # player-1 payoff (wants to match opponent) | |
| LAM = np.linspace(0.0, 0.9, 30) # mixture-method betting grid (prior nu = uniform) | |
| def eprocess_detect(eta, alpha, rng, Tmax=200000): | |
| """Simulate play with opponent mixing q=0.5+eta (eta=0 => equilibrium/null); return detection time | |
| (or Tmax if none) and whether a false alarm occurred under the null.""" | |
| b = 2.0 / alpha # threshold: |A_1| actions / alpha | |
| logM = np.zeros((2, len(LAM))) # log of the per-lambda betting wealth, per action a' | |
| q = 0.5 + eta | |
| for t in range(1, Tmax + 1): | |
| a1 = 0 if rng.random() < 0.5 else 1 # player 1 plays equilibrium (0.5/0.5) | |
| a2 = 0 if rng.random() < q else 1 # opponent play (deviates if eta>0) | |
| for ap in (0, 1): | |
| X = U1[ap, a2] - U1[a1, a2] # gain from switching to a' | |
| logM[ap] += np.log(np.maximum(1.0 + LAM * X, 1e-12)) | |
| # e-value = mixture over lambda (mean of betting wealths); detect if any action's e-value > b | |
| from scipy.special import logsumexp | |
| ev = np.array([logsumexp(logM[ap]) - np.log(len(LAM)) for ap in (0, 1)]) | |
| if np.max(ev) > np.log(b): | |
| return t | |
| return None | |
| def main(): | |
| R = {"claim": "Strategic_deviation_detection_time", "paper": "arXiv:2601.05427"} | |
| alpha = 0.05 | |
| # ---------- [false-alarm control] under the NULL (eta=0), reject rate <= alpha ---------- | |
| false_alarms = 0; NT = 300 | |
| for s in range(NT): | |
| if eprocess_detect(0.0, alpha, np.random.default_rng(7000 + s), Tmax=5000) is not None: | |
| false_alarms += 1 | |
| R["null_false_alarm_rate"] = round(false_alarms / NT, 3) | |
| R["thm2.6_false_alarm_le_alpha"] = (false_alarms / NT) <= alpha + 0.02 | |
| # ---------- [2]/[4] detection time scales as ~ log(1/alpha)/eta^2 ---------- | |
| rows = [] | |
| for eta in (0.15, 0.2, 0.3, 0.4): | |
| times = [] | |
| for s in range(120): | |
| t = eprocess_detect(eta, alpha, np.random.default_rng(100 + s)) | |
| if t is not None: times.append(t) | |
| mt = float(np.mean(times)) if times else float("inf") | |
| rows.append({"eta": eta, "mean_detect_time": round(mt, 1), "detected_frac": round(len(times)/120, 2), | |
| "time_x_eta2": round(mt * eta ** 2, 2)}) | |
| R["detection_vs_eta"] = rows | |
| # detection time * eta^2 should be ~ constant (=> time ~ 1/eta^2); and log-log slope ~ -2 | |
| etas = np.array([r["eta"] for r in rows]); mts = np.array([r["mean_detect_time"] for r in rows]) | |
| slope = float(np.polyfit(np.log(etas), np.log(mts), 1)[0]) | |
| R["detect_loglog_slope_vs_eta"] = round(slope, 2) | |
| R["thm2.7_time_scales_as_inv_eta2"] = -2.6 < slope < -1.4 | |
| prods = [r["time_x_eta2"] for r in rows] | |
| R["time_x_eta2_roughly_constant"] = (max(prods) / min(prods)) < 2.2 | |
| R["verdict"] = "supports" if (R["thm2.6_false_alarm_le_alpha"] and R["thm2.7_time_scales_as_inv_eta2"] | |
| and R["time_x_eta2_roughly_constant"]) else "inconclusive" | |
| print("claim: " + R["claim"]) | |
| print(f"[false alarm] null (eta=0) reject rate = {R['null_false_alarm_rate']} (<= alpha={alpha}): {R['thm2.6_false_alarm_le_alpha']}") | |
| print(f"[2]/[4] detection time vs deviation eta:") | |
| for r in rows: print(f" eta={r['eta']}: mean detect time={r['mean_detect_time']} (detected {r['detected_frac']}), time*eta^2={r['time_x_eta2']}") | |
| print(f" log-log slope(time vs eta) = {R['detect_loglog_slope_vs_eta']} (~ -2 => ~1/eta^2): {R['thm2.7_time_scales_as_inv_eta2']}; time*eta^2 ~ const: {R['time_x_eta2_roughly_constant']}") | |
| print(f"verdict: {R['verdict']}") | |
| def _np(o): | |
| if isinstance(o, np.bool_): return bool(o) | |
| if isinstance(o, np.integer): return int(o) | |
| if isinstance(o, np.floating): return float(o) | |
| raise TypeError | |
| import os; os.makedirs("outputs", exist_ok=True) | |
| open("outputs/detection_results.json", "w").write(json.dumps(R, indent=2, default=_np)) | |
| print("RESULTS_SHA256=" + hashlib.sha256(json.dumps(R, sort_keys=True, default=_np).encode()).hexdigest()) | |
| return 0 if R["verdict"] == "supports" else 1 | |
| if __name__ == "__main__": | |
| raise SystemExit(main()) | |