diff --git a/shortdeck_arena/__pycache__/card.cpython-313.pyc b/shortdeck_arena/__pycache__/card.cpython-313.pyc index bb2f87c..0949e26 100644 Binary files a/shortdeck_arena/__pycache__/card.cpython-313.pyc and b/shortdeck_arena/__pycache__/card.cpython-313.pyc differ diff --git a/shortdeck_arena/__pycache__/game_stage.cpython-313.pyc b/shortdeck_arena/__pycache__/game_stage.cpython-313.pyc index 8653e8f..75546cd 100644 Binary files a/shortdeck_arena/__pycache__/game_stage.cpython-313.pyc and b/shortdeck_arena/__pycache__/game_stage.cpython-313.pyc differ diff --git a/shortdeck_arena/__pycache__/hand_evaluator.cpython-313.pyc b/shortdeck_arena/__pycache__/hand_evaluator.cpython-313.pyc new file mode 100644 index 0000000..6b61c53 Binary files /dev/null and b/shortdeck_arena/__pycache__/hand_evaluator.cpython-313.pyc differ diff --git a/shortdeck_arena/__pycache__/hand_ranking.cpython-313.pyc b/shortdeck_arena/__pycache__/hand_ranking.cpython-313.pyc new file mode 100644 index 0000000..ddce3d1 Binary files /dev/null and b/shortdeck_arena/__pycache__/hand_ranking.cpython-313.pyc differ diff --git a/shortdeck_arena/__pycache__/side_pot.cpython-313.pyc b/shortdeck_arena/__pycache__/side_pot.cpython-313.pyc new file mode 100644 index 0000000..6db3e82 Binary files /dev/null and b/shortdeck_arena/__pycache__/side_pot.cpython-313.pyc differ diff --git a/shortdeck_arena/__pycache__/simulation.cpython-313.pyc b/shortdeck_arena/__pycache__/simulation.cpython-313.pyc index b0ea0eb..1747fda 100644 Binary files a/shortdeck_arena/__pycache__/simulation.cpython-313.pyc and b/shortdeck_arena/__pycache__/simulation.cpython-313.pyc differ diff --git a/shortdeck_arena/card.py b/shortdeck_arena/card.py index 3da1e94..eb4ba5c 100644 --- a/shortdeck_arena/card.py +++ b/shortdeck_arena/card.py @@ -30,6 +30,14 @@ class Rank(IntEnum): if self.value <= 9: return str(self.value) return {10: "T", 11: "J", 12: "Q", 13: "K", 14: "A"}[self.value] + + @property + def numeric_value(self): + return self.value + + @property + def symbol(self): + return str(self) class Card: diff --git a/shortdeck_arena/game_stage.py b/shortdeck_arena/game_stage.py index 8406d57..b99b8e5 100644 --- a/shortdeck_arena/game_stage.py +++ b/shortdeck_arena/game_stage.py @@ -41,8 +41,8 @@ class PlayerState(Enum): ALLIN = "allin" # 全下 CALLED = "called" # 已跟注 RAISE = "raised" # 已加注 - WAITING = "waiting" # 等待其他玩家 - OUT = "out" # 出局 + # WAITING = "waiting" # 等待其他玩家 + # OUT = "out" # 出局 def __str__(self) -> str: return self.value @@ -74,7 +74,7 @@ class BlindConfig: return (dealer_position + 1) % 2 # heads-up: 大盲 = 庄位 return (dealer_position + 2) % num_players # 多人: 大盲为庄位后两位 - def get_first_to_act(self, stage: GameStage, num_players) -> int: + def get_first_to_act(self, stage: GameStage, num_players, dealer_position=0) -> int: """ 获取首个行动玩家位置 区分preflop和postflop @@ -83,9 +83,9 @@ class BlindConfig: """ if stage == GameStage.PREFLOP: if num_players == 2: - return self.get_sb_position(num_players) # heads-up: 小盲先行动 + return self.get_sb_position(num_players, dealer_position) # heads-up: 小盲先行动 else: # preflop: 大盲后第一位 - return (self.get_bb_position(num_players) + 1) % num_players + return (self.get_bb_position(num_players, dealer_position) + 1) % num_players else: # flop/river/turn: 小盲位先行动 - return self.get_sb_position(num_players) \ No newline at end of file + return self.get_sb_position(num_players, dealer_position) \ No newline at end of file diff --git a/shortdeck_arena/hand_evaluator.py b/shortdeck_arena/hand_evaluator.py index 455f884..5e66846 100644 --- a/shortdeck_arena/hand_evaluator.py +++ b/shortdeck_arena/hand_evaluator.py @@ -91,9 +91,8 @@ class HandEvaluator: @staticmethod def _isStraight(ranks: List[Rank]) -> Tuple[bool, Rank]: - values = sorted([rank.numeric_value for rank in ranks], reverse=True) - + is_regular_straight = True for i in range(1, len(values)): if values[i-1] - values[i] != 1: @@ -101,7 +100,6 @@ class HandEvaluator: break if is_regular_straight: - # 返回最高牌 highest_rank = None for rank in ranks: if rank.numeric_value == values[0]: @@ -109,7 +107,6 @@ class HandEvaluator: break return True, highest_rank - if values == [14, 5, 4, 3, 2]: # A, 5, 4, 3, 2 - return True, Rank.FIVE - + if values == {14, 9, 8, 7, 6}: # A, T, 9, 8, 7, 6 + return True, Rank.R9 return False, None \ No newline at end of file diff --git a/shortdeck_arena/hand_ranking.py b/shortdeck_arena/hand_ranking.py index 9a29768..c6cd0d5 100644 --- a/shortdeck_arena/hand_ranking.py +++ b/shortdeck_arena/hand_ranking.py @@ -4,13 +4,14 @@ from .card import Card, Rank class HandType(Enum): + # 短牌规则：同花 > 葫芦 HIGH_CARD = (1, "High Card") ONE_PAIR = (2, "Pair") TWO_PAIR = (3, "Two Pair") THREE_OF_A_KIND = (4, "Three of a Kind") STRAIGHT = (5, "Straight") - FLUSH = (6, "Flush") - FULL_HOUSE = (7, "Full House") + FULL_HOUSE = (6, "Full House") + FLUSH = (7, "Flush") FOUR_OF_A_KIND = (8, "Four of a Kind") STRAIGHT_FLUSH = (9, "Straight Flush") ROYAL_FLUSH = (10, "Royal Flush") @@ -55,3 +56,26 @@ class HandRanking: return f"Pair({self.key_ranks[0].symbol})" else: return f"High Card({self.key_ranks[0].symbol})" + + def __lt__(self, other): + """比较牌力，用于排序""" + if not isinstance(other, HandRanking): + return NotImplemented + + if self.hand_type.strength != other.hand_type.strength: + return self.hand_type.strength < other.hand_type.strength + + for my_rank, other_rank in zip(self.key_ranks, other.key_ranks): + if my_rank.numeric_value != other_rank.numeric_value: + return my_rank.numeric_value < other_rank.numeric_value + + return False + def get_strength(self) -> int: + # 返回牌力 还是 牌型＋点数 + # 基础强度 = 牌型强度 * 1000000 + strength = self.hand_type.strength * 1000000 + + for i, rank in enumerate(self.key_ranks): + strength += rank.numeric_value * (100 ** (4 - i)) + + return strength diff --git a/shortdeck_arena/side_pot.py b/shortdeck_arena/side_pot.py new file mode 100644 index 0000000..bad36a6 --- /dev/null +++ b/shortdeck_arena/side_pot.py @@ -0,0 +1,104 @@ +from __future__ import annotations + +from typing import List, Set, Dict +from dataclasses import dataclass + + +@dataclass +class SidePot: + amount: int + eligible_players: Set[int] + is_main_pot: bool = False + + def __post_init__(self): + if not self.eligible_players: + self.eligible_players = set() + + +class SidePotManager: + def __init__(self): + self.pots: List[SidePot] = [] + self.player_total_investment: Dict[int, int] = {} + + def add_investment(self, player_id: int, amount: int): + if player_id not in self.player_total_investment: + self.player_total_investment[player_id] = 0 + self.player_total_investment[player_id] += amount + + def create_side_pots(self, active_players: List[int]) -> List[SidePot]: + if not self.player_total_investment: + return [] + + # 按投入金额排序 + sorted_investments = sorted( + [(pid, amount) for pid, amount in self.player_total_investment.items() + if pid in active_players and amount > 0], + key=lambda x: x[1] + ) + + if not sorted_investments: + return [] + + pots = [] + prev_level = 0 + + for i, (player_id, investment) in enumerate(sorted_investments): + if investment > prev_level: + # 计算本层级的贡献 + level_contribution = investment - prev_level + + # 找到有资格竞争本层级的玩家 + eligible_players = { + pid for pid, inv in sorted_investments[i:] + if inv >= investment + } + + pot_amount = level_contribution * len(eligible_players) + + side_pot = SidePot( + amount=pot_amount, + eligible_players=eligible_players, + is_main_pot=(len(pots) == 0) + ) + + pots.append(side_pot) + prev_level = investment + + self.pots = pots + return pots + + def get_total_pot(self) -> int: + return sum(pot.amount for pot in self.pots) + + def distribute_winnings(self, hand_rankings: Dict[int, int]) -> Dict[int, int]: + winnings = {} + + for pot in self.pots: + + eligible = [pid for pid in pot.eligible_players if pid in hand_rankings] + if not eligible: + continue + + # 找到最强手牌 + best_strength = max(hand_rankings[pid] for pid in eligible) + winners = [pid for pid in eligible + if hand_rankings[pid] == best_strength] + + # 平分底池 + winnings_per_winner = pot.amount // len(winners) + remainder = pot.amount % len(winners) + + for i, winner in enumerate(winners): + if winner not in winnings: + winnings[winner] = 0 + winnings[winner] += winnings_per_winner + # 余数给前面的获胜者 + if i < remainder: + winnings[winner] += 1 + + return winnings + + def reset(self): + """重置边池管理器""" + self.pots.clear() + self.player_total_investment.clear() \ No newline at end of file diff --git a/shortdeck_arena/simulation.py b/shortdeck_arena/simulation.py index 8b6a390..ca48dc7 100644 --- a/shortdeck_arena/simulation.py +++ b/shortdeck_arena/simulation.py @@ -10,6 +10,9 @@ if TYPE_CHECKING: from .agent import Agent from .card import Card from .game_stage import GameStage, PlayerState, BlindConfig +from .side_pot import SidePotManager +from .hand_evaluator import HandEvaluator +from .hand_ranking import HandRanking class Simulation: @@ -35,6 +38,10 @@ class Simulation: self.min_raise = self.blind_config.big_blind self.dealer_position = -1 + # 边池管理和筹码 + self.side_pot_manager = SidePotManager() + self.stacks: List[int] = [1000] * len(agents) # 默认筹码 + self.new_round() def new_round(self): @@ -54,6 +61,9 @@ class Simulation: self.last_raise_amount = 0 self.min_raise = self.blind_config.big_blind + # 重置边池管理器 + self.side_pot_manager.reset() + # 设置盲注 self._setup_blinds() @@ -61,7 +71,6 @@ class Simulation: self.dealer_position = random.choice(range(len(self.agents))) def _setup_blinds(self): - """设置盲注""" num_players = len(self.agents) # 至少需要2个玩家才能设置盲注 @@ -78,25 +87,31 @@ class Simulation: return # 扣除小盲 - self.pot[sb_pos] = self.blind_config.small_blind - self.total_pot += self.blind_config.small_blind + sb_amount = min(self.blind_config.small_blind, self.stacks[sb_pos]) + self.pot[sb_pos] = sb_amount + self.stacks[sb_pos] -= sb_amount + self.total_pot += sb_amount + self.side_pot_manager.add_investment(sb_pos, sb_amount) self.history.append({ "pid": sb_pos, "action": "small_blind", - "amount": self.blind_config.small_blind + "amount": sb_amount }) # 扣除大盲 - self.pot[bb_pos] = self.blind_config.big_blind - self.total_pot += self.blind_config.big_blind + bb_amount = min(self.blind_config.big_blind, self.stacks[bb_pos]) + self.pot[bb_pos] = bb_amount + self.stacks[bb_pos] -= bb_amount + self.total_pot += bb_amount + self.side_pot_manager.add_investment(bb_pos, bb_amount) self.history.append({ "pid": bb_pos, "action": "big_blind", - "amount": self.blind_config.big_blind + "amount": bb_amount }) # 首个行动玩家 - self.current_turn = self.blind_config.get_first_to_act(self.current_stage, num_players) + self.current_turn = self.blind_config.get_first_to_act(self.current_stage, num_players, self.dealer_position) self.last_raise_amount = self.blind_config.big_blind def player_cards(self, pid) -> List[Card]: @@ -114,19 +129,113 @@ class Simulation: return [] def get_current_max_bet(self) -> int: - """ - 获取当前最高下注额, 用于计算跟注金额 - """ return max(self.pot) if self.pot else 0 def get_call_amount(self, pid) -> int: """ - 计算玩家跟注所需金额 + 跟注金额 """ if pid >= len(self.pot): return 0 max_pot = self.get_current_max_bet() return max(0, max_pot - self.pot[pid]) + + def get_min_raise_amount(self, pid) -> int: + """最小加注金额""" + call_amount = self.get_call_amount(pid) + min_raise = call_amount + max(self.last_raise_amount, self.blind_config.big_blind) + return min_raise + + def get_max_bet_amount(self, pid) -> int: + """最大下注金额（剩余筹码）""" + if pid >= len(self.stacks): + return 0 + return self.stacks[pid] + + def is_all_in_amount(self, pid, amount) -> bool: + """检查是否为allin""" + return amount >= self.stacks[pid] + + def validate_bet_amount(self, pid, action, amount) -> tuple[bool, str, int]: + """ + 验证下注金额合法性 + """ + if pid >= len(self.stacks): + return False, "无效玩家", amount + + available_stack = self.stacks[pid] + call_amount = self.get_call_amount(pid) + + if action == "fold": + return True, "", 0 + + elif action == "check": + if call_amount > 0: + return False, "不能过牌，需跟注或弃牌", 0 + return True, "", 0 + + elif action == "call": + if call_amount == 0: + return False, "不需要跟注", 0 + + # All-in call + if call_amount >= available_stack: + return True, "", available_stack + + return True, "", call_amount + + elif action in ["bet", "raise"]: + if amount <= 0: + return False, "无效下注金额", amount + + # allin + if amount >= available_stack: + return True, "", available_stack + + + if action == "raise": + min_raise = self.get_min_raise_amount(pid) + if amount < min_raise: + return False, f"最小加注金额为 {min_raise}", amount + + if action == "bet" and max(self.pot) == 0: + if amount < self.blind_config.big_blind: + return False, f"最小下注金额为 {self.blind_config.big_blind}", amount + + return True, "", amount + + return False, "无效行为", amount + + def get_available_actions(self, pid: int) -> dict: + if pid != self.current_turn: + return {"can_act": False, "reason": "不是你的回合"} + + if pid >= len(self.player_states): + return {"can_act": False, "reason": "无效玩家"} + + state = self.player_states[pid] + if state in [PlayerState.FOLDED, PlayerState.ALLIN, PlayerState.OUT]: + return {"can_act": False, "reason": f"Player state: {state}"} + + call_amount = self.get_call_amount(pid) + available_stack = self.stacks[pid] + + actions = { + "can_act": True, + "can_fold": True, + "can_check": call_amount == 0, + "can_call": call_amount > 0 and call_amount < available_stack, + "can_bet": max(self.pot) == 0 and available_stack > 0, + "can_raise": call_amount > 0 and available_stack > call_amount, + "can_allin": available_stack > 0, + "call_amount": call_amount, + "min_bet": self.blind_config.big_blind if max(self.pot) == 0 else 0, + "min_raise": self.get_min_raise_amount(pid) if call_amount > 0 else 0, + "max_bet": available_stack, + "stack": available_stack + } + + return actions def is_betting_round_complete(self) -> bool: """ @@ -138,37 +247,59 @@ class Simulation: if len(active_players) <= 1: return True - # 检查所有active玩家是否都已投入相同金额 + # 检查所有active玩家是否都已投入相同金额，且所有人都已经行动过 max_pot = self.get_current_max_bet() - for i in active_players: # allin玩家不要求跟注 - if self.pot[i] < max_pot and self.player_states[i] != PlayerState.ALLIN: - return False - return True + + # 统计还需要行动的玩家 + players_need_action = [] + for i in active_players: + # allin + if self.player_states[i] == PlayerState.ALLIN: + continue + # 投入金额不足的玩家需要行动 + if self.pot[i] < max_pot: + players_need_action.append(i) + # Active状态的玩家如果还没有在本轮行动过，也需要行动 + elif self.player_states[i] == PlayerState.ACTIVE: + # 在翻前，大盲玩家即使投入了足够金额，也有权行动一次 + if (self.current_stage == GameStage.PREFLOP and + i == self.blind_config.get_bb_position(len(self.agents), self.dealer_position)): + # 检查大盲是否已经行动过（除了盲注） + bb_actions = [h for h in self.history if h.get('pid') == i and h.get('action') not in ['big_blind']] + if not bb_actions: + players_need_action.append(i) + + return len(players_need_action) == 0 def advance_to_next_street(self): if self.current_stage == GameStage.FINISHED: return - + next_stage = GameStage.get_next_stage(self.current_stage) if next_stage is None: self.current_stage = GameStage.FINISHED + self.complete_hand() return self.current_stage = next_stage + active_players = self.get_active_players() + if len(active_players) <= 1: + self.current_stage = GameStage.FINISHED + self.complete_hand() + return + # 重置下注轮状态 self.betting_round_complete = False - #### 重置pot为累计投入 (不清零，为了计算边池) - # 重置行动状态 for i, state in enumerate(self.player_states): if state == PlayerState.CALLED: self.player_states[i] = PlayerState.ACTIVE - # 设置首个行动玩家 + # 首个行动玩家 num_players = len(self.agents) - self.current_turn = self.blind_config.get_first_to_act(self.current_stage, num_players) + self.current_turn = self.blind_config.get_first_to_act(self.current_stage, num_players, self.dealer_position) self.last_raise_amount = 0 self.min_raise = self.blind_config.big_blind @@ -179,21 +310,25 @@ class Simulation: return pos return None + def get_side_pots(self) -> List: + active_players = [ + i for i, state in enumerate(self.player_states) + if state not in [PlayerState.FOLDED, PlayerState.OUT] + ] + return self.side_pot_manager.create_side_pots(active_players) + def node_info(self) -> Dict: if self.current_turn >= len(self.pot): return {"bet_min": self.min_raise, "bet_max": 0, "call_amount": 0} - # 跟注 - call_amount = self.get_call_amount(self.current_turn) + + actions = self.get_available_actions(self.current_turn) return { - "bet_min": max(self.min_raise, call_amount + self.min_raise), - "bet_max": 100, ########## 需要从ArenaGame获取实际大小 - "call_amount": call_amount + "bet_min": actions.get("min_bet", self.min_raise), + "bet_max": actions.get("max_bet", 100), + "call_amount": actions.get("call_amount", 0) } def apply_action(self, pid, action, amount): - """ - 应用玩家动作 - """ if pid != self.current_turn: raise ValueError(f"不是玩家 {pid} 的回合") @@ -201,6 +336,14 @@ class Simulation: raise ValueError(f"玩家 {pid} 无法行动，当前状态: {self.player_states[pid]}") action = action.lower() + + # 验证动作合法性 + is_valid, error_msg, adjusted_amount = self.validate_bet_amount(pid, action, amount or 0) + if not is_valid: + raise ValueError(error_msg) + + # 使用调整后的金额 + amount = adjusted_amount self.history.append({"pid": pid, "action": action, "amount": amount}) @@ -212,10 +355,19 @@ class Simulation: if call_amount == 0: # check self.history[-1]["action"] = "check" + self.player_states[pid] = PlayerState.CALLED else: - self.pot[pid] += call_amount - self.total_pot += call_amount - self.player_states[pid] = PlayerState.CALLED + # 检查是否all-in + actual_amount = min(call_amount, self.stacks[pid]) + if actual_amount >= self.stacks[pid]: + self.player_states[pid] = PlayerState.ALLIN + else: + self.player_states[pid] = PlayerState.CALLED + + self.pot[pid] += actual_amount + self.stacks[pid] -= actual_amount + self.total_pot += actual_amount + self.side_pot_manager.add_investment(pid, actual_amount) elif action == "check": call_amount = self.get_call_amount(pid) @@ -226,19 +378,25 @@ class Simulation: elif action in ("bet", "raise"): if amount is None: raise ValueError(f"{action} 需要指定金额") - # 加注需要补齐跟注金额 - call_amount = self.get_call_amount(pid) - total_amount = call_amount + (amount or 0) - self.pot[pid] += total_amount - self.total_pot += total_amount + # 检查是否all-in + actual_amount = min(amount, self.stacks[pid]) + if actual_amount >= self.stacks[pid]: + self.player_states[pid] = PlayerState.ALLIN + else: + self.player_states[pid] = PlayerState.CALLED - if amount and amount > 0: - self.last_raise_amount = amount - self.min_raise = amount - self.player_states[pid] = PlayerState.CALLED + self.pot[pid] += actual_amount + self.stacks[pid] -= actual_amount + self.total_pot += actual_amount + self.side_pot_manager.add_investment(pid, actual_amount) - # 其他跟注的玩家 + # 更新最后加注金额 + call_amount = self.get_call_amount(pid) + raise_amount = actual_amount - call_amount + if raise_amount > 0: + self.last_raise_amount = raise_amount + self.min_raise = raise_amount for i, state in enumerate(self.player_states): if i != pid and state == PlayerState.CALLED: self.player_states[i] = PlayerState.ACTIVE @@ -285,3 +443,128 @@ class Simulation: f.write(json.dumps(self.to_save_data())) f.write("\n") self.saved = True + + + def evaluate_player_hand(self, pid: int) -> Optional[HandRanking]: + """评估玩家手牌强度""" + if pid >= len(self.agents): + return None + + if self.player_states[pid] == PlayerState.FOLDED: + return None + + try: + # 获取玩家手牌 + player_cards = self.player_cards(pid) + + # 获取公共牌 + board_cards = self.board_cards(self.current_stage.value) + + # 至少需要5张牌才能评估 + all_cards = player_cards + board_cards + if len(all_cards) < 5: + return None + + # 如果正好5张牌，直接评估 + if len(all_cards) == 5: + return HandEvaluator.evaluate5Cards(all_cards) + + # 如果超过5张牌，找最佳组合 + return HandEvaluator.evaluateHand(all_cards) + + except Exception as e: + print(f"评估玩家 {pid} 手牌时出错: {e}") + return None + + def get_active_players(self) -> List[int]: + return [i for i, state in enumerate(self.player_states) + if state not in [PlayerState.FOLDED, PlayerState.OUT]] + + def is_hand_complete(self) -> bool: + active_players = self.get_active_players() + + if len(active_players) <= 1: + return True + + # 到达河牌且所有下注完成 + if (self.current_stage == GameStage.FINISHED or + (self.current_stage == GameStage.RIVER and self.betting_round_complete)): + return True + + return False + + def determine_winners(self) -> Dict[int, HandRanking]: + active_players = self.get_active_players() + + if not active_players: + return {} + + if len(active_players) == 1: + return {active_players[0]: None} # 不需要摊牌 + + # 多人摊牌 + hand_rankings = {} + for pid in active_players: + ranking = self.evaluate_player_hand(pid) + if ranking is not None: + hand_rankings[pid] = ranking + + return hand_rankings + + def distribute_pot(self) -> Dict[int, int]: + winners = self.determine_winners() + + if not winners: + return {} + + # 只有一人获胜（其他人弃牌） + if len(winners) == 1 and list(winners.values())[0] is None: + winner_id = list(winners.keys())[0] + return {winner_id: self.total_pot} + + # 多人摊牌，使用边池分配 + if len(winners) > 1: + #转换HandRanking为数值强度 + hand_strengths = {} + for pid, ranking in winners.items(): + if ranking is not None: + hand_strengths[pid] = ranking.get_strength() + else: + hand_strengths[pid] = 0 # 弃牌玩家 + + return self.side_pot_manager.distribute_winnings(hand_strengths) + + return {} + + def complete_hand(self) -> Dict: + if not self.is_hand_complete(): + return {"complete": False, "message": "牌局未结束"} + + winners = self.determine_winners() + + winnings = self.distribute_pot() + + # 更新筹码 + for pid, amount in winnings.items(): + if pid < len(self.stacks): + self.stacks[pid] += amount + self.current_stage = GameStage.FINISHED + + result = { + "complete": True, + "winners": list(winners.keys()), + "winnings": winnings, + "final_stacks": self.stacks.copy(), + "showdown_hands": {} + } + + # 摊牌信息 + for pid, ranking in winners.items(): + if ranking is not None: + result["showdown_hands"][pid] = { + "cards": [str(card) for card in self.player_cards(pid)], + "hand_type": ranking.hand_type.type_name, + "description": str(ranking) + } + + return result \ No newline at end of file diff --git a/shortdeck_server/__pycache__/arena_adapter.cpython-313.pyc b/shortdeck_server/__pycache__/arena_adapter.cpython-313.pyc index 6f49232..210aab3 100644 Binary files a/shortdeck_server/__pycache__/arena_adapter.cpython-313.pyc and b/shortdeck_server/__pycache__/arena_adapter.cpython-313.pyc differ diff --git a/shortdeck_server/arena_adapter.py b/shortdeck_server/arena_adapter.py index c0aa5e6..4b92947 100644 --- a/shortdeck_server/arena_adapter.py +++ b/shortdeck_server/arena_adapter.py @@ -1,12 +1,8 @@ -from __future__ import annotations - -from typing import List, Dict, Optional -from pathlib import Path - -from shortdeck_arena.simulation import Simulation -from shortdeck_arena.agent import HumanAgent -from shortdeck_arena.game_stage import BlindConfig, GameStage, PlayerState import uuid +from typing import List, Optional, Dict +from shortdeck_arena.simulation import Simulation +from shortdeck_arena.agent import Agent, HumanAgent +from shortdeck_arena.game_stage import BlindConfig class ArenaGame: @@ -18,163 +14,153 @@ class ArenaGame: self.max_players = max_players self.sim: Optional[Simulation] = None - # 筹码管理 - self.stacks: List[int] = [] - - # 盲注配置 self.blind_config = BlindConfig(small_blind, big_blind, ante=0) - - # 游戏标识 self.game_id = str(uuid.uuid4()) - - def join_game(self, name: str) -> int: - if len(self.player_names) >= self.max_players: - raise ValueError("table full") - - pid = len(self.player_names) - self.player_names.append(name) - agent = HumanAgent(pid) + + def join_game(self, name) -> int: + if len(self.agents) >= self.max_players: + raise ValueError("Game is full") + + player_id = len(self.agents) + agent = HumanAgent(player_id) self.agents.append(agent) - self.stacks.append(self.starting_stack) + self.player_names.append(name) + + if len(self.agents) == 1: + self.sim = Simulation(self.agents, blind_config=self.blind_config) + # 筹码默认1000 + self.sim.stacks = [self.starting_stack] * len(self.agents) + elif self.sim: + self.sim.agents = self.agents + self.sim.player_states.append(self.sim.player_states[0].__class__.ACTIVE) + self.sim.pot.append(0) + self.sim.stacks.append(self.starting_stack) + + # 当有至少2个玩家时，触发新一轮 + if len(self.agents) >= 2 and self.sim: + self.sim.stacks = [self.starting_stack] * len(self.agents) + self.sim.new_round() + + return player_id - self.sim = Simulation(self.agents, self.blind_config) - return pid + def apply_action(self, player_id, action, amount: Optional[int] = None) -> dict: + if not self.sim: + return {"success": False, "message": "游戏未开始"} + + try: + self.sim.apply_action(player_id, action, amount) + + self.sim.dump_data() + + return {"success": True, "message": f"Applied {action}"} + except Exception as e: + return {"success": False, "message": str(e)} def info(self, player_id: Optional[int] = None) -> Dict: - """获取游戏状态信息""" + if not self.sim: - return { - "game_id": self.game_id, - "players": self.player_names, - "stacks": [], - "dealer_index": 0, - "current_turn": 0, - "pot": 0, - "stage": "preflop", - "actions": {}, - "player_cards": [], - "board_cards": [], - } + return {"error": "游戏未初始化"} - # 更新栈大小 (扣除已投入底池的金额) - updated_stacks = [] - for i, base_stack in enumerate(self.stacks): - pot_contribution = self.sim.pot[i] if i < len(self.sim.pot) else 0 - updated_stacks.append(max(0, base_stack - pot_contribution)) - - for i in range(len(self.stacks)): - if i < len(updated_stacks): - self.stacks[i] = updated_stacks[i] - - player_cards = [] - board_cards = [] - - # 获取玩家手牌 - try: - if player_id is not None and 0 <= player_id < len(self.agents): - player_cards = [str(c) for c in self.sim.player_cards(player_id)] - except Exception: - player_cards = [] - - # 获取公共牌 (根据当前阶段) - try: - current_stage = self.sim.current_stage.value - board_cards = [str(c) for c in self.sim.board_cards(current_stage)] - except Exception: - board_cards = [] - - # 获取可用动作信息 - actions = {} - if (player_id is not None and player_id == self.sim.current_turn and - self.sim.current_stage != GameStage.FINISHED): - - call_amount = self.sim.get_call_amount(player_id) - available_stack = updated_stacks[player_id] if player_id < len(updated_stacks) else 0 - - # 更新node_info以包含实际栈信息 - node_info = self.sim.node_info() - node_info["bet_max"] = available_stack - - actions = { - "call_amount": call_amount, - "bet_min": node_info["bet_min"], - "bet_max": node_info["bet_max"], - "can_check": call_amount == 0, - "can_fold": True, - "can_call": call_amount > 0 and call_amount <= available_stack, - "can_bet": available_stack > call_amount, - } - - return { + info_data = { "game_id": self.game_id, "players": self.player_names, - "stacks": updated_stacks, - "dealer_index": 0, # 简化：固定庄家位置, （优化轮询） + "dealer_index": self.sim.dealer_position, "current_turn": self.sim.current_turn, - "pot": self.sim.total_pot, "stage": self.sim.current_stage.value, - "actions": actions, - "player_cards": player_cards, - "board_cards": board_cards, - "player_states": [state.value for state in self.sim.player_states], + "total_pot": self.sim.total_pot, + "side_pots": [{"amount": pot.amount, "eligible_players": list(pot.eligible_players)} + for pot in self.sim.get_side_pots()], } - def apply_action(self, pid: int, action: str, amount: Optional[int] = None): - if not self.sim: - raise ValueError("no game") + if player_id is not None and 0 <= player_id < len(self.sim.stacks): + try: + player_cards = self.sim.player_cards(player_id) + info_data["player_cards"] = [str(card) for card in player_cards] + except Exception: + info_data["player_cards"] = [] - # 验证动作合法性 - if pid != self.sim.current_turn: - raise ValueError(f"not your turn, current turn: {self.sim.current_turn}") - - if self.sim.current_stage == GameStage.FINISHED: - raise ValueError("game already finished") - - # 获取玩家可用筹码 - pot_contribution = self.sim.pot[pid] if pid < len(self.sim.pot) else 0 - available_stack = self.stacks[pid] - pot_contribution - - # 预处理动作和金额 - action = action.lower() - - if action in ("bet", "raise") and amount is not None: - # 限制下注金额不超过可用筹码 - if amount > available_stack: - amount = available_stack - # 如果全下，可能需要标记为allin - if amount == available_stack and available_stack > 0: - self.sim.player_states[pid] = PlayerState.ALLIN - - elif action == "call": - # call动作的金额验证 - call_amount = self.sim.get_call_amount(pid) - if call_amount > available_stack: - # 不够跟注，自动allin - amount = available_stack - if available_stack > 0: - self.sim.player_states[pid] = PlayerState.ALLIN + info_data["actions"] = self.sim.get_available_actions(player_id) + else: + info_data["player_cards"] = [] + info_data["actions"] = {"can_act": False, "reason": "Invalid player"} try: - self.sim.apply_action(pid, action, amount) - except ValueError as e: - raise ValueError(f"invalid action: {e}") - - self.sim.dump_data(Path.cwd() / "shortdeck_arena_history.jsonl") + board_cards = self.sim.board_cards(self.sim.current_stage.value) + info_data["board_cards"] = [str(card) for card in board_cards] + except Exception: + info_data["board_cards"] = [] + + info_data["stacks"] = self.sim.stacks.copy() + info_data["player_states"] = [state.value for state in self.sim.player_states] + info_data["current_pot"] = self.sim.pot.copy() + + return info_data + + + def get_hand_strength(self, player_id: int) -> Dict: + ranking = self.sim.evaluate_player_hand(player_id) + if ranking is None: + return {"error": "无法评估手牌"} + return { + "hand_type": ranking.hand_type.type_name, + "description": str(ranking), + "strength": ranking.get_strength(), + "cards": [str(card) for card in ranking.cards] + } + + def check_hand_complete(self) -> bool: + return self.sim.is_hand_complete() if self.sim else False + + def get_winners(self) -> Dict: - @property - def current_turn(self) -> int: if not self.sim: - return 0 - return self.sim.current_turn + return {"error": "游戏未初始化"} + + if not self.sim.is_hand_complete(): + return {"error": "手牌未完成"} + + return self.sim.complete_hand() + + def showdown(self) -> Dict: + if not self.sim: + return {"error": "游戏未初始化"} + + winners = self.sim.determine_winners() + showdown_info = {} + + for pid, ranking in winners.items(): + if ranking is not None: + showdown_info[pid] = { + "cards": [str(card) for card in self.sim.player_cards(pid)], + "hand_type": ranking.hand_type.type_name, + "description": str(ranking), + "strength": ranking.get_strength() + } + else: + showdown_info[pid] = { + "cards": [str(card) for card in self.sim.player_cards(pid)], + "hand_type": "Winner by default", + "description": "Other players folded", + "strength": float('inf') + } + + return { + "showdown": showdown_info, + "pot_distribution": self.sim.distribute_pot() + } @property def pot(self) -> int: - if not self.sim: - return 0 - return self.sim.total_pot - + return self.sim.total_pot if self.sim else 0 + + @property + def stacks(self) -> List[int]: + return self.sim.stacks if self.sim else [] + + @property + def current_turn(self) -> int: + return self.sim.current_turn if self.sim else -1 + @property def history(self) -> List[Dict]: - if not self.sim: - return [] - return self.sim.history + return self.sim.history if self.sim else [] \ No newline at end of file diff --git a/shortdeck_server/game_stage.py b/shortdeck_server/game_stage.py index 2010fd1..a761314 100644 --- a/shortdeck_server/game_stage.py +++ b/shortdeck_server/game_stage.py @@ -12,11 +12,11 @@ class GameStage(Enum): # # 设置首个行动玩家 # pass - def advance_street(self): - # 检查下注轮是否结束 - # 发放公共牌 - # 重置行动顺序 - pass + # def advance_street(self): + # # 检查下注轮是否结束 + # # 发放公共牌 + # # 重置行动顺序 + # pass # def get_call_amount(self, pid): # # 计算跟注所需金额 @@ -31,10 +31,10 @@ class GameStage(Enum): # 标记玩家状态 pass - def evaluate_hand(self, hole_cards, board_cards): - # 短牌手牌强度排序 - # A-6 低顺特殊处理 - pass + # def evaluate_hand(self, hole_cards, board_cards): + # # 短牌手牌强度排序 + # # A-6 低顺特殊处理 + # pass def determine_winners(self, active_players): # 边池分配 diff --git a/shortdeck_server/tests/__pycache__/test_game.cpython-313-pytest-8.4.2.pyc b/shortdeck_server/tests/__pycache__/test_game.cpython-313-pytest-8.4.2.pyc index cef96d9..8b018d6 100644 Binary files a/shortdeck_server/tests/__pycache__/test_game.cpython-313-pytest-8.4.2.pyc and b/shortdeck_server/tests/__pycache__/test_game.cpython-313-pytest-8.4.2.pyc differ diff --git a/shortdeck_server/tests/test_game.py b/shortdeck_server/tests/test_game.py index 6281683..14d5603 100644 --- a/shortdeck_server/tests/test_game.py +++ b/shortdeck_server/tests/test_game.py @@ -19,10 +19,9 @@ def test_join_and_actions(): assert g.current_turn == 0 # 测试错误的玩家尝试行动 - try: - g.apply_action(1, "fold") - except ValueError as e: - assert "not your turn" in str(e) + result = g.apply_action(1, "fold") + assert result["success"] == False + assert "不是玩家" in result["message"] or "turn" in result["message"].lower() # 小盲玩家call (跟注到大盲) g.apply_action(0, "call") diff --git a/shortdeck_server/tools/__pycache__/run_smoke.cpython-313.pyc b/shortdeck_server/tools/__pycache__/run_smoke.cpython-313.pyc deleted file mode 100644 index 605b6cf..0000000 Binary files a/shortdeck_server/tools/__pycache__/run_smoke.cpython-313.pyc and /dev/null differ diff --git a/shortdeck_server/tools/run_smoke.py b/shortdeck_server/tools/run_smoke.py deleted file mode 100644 index dc46555..0000000 --- a/shortdeck_server/tools/run_smoke.py +++ /dev/null @@ -1,36 +0,0 @@ -from __future__ import annotations - -import json -from fastapi.testclient import TestClient - -from shortdeck_server.main import app - -client = TestClient(app) - -def pretty(o): - print(json.dumps(o, ensure_ascii=False, indent=2)) - - -def run(): - print('POST /join aa') - r = client.post('/join', json={'name':'aa'}) - pretty(r.json()) - - print('POST /join bb') - r = client.post('/join', json={'name':'bb'}) - pretty(r.json()) - - print('GET /get_game_state?player_id=0') - r = client.get('/get_game_state', params={'player_id':0}) - pretty(r.json()) - - print('POST /apply_action check by pid=0') - r = client.post('/apply_action', json={'player_id':0, 'action':'check'}) - pretty(r.json()) - - print('POST /apply_action bet 10 by pid=1') - r = client.post('/apply_action', json={'player_id':1, 'action':'bet', 'amount':10}) - pretty(r.json()) - -if __name__ == '__main__': - run()