"""Knob-curve + misread-injection experiments for the risk engine. Experiment 1 — knob calibration: sample a seeded scenario suite, compute each scenario's equity ONCE (equity is r-independent), then run the policy at r = 0..1 and measure: VPIP, PFR, aggression frequency, mean bet-to-pot ratio, mean heuristic EV, and mean decision loss vs the EV-oracle. Monotonicity of aggression metrics and a clean EV peak are the calibration test. Experiment 2 — misread injection: corrupt card reads (swap rank within suit or suit within rank) with probability p, recompute decisions at r=0.5, and measure action-flip rate and EV loss vs the clean read. Ties perception noise to money. """ import argparse import json import random from pathlib import Path from decision_engine import RiskEngine, TableState RANKS = ["A", "K", "Q", "J", "T", "9", "8", "7", "6", "5", "4", "3", "2"] SUITS = ["S", "H", "D", "C"] POSITIONS = ["BTN", "SB", "BB", "UTG", "MP", "MP2", "CO"] def make_scenario(rng, idx): street = rng.choice(["preflop", "flop", "turn", "river"]) n_board = {"preflop": 0, "flop": 3, "turn": 4, "river": 5}[street] deck = [(r, s) for r in RANKS for s in SUITS] rng.shuffle(deck) cards = [r + s for r, s in deck] hero, board = cards[:2], cards[2:2 + n_board] n_players = rng.choice([2, 3, 4, 5, 6]) pot = rng.choice([40, 80, 160, 320, 640, 1280]) to_call = rng.choice([0, 0, 20, 40, 80, 160, 320]) to_call = min(to_call, pot) return TableState(hero=hero, board=board, pot=pot, stack=rng.choice([400, 1000, 3000]), to_call=to_call, n_players=n_players, position=rng.choice(POSITIONS), street=street), idx def main(): ap = argparse.ArgumentParser() ap.add_argument("--n-scenarios", type=int, default=120) ap.add_argument("--sims", type=int, default=1500) ap.add_argument("--seed", type=int, default=11) ap.add_argument("--out", default="knob_results.json") args = ap.parse_args() eng = RiskEngine(n_sims=args.sims, seed=args.seed) rng = random.Random(args.seed) scen = [make_scenario(rng, i) for i in range(args.n_scenarios)] # equity once per scenario (r-independent) eqs = [] for st, i in scen: e = eng.equity(st.hero, st.board, max(1, st.n_players - 1)) eqs.append(e) if i % 20 == 0: print(f"equity {i}/{len(scen)}", flush=True) # --- experiment 1: knob curves --- curves = [] for r in [0.0, 0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0]: vpip = pfr = n_pre = aggr = calls_post = n_post = 0 ev_sum = loss_sum = bet_pot = n_bet = 0 for (st, _), eq in zip(scen, eqs): d = eng.decide(st, r, eq=eq) a = d["action"] ev_sum += eng.ev_of(st, d, eq) loss_sum += max(0.0, eng.best_ev(st, eq) - eng.ev_of(st, d, eq)) if st.street == "preflop": n_pre += 1 if a in ("call", "raise", "bet"): vpip += 1 if a in ("raise", "bet"): pfr += 1 else: n_post += 1 if a in ("bet", "raise"): aggr += 1 elif a == "call": calls_post += 1 if a in ("bet", "raise") and st.pot > 0: bet_pot += d["amount"] / st.pot n_bet += 1 row = {"r": r, "vpip": round(vpip / max(n_pre, 1), 3), "pfr": round(pfr / max(n_pre, 1), 3), "af": round(aggr / max(calls_post, 1), 3), "aggr_rate": round(aggr / max(n_post, 1), 3), "mean_bet_over_pot": round(bet_pot / max(n_bet, 1), 3), "mean_ev": round(ev_sum / len(scen), 2), "mean_decision_loss": round(loss_sum / len(scen), 2)} curves.append(row) print(row, flush=True) # monotonicity check on aggression metrics mono = {} for key in ("vpip", "pfr", "aggr_rate", "mean_bet_over_pot"): vals = [c[key] for c in curves] mono[key] = all(b >= a - 1e-9 for a, b in zip(vals, vals[1:])) ev_peak_r = max(curves, key=lambda c: c["mean_ev"])["r"] # --- experiment 2: misread injection --- misread = [] for p in [0.0, 0.05, 0.1, 0.2, 0.3]: flips = ev_loss = n = 0 for (st, _), eq in zip(scen, eqs): d_clean = eng.decide(st, 0.5, eq=eq) st2 = corrupt_state(st, rng, p) eq2 = eng.equity(st2.hero, st2.board, max(1, st2.n_players - 1)) d_bad = eng.decide(st2, 0.5, eq=eq2) n += 1 if d_clean["action"] != d_bad["action"]: flips += 1 ev_loss += max(0.0, eng.ev_of(st, d_clean, eq) - eng.ev_of(st, d_bad, eq)) row = {"p": p, "action_flip_rate": round(flips / n, 3), "mean_ev_loss": round(ev_loss / n, 2)} misread.append(row) print(row, flush=True) out = {"config": {"n_scenarios": args.n_scenarios, "sims": args.sims, "seed": args.seed}, "knob_curves": curves, "monotonic": mono, "ev_peak_r": ev_peak_r, "misread_injection": misread} Path(args.out).write_text(json.dumps(out, indent=1)) print(json.dumps({"monotonic": mono, "ev_peak_r": ev_peak_r}, indent=2), flush=True) def corrupt_state(st, rng, p): used = set(st.hero + st.board) def corrupt(card): if rng.random() >= p: return card rank, suit = card[:-1], card[-1] # corrupted card must stay unique vs all other known cards (treys # raises KeyError on duplicate primes when cards repeat) for _ in range(40): if rng.random() < 0.5: # swap rank, keep suit cand = rng.choice([r for r in RANKS if r != rank]) + suit else: # swap suit, keep rank cand = rank + rng.choice([s for s in SUITS if s != suit]) if cand not in used: used.discard(card) used.add(cand) return cand return card # no unique replacement available: keep clean read return TableState(hero=[corrupt(c) for c in st.hero], board=[corrupt(c) for c in st.board], pot=st.pot, stack=st.stack, to_call=st.to_call, n_players=st.n_players, position=st.position, street=st.street) if __name__ == "__main__": main()