| """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)] |
|
|
| |
| 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) |
|
|
| |
| 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) |
|
|
| |
| 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"] |
|
|
| |
| 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] |
| |
| |
| for _ in range(40): |
| if rng.random() < 0.5: |
| cand = rng.choice([r for r in RANKS if r != rank]) + suit |
| else: |
| 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 |
|
|
| 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() |
|
|
