#!/usr/bin/env python3 """Exact same-stub run-it-twice enumeration using only Python's standard library. Local/public mode is the safe default and writes beside the script. Repository maintainers may pass --repo-sync from the retained research-bundle location. """ from __future__ import annotations import argparse import csv import hashlib import json import math import shutil from collections import Counter from fractions import Fraction from itertools import combinations from pathlib import Path STUDY_ID = "gtogecko-run-it-twice-2026-09-03" ACCESSED = "2026-09-03" RANKS = "23456789TJQKA" SUITS = "cdhs" DECK = tuple(rank + suit for rank in RANKS for suit in SUITS) SCRIPT_PATH = Path(__file__).resolve() BUNDLE = SCRIPT_PATH.parent def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument( "--repo-sync", action="store_true", help="Use the retained repository fixture/output paths and sync public data.", ) return parser.parse_args() ARGS = parse_args() REPO_SYNC = ARGS.repo_sync if REPO_SYNC: REPO_ROOT = BUNDLE.parents[2] if BUNDLE.name != "2026-09-03-running-it-twice" or not (REPO_ROOT / "package.json").is_file(): raise SystemExit("--repo-sync is allowed only from the retained repository research bundle") INPUT_PATH = BUNDLE / "scenarios.csv" OUTPUT_DIR = BUNDLE / "outputs" else: REPO_ROOT = None INPUT_PATH = BUNDLE / "run-it-twice-scenarios.csv" OUTPUT_DIR = BUNDLE / "run-it-twice-output" def fraction_record(value: Fraction, places: int = 12) -> dict[str, str]: return { "fraction": f"{value.numerator}/{value.denominator}", "decimal": f"{float(value):.{places}f}", } def rank_five(cards: tuple[str, ...]) -> tuple[int, ...]: ranks = sorted((RANKS.index(card[0]) + 2 for card in cards), reverse=True) counts = Counter(ranks) groups = sorted(((count, rank) for rank, count in counts.items()), reverse=True) flush = len({card[1] for card in cards}) == 1 unique = sorted(set(ranks), reverse=True) if 14 in unique: unique.append(1) straight_high = next( (unique[i] for i in range(len(unique) - 4) if unique[i] - unique[i + 4] == 4), 0, ) if flush and straight_high: return (8, straight_high) if groups[0][0] == 4: four = groups[0][1] return (7, four, max(rank for rank in ranks if rank != four)) trips = sorted((rank for rank, count in counts.items() if count == 3), reverse=True) pairs = sorted((rank for rank, count in counts.items() if count == 2), reverse=True) if trips and (pairs or len(trips) > 1): return (6, trips[0], pairs[0] if pairs else trips[1]) if flush: return (5, *ranks) if straight_high: return (4, straight_high) if trips: kickers = sorted((rank for rank in ranks if rank != trips[0]), reverse=True)[:2] return (3, trips[0], *kickers) if len(pairs) >= 2: kicker = max(rank for rank in ranks if rank not in pairs[:2]) return (2, pairs[0], pairs[1], kicker) if len(pairs) == 1: kickers = sorted((rank for rank in ranks if rank != pairs[0]), reverse=True)[:3] return (1, pairs[0], *kickers) return (0, *ranks) def rank_seven(cards: tuple[str, ...]) -> tuple[int, ...]: return max(rank_five(combo) for combo in combinations(cards, 5)) def half_pot_units( hero: tuple[str, str], villain: tuple[str, str], board: tuple[str, ...] ) -> int: """Return Hero's share in half-pot units: loss=0, tie=1, win=2.""" hero_rank = rank_seven(hero + board) villain_rank = rank_seven(villain + board) if hero_rank > villain_rank: return 2 if hero_rank < villain_rank: return 0 return 1 def validate_card(code: str) -> None: if len(code) != 2 or code[0] not in RANKS or code[1] not in SUITS: raise ValueError(f"Invalid card code: {code!r}") def load_scenarios(path: Path) -> list[dict[str, object]]: scenarios: list[dict[str, object]] = [] with path.open(newline="", encoding="utf-8") as handle: for row in csv.DictReader(handle): hero = (row["hero_card_1"], row["hero_card_2"]) villain = (row["villain_card_1"], row["villain_card_2"]) board = tuple(row[key] for key in ("board_1", "board_2", "board_3", "board_4") if row[key]) cards = hero + villain + board for card in cards: validate_card(card) if len(set(cards)) != len(cards): raise ValueError(f"Duplicate card in {row['scenario_id']}") expected_board = 4 if row["stage"] == "turn" else 3 if row["stage"] == "flop" else -1 if len(board) != expected_board: raise ValueError(f"Stage/board mismatch in {row['scenario_id']}") scenarios.append( { "scenario_id": row["scenario_id"], "label": row["label"], "stage": row["stage"], "hero": hero, "villain": villain, "board": board, } ) if not scenarios: raise ValueError("No scenarios found") return scenarios def analyze_scenario(scenario: dict[str, object]) -> dict[str, object]: hero = scenario["hero"] villain = scenario["villain"] board = scenario["board"] assert isinstance(hero, tuple) and isinstance(villain, tuple) and isinstance(board, tuple) known = set(hero + villain + board) stub = tuple(card for card in DECK if card not in known) cards_to_come = 5 - len(board) runouts = list(combinations(stub, cards_to_come)) shares = [half_pot_units(hero, villain, board + runout) for runout in runouts] single_count = len(runouts) equity = Fraction(sum(shares), 2 * single_count) run_once_variance = Fraction(sum(value * value for value in shares), 4 * single_count) - equity * equity paired_count = 0 quarter_sum = 0 quarter_square_sum = 0 product_sum = 0 outcome_counts: Counter[int] = Counter() for index, first in enumerate(runouts): first_cards = set(first) first_share = shares[index] for second_index, second in enumerate(runouts): if first_cards.isdisjoint(second): second_share = shares[second_index] quarter_units = first_share + second_share paired_count += 1 quarter_sum += quarter_units quarter_square_sum += quarter_units * quarter_units product_sum += first_share * second_share outcome_counts[quarter_units] += 1 run_twice_equity = Fraction(quarter_sum, 4 * paired_count) run_twice_variance = Fraction(quarter_square_sum, 16 * paired_count) - run_twice_equity * run_twice_equity covariance = Fraction(product_sum, 4 * paired_count) - equity * equity expected_pairs = math.comb(len(stub), cards_to_come) * math.comb( len(stub) - cards_to_come, cards_to_come ) if paired_count != expected_pairs: raise AssertionError((paired_count, expected_pairs)) if run_twice_equity != equity: raise AssertionError((run_twice_equity, equity)) if run_twice_variance != run_once_variance / 2 + covariance / 2: raise AssertionError("Variance/covariance identity failed") variance_ratio = run_twice_variance / run_once_variance standard_deviation_ratio = math.sqrt(float(variance_ratio)) outcome_labels = {0: "zero", 1: "quarter", 2: "half", 3: "three_quarters", 4: "scoop"} outcomes = { outcome_labels[units]: { "pot_share": f"{units}/4", "count": outcome_counts.get(units, 0), "probability": fraction_record(Fraction(outcome_counts.get(units, 0), paired_count)), } for units in range(5) } single_distribution = Counter(shares) return { "scenario_id": scenario["scenario_id"], "label": scenario["label"], "stage": scenario["stage"], "hero": list(hero), "villain": list(villain), "board": list(board), "stub_cards": len(stub), "cards_to_come_per_board": cards_to_come, "single_runout_count": single_count, "paired_runout_count": paired_count, "hero_equity_run_once": fraction_record(equity), "hero_equity_run_twice": fraction_record(run_twice_equity), "run_once_variance": fraction_record(run_once_variance), "run_twice_variance": fraction_record(run_twice_variance), "independent_two_board_variance": fraction_record(run_once_variance / 2), "same_stub_covariance": fraction_record(covariance), "variance_retained": fraction_record(variance_ratio), "standard_deviation_retained": f"{standard_deviation_ratio:.12f}", "single_run_distribution": { "losses": single_distribution.get(0, 0), "ties": single_distribution.get(1, 0), "wins": single_distribution.get(2, 0), }, "run_twice_outcomes": outcomes, } def write_results_csv(path: Path, results: list[dict[str, object]]) -> None: fields = [ "scenario_id", "label", "stage", "hero", "villain", "board", "stub_cards", "cards_to_come_per_board", "single_runout_count", "paired_runout_count", "hero_equity_pct", "run_once_variance", "run_twice_variance", "variance_retained_pct", "standard_deviation_retained_pct", "same_stub_covariance", "scoop_pct", "three_quarters_pct", "half_pct", "quarter_pct", "zero_pct", ] with path.open("w", newline="", encoding="utf-8") as handle: writer = csv.DictWriter(handle, fieldnames=fields, lineterminator="\n") writer.writeheader() for result in results: outcomes = result["run_twice_outcomes"] assert isinstance(outcomes, dict) writer.writerow( { "scenario_id": result["scenario_id"], "label": result["label"], "stage": result["stage"], "hero": " ".join(result["hero"]), "villain": " ".join(result["villain"]), "board": " ".join(result["board"]), "stub_cards": result["stub_cards"], "cards_to_come_per_board": result["cards_to_come_per_board"], "single_runout_count": result["single_runout_count"], "paired_runout_count": result["paired_runout_count"], "hero_equity_pct": f"{100 * float(Fraction(result['hero_equity_run_once']['fraction'])):.6f}", "run_once_variance": result["run_once_variance"]["decimal"], "run_twice_variance": result["run_twice_variance"]["decimal"], "variance_retained_pct": f"{100 * float(Fraction(result['variance_retained']['fraction'])):.6f}", "standard_deviation_retained_pct": f"{100 * float(result['standard_deviation_retained']):.6f}", "same_stub_covariance": result["same_stub_covariance"]["decimal"], "scoop_pct": f"{100 * float(Fraction(outcomes['scoop']['probability']['fraction'])):.6f}", "three_quarters_pct": f"{100 * float(Fraction(outcomes['three_quarters']['probability']['fraction'])):.6f}", "half_pct": f"{100 * float(Fraction(outcomes['half']['probability']['fraction'])):.6f}", "quarter_pct": f"{100 * float(Fraction(outcomes['quarter']['probability']['fraction'])):.6f}", "zero_pct": f"{100 * float(Fraction(outcomes['zero']['probability']['fraction'])):.6f}", } ) def self_test_evaluator() -> None: tests = [ (("As", "Ks", "Qs", "Js", "Ts"), (8, 14)), (("9c", "9d", "9h", "9s", "2d"), (7, 9, 2)), (("Ac", "Ad", "Ah", "Kc", "Kd"), (6, 14, 13)), (("Ac", "Jc", "9c", "5c", "2c"), (5, 14, 11, 9, 5, 2)), (("5c", "4d", "3h", "2s", "Ac"), (4, 5)), ] for cards, expected in tests: actual = rank_five(cards) if actual != expected: raise AssertionError((cards, actual, expected)) def write_outputs() -> None: self_test_evaluator() scenarios = load_scenarios(INPUT_PATH) results = [analyze_scenario(scenario) for scenario in scenarios] total_pairs = sum(int(result["paired_runout_count"]) for result in results) study = { "study_id": STUDY_ID, "study_date": "2026-09-03", "source_access_date": ACCESSED, "method": "Exact exhaustive enumeration; no simulation or random seed.", "model": { "game": "Heads-up Texas Hold'em showdown after all betting is complete", "settlement": "Two equal pot fractions, one attached to each board", "dealing": "Remaining visible board cards are sampled successively without replacement from one uniformly shuffled stub", "burn_card_note": "Unobserved burn cards are marginalized out; projecting a uniform shuffled stub onto visible board positions yields the same uniform disjoint runout distribution.", "share_values": "Each run awards Hero 0, 1/2, or 1 of that run's pot fraction.", "excluded": [ "fees that change when the feature is used", "known dead or exposed cards not listed in the fixture", "action-conditioned folded-card ranges", "multiway pots and side-pot eligibility", "unequal board weights", "strategic action before the board is locked", ], }, "sample_space": { "flop_single_visible_runouts": "C(45,2) = 990", "flop_labeled_disjoint_board_pairs": "C(45,2) * C(43,2) = 893970", "turn_single_visible_rivers": "44", "turn_labeled_distinct_river_pairs": "44 * 43 = 1892", "total_paired_runouts_across_fixtures": total_pairs, }, "sources": [ { "title": "Variance reduction in Texas hold'em and in video poker", "url": "https://arxiv.org/html/2409.03607v2", "accessed": ACCESSED, }, { "title": "PokerStars: What is Run It Twice?", "url": "https://www.pokerstars.com/help/articles/run-it-twice-new/15582/", "accessed": ACCESSED, }, ], "results": results, } OUTPUT_DIR.mkdir(parents=True, exist_ok=True) study_path = OUTPUT_DIR / "study.json" results_path = OUTPUT_DIR / "results.csv" with study_path.open("w", encoding="utf-8", newline="\n") as handle: json.dump(study, handle, indent=2) handle.write("\n") write_results_csv(results_path, results) shutil.copyfile(INPUT_PATH, OUTPUT_DIR / "scenarios.csv") if REPO_SYNC: assert REPO_ROOT is not None public_data = REPO_ROOT / "public" / "data" public_data.mkdir(parents=True, exist_ok=True) shutil.copyfile(INPUT_PATH, public_data / "run-it-twice-scenarios.csv") shutil.copyfile(results_path, public_data / "run-it-twice-results.csv") shutil.copyfile(study_path, public_data / "run-it-twice-study.json") digest = hashlib.sha256(study_path.read_bytes()).hexdigest() print(f"Wrote {study_path}") print(f"Wrote {results_path}") print(f"Enumerated {total_pairs:,} outcome-equivalent labeled final-board pairs") print(f"study.json sha256 {digest}") if __name__ == "__main__": write_outputs()