from __future__ import annotations from dataclasses import dataclass, field from enum import StrEnum from typing import Any from token_holdem.cards import cards_label, new_deck from token_holdem.evaluator import best_players, hand_summary from token_holdem.logging_utils import get_logger, log_event logger = get_logger("token_holdem.engine") STARTING_STACK = 1_000 SMALL_BLIND = 10 BIG_BLIND = 20 class Street(StrEnum): PREFLOP = "preflop" FLOP = "flop" TURN = "turn" RIVER = "river" SHOWDOWN = "showdown" COMPLETE = "complete" @dataclass class PlayerState: id: str name: str stack: int = STARTING_STACK hole_cards: list[str] = field(default_factory=list) folded: bool = False all_in: bool = False bet: int = 0 contributed: int = 0 is_human: bool = False @dataclass class HandResult: winners: list[str] pot: int board: list[str] hand_name: str deltas: dict[str, int] summary: str showdown: dict[str, str] = field(default_factory=dict) pot_awards: list[dict[str, Any]] = field(default_factory=list) @dataclass class GameState: players: list[PlayerState] dealer_index: int = 0 small_blind: int = SMALL_BLIND big_blind: int = BIG_BLIND seed: int | None = None session_id: str = "" hand_id: str = "" orbit_id: str = "" deck: list[str] = field(default_factory=list) community_cards: list[str] = field(default_factory=list) street: Street = Street.PREFLOP current_bet: int = 0 current_player_index: int = 0 small_blind_index: int | None = None big_blind_index: int | None = None last_raise: int = BIG_BLIND action_log: list[str] = field(default_factory=list) acted_player_ids: set[str] = field(default_factory=set) result: HandResult | None = None starting_stacks: dict[str, int] = field(default_factory=dict) def start_hand(self) -> None: self.deck = new_deck(self.seed) self.community_cards = [] self.street = Street.PREFLOP self.current_bet = self.big_blind self.last_raise = self.big_blind self.action_log = [] self.acted_player_ids = set() self.result = None self.starting_stacks = {player.id: player.stack for player in self.players} for player in self.players: player.hole_cards = [self.deck.pop(), self.deck.pop()] player.folded = False player.all_in = False player.bet = 0 player.contributed = 0 if len(self.active_players()) < 2: raise ValueError("At least two players with chips are required") sb_index = self.dealer_index if len(self.active_players()) == 2 else self.next_active_index(self.dealer_index) bb_index = self.next_active_index(sb_index) self.small_blind_index = sb_index self.big_blind_index = bb_index self.post_blind(sb_index, self.small_blind, "small blind") self.post_blind(bb_index, self.big_blind, "big blind") self.current_player_index = self.next_active_index(bb_index) log_event( logger, "hand_started", seed=self.seed, session_id=self.session_id, hand_id=self.hand_id, orbit_id=self.orbit_id, players=[player.name for player in self.players], button=self.players[self.dealer_index].name, small_blind=self.players[sb_index].name, big_blind=self.players[bb_index].name, pot=self.pot(), ) def active_players(self) -> list[PlayerState]: return [player for player in self.players if player.stack > 0 or player.contributed > 0] def live_players(self) -> list[PlayerState]: return [player for player in self.players if not player.folded] def actionable_players(self) -> list[PlayerState]: return [player for player in self.live_players() if not player.all_in and player.stack > 0] def player(self, player_id: str) -> PlayerState: return next(player for player in self.players if player.id == player_id) def current_player(self) -> PlayerState: return self.players[self.current_player_index] def pot(self) -> int: return sum(player.contributed for player in self.players) def amount_to_call(self, player: PlayerState) -> int: return max(0, self.current_bet - player.bet) def legal_actions(self, player_id: str | None = None) -> dict[str, Any]: player = self.player(player_id) if player_id else self.current_player() to_call = min(self.amount_to_call(player), player.stack) actions: list[str] = [] if to_call == 0: actions.append("check") else: actions.extend(["fold", "call"]) if player.stack > to_call: actions.append("raise") if player.stack > 0: actions.append("all_in") min_raise_to = self.current_bet + self.last_raise min_raise_amount = max(0, min_raise_to - player.bet) half_pot = to_call + max(self.big_blind, self.pot() // 2) pot_raise = to_call + max(self.big_blind, self.pot()) presets = { "min": min(player.stack, max(min_raise_amount, to_call + self.big_blind)), "half_pot": min(player.stack, max(half_pot, min_raise_amount)), "pot": min(player.stack, max(pot_raise, min_raise_amount)), "all_in": player.stack, } if "raise" not in actions: presets = {"all_in": player.stack} return {"actions": actions, "to_call": to_call, "raise_presets": presets} def apply_action(self, action: str, amount: int = 0) -> None: if self.street in {Street.SHOWDOWN, Street.COMPLETE} or self.result: return player = self.current_player() legal = self.legal_actions(player.id) requested_action = action requested_amount = amount action = self._repair_action(action, amount, legal) amount = self._repair_amount(action, amount, legal) log_event( logger, "action_applied", player=player.name, street=self.street.value, requested_action=requested_action, requested_amount=requested_amount, action=action, amount=amount, legal=legal, session_id=self.session_id, hand_id=self.hand_id, orbit_id=self.orbit_id, pot_before=self.pot(), stack_before=player.stack, ) if action == "fold": player.folded = True self.acted_player_ids.add(player.id) self.action_log.append(f"{player.name} folds") elif action == "check": self.acted_player_ids.add(player.id) self.action_log.append(f"{player.name} checks") elif action == "call": paid = self.take_chips(player, legal["to_call"]) self.acted_player_ids.add(player.id) self.action_log.append(f"{player.name} calls {paid}") elif action == "raise": previous_bet = player.bet previous_current_bet = self.current_bet paid = self.take_chips(player, amount) self.current_bet = max(self.current_bet, player.bet) self.last_raise = max(self.big_blind, self.current_bet - previous_bet) self.acted_player_ids = {player.id} if previous_current_bet == 0: self.action_log.append(f"{player.name} bets {paid}") else: self.action_log.append(f"{player.name} raises to {player.bet}") elif action == "all_in": previous_bet = player.bet previous_current_bet = self.current_bet paid = self.take_chips(player, player.stack) if player.bet > self.current_bet: self.current_bet = player.bet self.last_raise = max(self.big_blind, self.current_bet - previous_bet) self.acted_player_ids = {player.id} else: self.acted_player_ids.add(player.id) if player.bet > previous_current_bet: self.action_log.append(f"{player.name} jams for {paid} more ({player.bet} total)") else: self.action_log.append(f"{player.name} calls all-in for {paid}") self.advance_after_action() def post_blind(self, index: int, amount: int, label: str) -> None: player = self.players[index] paid = self.take_chips(player, amount) self.action_log.append(f"{player.name} posts {label} {paid}") def take_chips(self, player: PlayerState, amount: int) -> int: paid = min(max(0, amount), player.stack) player.stack -= paid player.bet += paid player.contributed += paid if player.stack == 0: player.all_in = True return paid def advance_after_action(self) -> None: if len(self.live_players()) == 1: self.finish_without_showdown() return if self.betting_round_complete(): self.advance_street() return self.current_player_index = self.next_actionable_index(self.current_player_index) def betting_round_complete(self) -> bool: actionable = self.actionable_players() if not actionable: return True return all(player.bet == self.current_bet and player.id in self.acted_player_ids for player in actionable) def advance_street(self) -> None: for player in self.players: player.bet = 0 self.acted_player_ids = set() self.current_bet = 0 self.last_raise = self.big_blind if self.street == Street.PREFLOP: self.community_cards.extend([self.deck.pop(), self.deck.pop(), self.deck.pop()]) self.street = Street.FLOP self.action_log.append(f"Flop: {cards_label(self.community_cards)}") elif self.street == Street.FLOP: self.community_cards.append(self.deck.pop()) self.street = Street.TURN self.action_log.append(f"Turn: {cards_label(self.community_cards)}") elif self.street == Street.TURN: self.community_cards.append(self.deck.pop()) self.street = Street.RIVER self.action_log.append(f"River: {cards_label(self.community_cards)}") elif self.street == Street.RIVER: self.finish_showdown() return if not self.actionable_players(): self.advance_street() return self.current_player_index = self.next_actionable_index(self.dealer_index) def finish_without_showdown(self) -> None: winner = self.live_players()[0] pot = self.pot() winner.stack += pot self.street = Street.COMPLETE self._set_result([winner.id], pot, "No showdown", f"{winner.name} wins {pot} after everyone else folds.") def finish_showdown(self) -> None: live_players = self.live_players() contenders = {player.id: player.hole_cards for player in live_players} winners, _score, hand_name = best_players(contenders, self.community_cards) showdown = {player.id: hand_summary(player.hole_cards, self.community_cards)[2] for player in self.live_players()} pot = self.pot() awards = self._award_showdown_pots() result_winners = [] for award in awards: for player_id in award["winners"]: if player_id not in result_winners: result_winners.append(player_id) for award in awards: share = award["amount"] // len(award["winners"]) remainder = award["amount"] % len(award["winners"]) for idx, player_id in enumerate(award["winners"]): self.player(player_id).stack += share + (1 if idx < remainder else 0) names = ", ".join(self.player(player_id).name for player_id in winners) self.street = Street.COMPLETE summary = f"{names} win {pot} with {hand_name}." if len(awards) > 1 or set(awards[0]["winners"]) != set(winners): parts = [] for award in awards: award_names = ", ".join(self.player(player_id).name for player_id in award["winners"]) parts.append(f"{award_names} win {award['amount']} with {award['hand_name']}") summary = "Side pots: " + "; ".join(parts) + "." self._set_result(result_winners or winners, pot, hand_name, summary, showdown=showdown, pot_awards=awards) def _award_showdown_pots(self) -> list[dict[str, Any]]: awards: list[dict[str, Any]] = [] previous_level = 0 levels = sorted({player.contributed for player in self.players if player.contributed > 0}) for level in levels: contributors = [player for player in self.players if player.contributed >= level] amount = (level - previous_level) * len(contributors) eligible = [player for player in contributors if not player.folded] previous_level = level if amount <= 0 or not eligible: continue contenders = {player.id: player.hole_cards for player in eligible} winners, _score, hand_name = best_players(contenders, self.community_cards) awards.append({"amount": amount, "winners": winners, "hand_name": hand_name}) return awards def _set_result(self, winners: list[str], pot: int, hand_name: str, summary: str, showdown: dict[str, str] | None = None, pot_awards: list[dict[str, Any]] | None = None) -> None: deltas = {player.id: player.stack - self.starting_stacks[player.id] for player in self.players} self.result = HandResult(winners=winners, pot=pot, board=list(self.community_cards), hand_name=hand_name, deltas=deltas, summary=summary, showdown=showdown or {}, pot_awards=pot_awards or []) self.action_log.append(summary) log_event( logger, "hand_completed", winners=winners, session_id=self.session_id, hand_id=self.hand_id, orbit_id=self.orbit_id, pot=pot, hand_name=hand_name, board=self.community_cards, deltas=deltas, showdown=showdown or {}, pot_awards=pot_awards or [], summary=summary, ) def next_active_index(self, start_index: int) -> int: return self._next_index(start_index, lambda player: player.stack > 0) def next_actionable_index(self, start_index: int) -> int: return self._next_index(start_index, lambda player: not player.folded and not player.all_in and player.stack > 0) def _next_index(self, start_index: int, predicate) -> int: for offset in range(1, len(self.players) + 1): idx = (start_index + offset) % len(self.players) if predicate(self.players[idx]): return idx return start_index def _repair_action(self, action: str, amount: int, legal: dict[str, Any]) -> str: action = (action or "").lower().strip() if action in legal["actions"]: return action if legal["to_call"] == 0: return "check" if "call" in legal["actions"]: return "call" return "fold" def _repair_amount(self, action: str, amount: int, legal: dict[str, Any]) -> int: if action != "raise": return 0 presets = legal["raise_presets"] legal_amounts = [value for key, value in presets.items() if key != "all_in" and value > legal["to_call"]] if not legal_amounts: return 0 if amount <= 0: return min(legal_amounts) return min(legal_amounts, key=lambda value: abs(value - amount)) def make_players(names: list[str], human_name: str | None = None) -> list[PlayerState]: players: list[PlayerState] = [] if human_name: players.append(PlayerState(id="human", name=human_name, is_human=True)) for idx, name in enumerate(names): players.append(PlayerState(id=f"ai_{idx}", name=name)) return players