#!/usr/bin/env python3 """Generate the PLO5/PLO6 Monte Carlo accuracy study. The exact oracle enumerates every legal five-card board for two known preflop hands. Monte Carlo replications then sample the oracle's win/tie/loss distribution. Sampling that categorical distribution is statistically equivalent to sampling a legal board uniformly and retaining only its equity outcome. A direct-board sampler is included as a reduction check. """ from __future__ import annotations import csv import gzip import hashlib import json import math import shutil from dataclasses import asdict, dataclass from importlib.metadata import version from itertools import combinations from pathlib import Path from typing import Iterable import numpy as np from phevaluator.evaluator import evaluate_5cards STUDY_ID = "gtogecko-plo-monte-carlo-accuracy-2026-08-29" REPLICATIONS = 2_000 TRIAL_COUNTS = (10_000, 25_000, 50_000, 100_000, 250_000, 500_000) DIRECT_BOARD_TRIALS = 25_000 Z_95 = 1.959963984540054 TABLE_SIZE = 52 * 52 * 52 NOT_A_RANK = 8_000 RANKS = "23456789TJQKA" SUITS = "cdhs" DECK = tuple(range(52)) SCRIPT_PATH = Path(__file__).resolve() BUNDLE = SCRIPT_PATH.parent PUBLISHED_BUNDLE = SCRIPT_PATH.name == "plo-monte-carlo-generate-study.py" REPO_CANDIDATE = next( (parent for parent in SCRIPT_PATH.parents if (parent / "package.json").is_file()), None, ) IN_REPOSITORY = REPO_CANDIDATE is not None ROOT = REPO_CANDIDATE if IN_REPOSITORY else Path.cwd() OUTPUT = BUNDLE / ("plo-monte-carlo-output" if PUBLISHED_BUNDLE else "outputs") PUBLIC_DATA = ROOT / "public" / "data" if IN_REPOSITORY else BUNDLE @dataclass(frozen=True) class Spot: key: str variant: str label: str hero: tuple[str, ...] villain: tuple[str, ...] SPOTS = ( Spot( "P5-A", "PLO5", "High pairs versus a connected rundown", ("As", "Ah", "Ks", "Kh", "Qc"), ("Jc", "Tc", "9d", "8d", "7h"), ), Spot( "P5-B", "PLO5", "Two Broadway rundowns; chop-heavy", ("As", "Ks", "Qh", "Jh", "Tc"), ("Ad", "Kd", "Qc", "Jc", "9s"), ), Spot( "P5-D", "PLO5", "High pairs versus a trip-deuce stress test", ("As", "Ah", "Ks", "Kh", "Qc"), ("2c", "2d", "2h", "3s", "4s"), ), Spot( "P6-A", "PLO6", "High pairs versus a six-card rundown", ("As", "Ah", "Ks", "Kh", "Qc", "Jc"), ("Tc", "9c", "8d", "7d", "6h", "5h"), ), Spot( "P6-B", "PLO6", "Two Broadway rundowns; chop-heavy", ("As", "Ks", "Qs", "Jh", "Th", "9d"), ("Ad", "Kd", "Qd", "Jc", "Tc", "8c"), ), Spot( "P6-D", "PLO6", "High pairs versus a four-deuce stress test", ("As", "Ah", "Ks", "Kh", "Qc", "Jc"), ("2c", "2d", "2h", "2s", "3d", "4d"), ), ) def card_id(card: str) -> int: if len(card) != 2 or card[0] not in RANKS or card[1] not in SUITS: raise ValueError(f"Invalid card: {card}") return RANKS.index(card[0]) * 4 + SUITS.index(card[1]) def triple_code(a: int, b: int, c: int) -> int: return a * 52 * 52 + b * 52 + c def board_rank(table: list[int], board: tuple[int, int, int, int, int]) -> int: a, b, c, d, e = board best = table[triple_code(a, b, c)] for code in ( triple_code(a, b, d), triple_code(a, b, e), triple_code(a, c, d), triple_code(a, c, e), triple_code(a, d, e), triple_code(b, c, d), triple_code(b, c, e), triple_code(b, d, e), triple_code(c, d, e), ): rank = table[code] if rank < best: best = rank return best def build_triple_table(remaining: tuple[int, ...], hole: tuple[int, ...]) -> list[int]: table = [NOT_A_RANK] * TABLE_SIZE hole_pairs = tuple(combinations(hole, 2)) for a, b, c in combinations(remaining, 3): best = NOT_A_RANK for h1, h2 in hole_pairs: rank = evaluate_5cards(a, b, c, h1, h2) if rank < best: best = rank table[triple_code(a, b, c)] = best return table def exact_spot(spot: Spot) -> tuple[dict[str, object], dict[str, object]]: hero = tuple(card_id(card) for card in spot.hero) villain = tuple(card_id(card) for card in spot.villain) known = hero + villain if len(set(known)) != len(known): raise AssertionError(f"Duplicate card in {spot.key}") if len(hero) != len(villain) or len(hero) not in (5, 6): raise AssertionError(f"Wrong hole-card count in {spot.key}") remaining = tuple(card for card in DECK if card not in set(known)) expected_boards = math.comb(len(remaining), 5) hero_table = build_triple_table(remaining, hero) villain_table = build_triple_table(remaining, villain) wins = ties = losses = 0 for board in combinations(remaining, 5): hero_rank = board_rank(hero_table, board) villain_rank = board_rank(villain_table, board) if hero_rank < villain_rank: wins += 1 elif hero_rank == villain_rank: ties += 1 else: losses += 1 total = wins + ties + losses if total != expected_boards: raise AssertionError(f"Board-count mismatch in {spot.key}: {total} != {expected_boards}") equity = (wins + 0.5 * ties) / total second_moment = (wins + 0.25 * ties) / total variance = second_moment - equity * equity direct_seed = deterministic_seed(spot.key, "direct-board") rng = np.random.Generator(np.random.PCG64(direct_seed)) direct_score = 0.0 direct_wins = direct_ties = direct_losses = 0 for _ in range(DIRECT_BOARD_TRIALS): board = tuple(sorted(int(x) for x in rng.choice(remaining, size=5, replace=False))) hero_rank = board_rank(hero_table, board) villain_rank = board_rank(villain_table, board) if hero_rank < villain_rank: direct_wins += 1 direct_score += 1.0 elif hero_rank == villain_rank: direct_ties += 1 direct_score += 0.5 else: direct_losses += 1 direct_equity = direct_score / DIRECT_BOARD_TRIALS direct_se = math.sqrt(variance / DIRECT_BOARD_TRIALS) direct_z = 0.0 if direct_se == 0 else (direct_equity - equity) / direct_se if abs(direct_z) > 5.5: raise AssertionError(f"Direct-board reduction check failed for {spot.key}: z={direct_z}") exact = { "spot": spot.key, "variant": spot.variant, "label": spot.label, "hero": " ".join(spot.hero), "villain": " ".join(spot.villain), "remaining_cards": len(remaining), "legal_boards": total, "wins": wins, "ties": ties, "losses": losses, "exact_equity": equity, "exact_equity_pct": equity * 100, "outcome_variance": variance, } direct = { "spot": spot.key, "trials": DIRECT_BOARD_TRIALS, "seed": direct_seed, "wins": direct_wins, "ties": direct_ties, "losses": direct_losses, "estimate_pct": direct_equity * 100, "exact_equity_pct": equity * 100, "signed_error_pp": (direct_equity - equity) * 100, "z_score": direct_z, "passed_abs_z_le_5_5": abs(direct_z) <= 5.5, } return exact, direct def deterministic_seed(*parts: object) -> int: payload = "|".join((STUDY_ID, *(str(part) for part in parts))).encode("utf-8") return int.from_bytes(hashlib.sha256(payload).digest()[:8], "big") def monte_carlo_rows(exact_rows: list[dict[str, object]]) -> list[dict[str, object]]: rows: list[dict[str, object]] = [] for exact in exact_rows: total = int(exact["legal_boards"]) probabilities = np.array( [exact["wins"] / total, exact["ties"] / total, exact["losses"] / total], dtype=float, ) truth = float(exact["exact_equity"]) for trials in TRIAL_COUNTS: seed = deterministic_seed(exact["spot"], trials) rng = np.random.Generator(np.random.PCG64(seed)) counts = rng.multinomial(trials, probabilities, size=REPLICATIONS) estimates = (counts[:, 0] + 0.5 * counts[:, 1]) / trials second_moments = (counts[:, 0] + 0.25 * counts[:, 1]) / trials sample_variances = (second_moments - estimates * estimates) * trials / (trials - 1) half_widths = Z_95 * np.sqrt(np.maximum(sample_variances, 0.0) / trials) for replicate, (count, estimate, half_width) in enumerate( zip(counts, estimates, half_widths, strict=True), start=1 ): error = float(estimate - truth) rows.append( { "spot": exact["spot"], "variant": exact["variant"], "trials": trials, "replicate": replicate, "seed": seed, "wins": int(count[0]), "ties": int(count[1]), "losses": int(count[2]), "exact_equity_pct": truth * 100, "estimate_pct": float(estimate) * 100, "signed_error_pp": error * 100, "abs_error_pp": abs(error) * 100, "ci_95_half_width_pp": float(half_width) * 100, "ci_95_covers_exact": bool(abs(error) <= half_width), } ) return rows def percentile(values: np.ndarray, q: float) -> float: return float(np.quantile(values, q, method="linear")) def summarize(rows: list[dict[str, object]]) -> list[dict[str, object]]: summaries: list[dict[str, object]] = [] scopes: list[tuple[str, Iterable[str]]] = [ ("all-six-spots", tuple(spot.key for spot in SPOTS)), ("PLO5", tuple(spot.key for spot in SPOTS if spot.variant == "PLO5")), ("PLO6", tuple(spot.key for spot in SPOTS if spot.variant == "PLO6")), ] scopes.extend((spot.key, (spot.key,)) for spot in SPOTS) for scope, keys in scopes: key_set = set(keys) for trials in TRIAL_COUNTS: selected = [row for row in rows if row["spot"] in key_set and row["trials"] == trials] errors = np.array([row["signed_error_pp"] for row in selected], dtype=float) absolute = np.abs(errors) half_widths = np.array([row["ci_95_half_width_pp"] for row in selected], dtype=float) coverage = np.array([row["ci_95_covers_exact"] for row in selected], dtype=float) summaries.append( { "scope": scope, "trials": trials, "replicates": len(selected), "median_abs_error_pp": percentile(absolute, 0.50), "p90_abs_error_pp": percentile(absolute, 0.90), "p95_abs_error_pp": percentile(absolute, 0.95), "max_abs_error_pp": float(np.max(absolute)), "rmse_pp": float(np.sqrt(np.mean(errors * errors))), "mean_ci_95_half_width_pp": float(np.mean(half_widths)), "ci_95_coverage_pct": float(np.mean(coverage) * 100), } ) return summaries def write_csv(path: Path, rows: list[dict[str, object]]) -> None: path.parent.mkdir(parents=True, exist_ok=True) with path.open("w", newline="", encoding="utf-8") as handle: writer = csv.DictWriter(handle, fieldnames=list(rows[0])) writer.writeheader() writer.writerows(rows) def sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def copy_if_different(source: Path, destination: Path) -> None: if source.resolve() != destination.resolve(): shutil.copyfile(source, destination) def main() -> None: OUTPUT.mkdir(parents=True, exist_ok=True) PUBLIC_DATA.mkdir(parents=True, exist_ok=True) exact_rows: list[dict[str, object]] = [] direct_checks: list[dict[str, object]] = [] for spot in SPOTS: exact, direct = exact_spot(spot) exact_rows.append(exact) direct_checks.append(direct) print( f"{spot.key}: {exact['wins']:,} W / {exact['ties']:,} T / " f"{exact['losses']:,} L = {exact['exact_equity_pct']:.6f}%" ) raw_rows = monte_carlo_rows(exact_rows) summary_rows = summarize(raw_rows) exact_path = OUTPUT / "exact-matchups.csv" raw_path = OUTPUT / "monte-carlo-replicates.csv" summary_path = OUTPUT / "monte-carlo-summary.csv" direct_path = OUTPUT / "direct-board-checks.json" results_path = OUTPUT / "study-results.json" write_csv(exact_path, exact_rows) write_csv(raw_path, raw_rows) write_csv(summary_path, summary_rows) direct_path.write_text(json.dumps(direct_checks, indent=2) + "\n", encoding="utf-8") artifacts = { path.name: sha256(path) for path in (exact_path, raw_path, summary_path, direct_path) } results = { "study_id": STUDY_ID, "run_date": "2026-08-29", "scope": "Fixed known hands, heads-up, preflop, standard PLO high", "equity_scoring": {"win": 1.0, "tie": 0.5, "loss": 0.0}, "trial_counts": list(TRIAL_COUNTS), "replications_per_spot_and_trial_count": REPLICATIONS, "direct_board_trials_per_spot": DIRECT_BOARD_TRIALS, "rng": "NumPy PCG64 with deterministic SHA-256-derived cell seeds", "monte_carlo_cell_seeds": [ { "spot": exact["spot"], "trials": trials, "seed": deterministic_seed(exact["spot"], trials), } for exact in exact_rows for trials in TRIAL_COUNTS ], "dependencies": { "numpy": np.__version__, "phevaluator": version("phevaluator"), }, "spots": [asdict(spot) for spot in SPOTS], "exact_results": exact_rows, "summary": summary_rows, "direct_board_checks": direct_checks, "artifact_sha256": artifacts, "limitations": [ "Six constructed fixed-hand matchups are not a universal worst-case guarantee.", "Range sampling, unknown opponents, multiway pots, double boards, Hi-Lo, runtime, and app performance are not tested.", "The 95% intervals use a normal approximation with sampled outcome variance; empirical coverage is reported.", ], } results_path.write_text(json.dumps(results, indent=2) + "\n", encoding="utf-8") verification_path = OUTPUT / "independent-verification.json" if PUBLISHED_BUNDLE: checksum_paths = [exact_path, raw_path, summary_path, direct_path, results_path] checksum_text = "\n".join( f"{sha256(path)} {path.relative_to(OUTPUT).as_posix()}" for path in checksum_paths ) (OUTPUT / "checksums.sha256").write_text(checksum_text + "\n", encoding="utf-8") print(f"Reproduction outputs written to {OUTPUT}") else: public_exact = PUBLIC_DATA / "plo-monte-carlo-exact-matchups.csv" public_summary = PUBLIC_DATA / "plo-monte-carlo-error-summary.csv" public_results = PUBLIC_DATA / "plo-monte-carlo-study.json" public_raw = PUBLIC_DATA / "plo-monte-carlo-replicates.csv.gz" public_generator = PUBLIC_DATA / "plo-monte-carlo-generate-study.py" public_verifier = PUBLIC_DATA / "plo-monte-carlo-verify-study.py" public_requirements = PUBLIC_DATA / "plo-monte-carlo-requirements.txt" public_verification = PUBLIC_DATA / "plo-monte-carlo-independent-verification.json" copy_if_different(exact_path, public_exact) copy_if_different(summary_path, public_summary) copy_if_different(results_path, public_results) with raw_path.open("rb") as source, public_raw.open("wb") as destination: with gzip.GzipFile( filename="", mode="wb", compresslevel=9, fileobj=destination, mtime=0 ) as compressed: shutil.copyfileobj(source, compressed) copy_if_different(SCRIPT_PATH, public_generator) copy_if_different(BUNDLE / "verify_study.py", public_verifier) copy_if_different(BUNDLE / "requirements.txt", public_requirements) if verification_path.is_file(): copy_if_different(verification_path, public_verification) checksum_paths = [ exact_path, raw_path, summary_path, direct_path, results_path, public_exact, public_summary, public_results, public_raw, public_generator, public_verifier, public_requirements, ] if verification_path.is_file(): checksum_paths.extend((verification_path, public_verification)) checksum_text = "\n".join( f"{sha256(path)} {path.relative_to(ROOT).as_posix()}" for path in checksum_paths ) (BUNDLE / "checksums.sha256").write_text(checksum_text + "\n", encoding="utf-8") all_summary = [row for row in summary_rows if row["scope"] == "all-six-spots"] for row in all_summary: print( f"{row['trials']:>6,} trials: median {row['median_abs_error_pp']:.4f} pp; " f"p95 {row['p95_abs_error_pp']:.4f} pp; coverage {row['ci_95_coverage_pct']:.2f}%" ) if __name__ == "__main__": main()