Spaces:
Running
Running
File size: 5,143 Bytes
4b84a03 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 | #!/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())
|