haiyingjiang7/gametree
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

gametree: fix player

Browse Source
This commit is contained in:
haiyingjiang7@1price.io
2025-11-06 17:41:59 +08:00
parent 61923407af
commit b4afdb06c4
7 changed files with 236 additions and 171 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
gametree/__init__.py
View File

@@ -1,6 +1,5 @@
from .model import (
PlayerId,
Player,
Action,
ActionType,
Street,
@@ -18,7 +17,6 @@ from .hand_evaluator import HandEvaluator
__all__ = [
"PlayerId",
"Player",
"Action",
"ActionType",
"Street",

3
gametree/deck.py
View File

@@ -52,3 +52,6 @@ class DeckManager:
burn = self.burn_card()
river = self.deal_card()
return burn, river
# def get_remaining_cards(self) -> List[Card]:
# return self.cards[self.position:]

237
gametree/game.py
View File

@@ -16,7 +16,9 @@ class PlayerState:
committed: int = 0 # 本轮已投入筹码
in_hand: bool = True # 玩家是否仍在本局中
folded: bool = False # 玩家是否已弃牌
all_in: bool = False # 玩家是否已全押
allin: bool = False # 玩家是否已全押
hand_cards: List[Card] = field(default_factory=list)
hand_rankings: Optional[Any] = None
def to_dict(self) -> Dict[str, Any]:
return {
@@ -25,7 +27,8 @@ class PlayerState:
"committed": self.committed,
"in_hand": self.in_hand,
"folded": self.folded,
"all_in": self.all_in,
"allin": self.allin,
"hand_cards": [str(c) for c in self.hand_cards],
}
@@ -65,13 +68,12 @@ class Pot:
class Game:
players_init: List[Tuple[PlayerId, int]] = field(default_factory=list)
dealer_idx: int = 0
bb_idx: int = 0
sb_idx: int = 0
_bb_idx: Optional[int] = field(default=None, init=False)
_sb_idx: Optional[int] = field(default=None, init=False)
small_blind: int = 1
big_blind: int = 2
current_street: Street = Street.PREFLOP
hand_cards: Dict[PlayerId, List[Card]] = field(default_factory=dict)
board: List[Card] = field(default_factory=list)
players: List[PlayerState] = field(default_factory=list)
@@ -85,10 +87,9 @@ class Game:
next_to_act_idx: Optional[int] = None
terminal: bool = False
winner: Optional[List[PlayerId]] = None
pot: List[Pot] = field(default_factory=list)
hand_rankings: Dict[PlayerId, Any] = field(default_factory=dict)
# hand_rankings: Dict[PlayerId, Any] = field(default_factory=dict)
deck: List[Card] = field(default_factory=list) # 剩余牌
burned_cards: List[Card] = field(default_factory=list)
@@ -116,7 +117,9 @@ class Game:
def _init_hand_cards(self):
player_ids = [pid for pid, _ in self.players_init]
self.hand_cards = self.deck_manager.deal_hand_cards(player_ids)
deal_card = self.deck_manager.deal_hand_cards(player_ids)
for p in self.players:
p.hand_cards = deal_card.get(p.pid, [])
def _init_street_actions(self):
self.street_actions = {
@@ -130,7 +133,7 @@ class Game:
self.deck = Card.all_short().copy()
random.shuffle(self.deck)
# 移除已发的底牌
used_cards = [card for hand in self.hand_cards.values() for card in hand]
used_cards = [card for hand in self.players for card in hand.hand_cards]
self.deck = [card for card in self.deck if card not in used_cards]
def _init_bet(self):
@@ -139,18 +142,18 @@ class Game:
return
if n_players == 2:
self.sb_idx = self.dealer_idx # 两人游戏：庄家是小盲
self.bb_idx = (self.dealer_idx + 1) % n_players
self._sb_idx = self.dealer_idx # 两人游戏：庄家是小盲
self._bb_idx = (self.dealer_idx + 1) % n_players
else:
self.sb_idx = (self.dealer_idx + 1) % n_players # 小盲位置 = 庄家下一位
self.bb_idx = (self.dealer_idx + 2) % n_players # 大盲位置 = 小盲下一位
self._sb_idx = (self.dealer_idx + 1) % n_players # 小盲位置 = 庄家下一位
self._bb_idx = (self.dealer_idx + 2) % n_players # 大盲位置 = 小盲下一位
# 小盲注
self.players[self.sb_idx].stack -= self.small_blind
self.players[self.sb_idx].committed += self.small_blind
self.players[self._sb_idx].stack -= self.small_blind
self.players[self._sb_idx].committed += self.small_blind
self.players[self.bb_idx].stack -= self.big_blind
self.players[self.bb_idx].committed += self.big_blind
self.players[self._bb_idx].stack -= self.big_blind
self.players[self._bb_idx].committed += self.big_blind
self.current_bet = self.big_blind
self.original_bet = self.big_blind
@@ -170,14 +173,14 @@ class Game:
return None
if self.current_street == Street.PREFLOP:
first_idx = (self.bb_idx + 1) % n_players if n_players >= 3 else self.sb_idx
first_idx = (self._bb_idx + 1) % n_players if n_players >= 3 else self._sb_idx
self.next_to_act_idx = first_idx
return first_idx
else:
for i in range(0, n_players):
idx = (self.sb_idx + i) % n_players
idx = (self._sb_idx + i) % n_players
player = self.players[idx]
if player.in_hand and not player.folded and not player.all_in:
if player.in_hand and not player.folded and not player.allin:
self.next_to_act_idx = idx
return idx
@@ -186,7 +189,7 @@ class Game:
return None
def add_action(self, action: Action):
def add_action(self, action: Action) -> Optional[List[PlayerId]]:
if self.terminal:
raise ValueError("Game is over")
@@ -203,7 +206,8 @@ class Game:
if not self.terminal:
if self.next_to_act_idx is None or self._is_round_over():
if self.current_street == Street.RIVER:
self.finalize_game()
winners = self.finalize_game()
return winners
else:
self.advance_to_next_street()
# all allin
@@ -213,8 +217,9 @@ class Game:
if (not self.terminal and self.current_street == Street.RIVER and
(self.next_to_act_idx is None or self._all_active_all_in())):
self.finalize_game()
winners = self.finalize_game()
return winners
return None
def _all_active_all_in(self) -> bool:
active = [p for p in self.players if p.in_hand and not p.folded]
@@ -222,13 +227,13 @@ class Game:
return True
if len(active) <= 1:
return True
return all(p.all_in for p in active)
return all(p.allin for p in active)
def finalize_game(self):
def finalize_game(self) -> Optional[List[PlayerId]]:
print("+++++++++++++++++++++++++++++++++++++++++++++++++++++++++")
print("DEBUG: finalize_game")
print(f"DEBUG: pot before finalize: {[(p.amount, len(p.eligible)) for p in self.pot]}")
print(f"DEBUG: active players: {[(p.pid, p.stack, p.committed, p.folded, p.all_in)
print(f"DEBUG: active players: {[(p.pid, p.stack, p.committed, p.folded, p.allin)
for p in self.players if p.in_hand]}")
if not self.pot:
self.finalize_pots()
@@ -241,19 +246,28 @@ class Game:
for p in self.players:
p.committed = 0
self.pot = []
self.winner = [win_pid]
self.terminal = True
return
return [win_pid]
winners = self.cal_winners()
self._distribute_pot()
winners_all = self.showdown()
if winners_all:
self.winner = winners_all
for p in self.players:
p.committed = 0
p.allin = False
self.pot = []
self.terminal = True
return winners
def _distribute_pot(self):
for pot in self.pot:
elig = [pid for pid in pot.eligible if pid in self.hand_rankings]
elig = [self._get_player(pid) for pid in pot.eligible]
elig = [p for p in elig if p and p.in_hand and not p.folded
and p.hand_rankings is not None]
if not elig:
continue
best_strength = max(self.hand_rankings[pid].get_strength() for pid in elig)
pot_winners = [pid for pid in elig if self.hand_rankings[pid].get_strength() == best_strength]
best_strength = max(p.hand_rankings.get_strength() for p in elig)
pot_winners = [p.pid for p in elig if
p.hand_rankings.get_strength() == best_strength]
share = pot.amount // len(pot_winners)
rem = pot.amount % len(pot_winners)
for i, pid in enumerate(pot_winners):
@@ -262,13 +276,28 @@ class Game:
continue
ps.stack += share + (rem if i == 0 else 0)
def _compute_hand_rankings(self) -> Dict[PlayerId, Any]:
p_rank: Dict[PlayerId, Any] = {}
for p in self.players:
p.committed = 0
p.all_in = False
self.pot = []
self.terminal = True
return
if p.in_hand and not p.folded:
ranking = self.evaluate_hand(p.pid)
if ranking:
p.hand_rankings = ranking
p_rank[p.pid] = ranking
return p_rank
def cal_winners(self) -> List[PlayerId]:
if len(self.board) != 5:
p_active = [p for p in self.players if p.in_hand and not p.folded]
if len(p_active) == 1:
return [p_active[0].pid]
return []
p_rank = self._compute_hand_rankings()
if not p_rank:
return []
best_r = max(r.get_strength() for r in p_rank.values())
winners = [pid for pid, r in p_rank.items() if r.get_strength() == best_r]
return winners
def _is_round_over(self) -> bool:
active = [p for p in self.players if p.in_hand and not p.folded]
@@ -277,9 +306,9 @@ class Game:
if self.current_bet == 0 and self.next_to_act_idx is None:
return True
for p in active:
if not p.all_in and p.committed < self.current_bet:
if not p.allin and p.committed < self.current_bet:
return False
if not p.all_in and p.committed!= self.current_bet:
if not p.allin and p.committed!= self.current_bet:
return False
return True
@@ -287,10 +316,10 @@ class Game:
player.folded = True
player.in_hand = False
# 检查是否只剩一个玩家，如果是的话，游戏结束，进行结算 todo
active_players = [p for p in self.players if p.in_hand and not p.folded]
if len(active_players) == 1:
self.terminal = True
self.winner = [active_players[0].pid]
# active_players = [p for p in self.players if p.in_hand and not p.folded]
# if len(active_players) == 1:
# self.terminal = True
# self.winner = [active_players[0].pid]
def _player_call_act(self, player: PlayerState, amount: Optional[int] = None):
if not player.in_hand or player.folded:
@@ -301,7 +330,7 @@ class Game:
player.stack -= pay
player.committed += pay
if player.stack == 0:
player.all_in = True
player.allin = True
def _player_check_act(self, player: PlayerState):
if not player.in_hand or player.folded:
@@ -310,7 +339,7 @@ class Game:
if to_call > 0:
raise ValueError(f"Cannot CHECK,need {to_call} to call")
if player.stack == 0:
player.all_in = True
player.allin = True
def _player_bet_act(self, player: PlayerState, amount: int):
if amount is None:
@@ -329,7 +358,7 @@ class Game:
player.committed += pay
if player.stack == 0:
player.all_in = True
player.allin = True
self.current_bet = player.committed
self.original_bet = player.committed
@@ -356,7 +385,7 @@ class Game:
player.committed += pay
if player.stack == 0:
player.all_in = True
player.allin = True
# 有效加注
if player.committed > self.current_bet:
@@ -368,13 +397,13 @@ class Game:
raise ValueError("Player cannot act")
if player.stack <= 0:
raise ValueError("Player no chips to all-in")
if player.all_in:
if player.allin:
raise ValueError("Player already all-in")
pay = player.stack
player.stack = 0
player.committed += pay
player.all_in = True
player.allin = True
if self.current_bet == 0: # 下注即allin
self.current_bet = player.committed
@@ -405,7 +434,7 @@ class Game:
self._player_bet_act(player, action.amount)
case ActionType.RAISE:
self._player_raise_act(player, action.amount)
case ActionType.ALL_IN:
case ActionType.ALLIN:
self._player_all_in_act(player)
def _update_next_to_act(self):
@@ -421,16 +450,16 @@ class Game:
for i in range(1, n_players + 1):
next_idx = (current_idx + i) % n_players
player = self.players[next_idx]
if player.in_hand and not player.folded and not player.all_in and player.stack > 0:
if player.in_hand and not player.folded and not player.allin and player.stack > 0:
self.next_to_act_idx = next_idx
return
# 没有找到可行动的玩家，进入下一轮 todo
self.next_to_act_idx = None
def advance_to_next_street(self):
def advance_to_next_street(self) -> Street:
if self.current_street == Street.RIVER:
# 已经是最后一街道，准备摊牌 todo
return
return self.current_street
round_pot = sum(player.committed for player in self.players)
if round_pot > 0:
@@ -448,8 +477,8 @@ class Game:
self.invalid_raise = False
self._deal_cards_for_street()
self._get_first_act_idx()
return self.current_street
def _deal_cards_for_street(self):
if self.current_street == Street.PREFLOP:
@@ -509,7 +538,8 @@ class Game:
return (min_raise_total, max_raise_total)
def get_hand_cards(self, player_id: PlayerId) -> List[Card]:
return self.hand_cards.get(player_id, [])
p = self._get_player(player_id)
return p.hand_cards if p else []
def get_current_board(self) -> List[Card]:
return self.board.copy()
@@ -561,21 +591,21 @@ class Game:
if self.invalid_raise:
if to_call == 0:
if player.stack > 0:
actions.append(ActionType.ALL_IN)
actions.append(ActionType.ALLIN)
else:
actions.append(ActionType.FOLD)
if player.stack > 0:
if player.stack >= to_call: # 暂时这么处理吧
actions.append(ActionType.CALL)
else:
actions.append(ActionType.ALL_IN)
actions.append(ActionType.ALLIN)
return actions
if to_call == 0:
actions.append(ActionType.CHECK)
if player.stack > 0:
actions.append(ActionType.BET)
actions.append(ActionType.ALL_IN)
actions.append(ActionType.ALLIN)
else:
actions.append(ActionType.FOLD)
if player.stack > 0:
@@ -585,52 +615,47 @@ class Game:
if max_raise_to < min_raise_to:
if player.stack >= to_call:
actions.append(ActionType.CALL)
actions.append(ActionType.ALL_IN)
actions.append(ActionType.ALLIN)
else:
actions.append(ActionType.ALL_IN)
actions.append(ActionType.ALLIN)
self.invalid_raise = True
else:
actions.append(ActionType.CALL)
actions.append(ActionType.RAISE)
actions.append(ActionType.ALL_IN)
actions.append(ActionType.ALLIN)
return actions
def evaluate_hand(self, player_id: PlayerId) -> Optional[Any]:
if player_id not in self.hand_cards or len(self.board) < 5:
p = self._get_player(player_id)
if not p or len(self.board) < 5:
return None
all_cards = self.hand_cards[player_id] + self.board
all_cards = p.hand_cards + self.board
if len(all_cards) != 7:
return None
return HandEvaluator.evaluateHand(all_cards)
def showdown(self) -> List[PlayerId]:
if len(self.board) != 5:
active_players = [p for p in self.players if p.in_hand and not p.folded]
if len(active_players) == 1:
self.winner = [active_players[0].pid]
self.terminal = True
return self.winner
return [active_players[0].pid]
return []
player_rankings = {}
in_rank = False
for player in self.players:
if player.in_hand and not player.folded:
ranking = self.evaluate_hand(player.pid)
if ranking:
player_rankings[player.pid] = ranking
if not player_rankings:
in_rank = True
player.hand_rankings = ranking
if not in_rank:
return []
best_strength = max(ranking.get_strength() for ranking in player_rankings.values())
winners = [pid for pid, ranking in player_rankings.items()
if ranking.get_strength() == best_strength]
self.hand_rankings = player_rankings
self.winner = winners
best_rank = max(p.hand_rankings.get_strength() for p in self.players
if p.in_hand and not None)
winners = [p.pid for p in self.players
if p.in_hand and p.hand_rankings is not None and
p.hand_rankings.get_strength() == best_rank]
self.terminal = True
return winners
@@ -674,10 +699,52 @@ class Game:
"board": [str(card) for card in self.board],
"current_bet": self.current_bet,
"pot": [pot.to_dict() for pot in self.pot],
"hand_cards": {str(pid): [str(card) for card in cards]
for pid, cards in self.hand_cards.items()},
"next_to_act_idx": self.next_to_act_idx,
"terminal": self.terminal,
"winner": [w.to_dict() for w in self.winner] if self.winner else None,
"winner": [w.to_dict() for w in (self.get_winners() or [])] if self.get_winners() else None,
"total_actions": len(self._all_actions),
}
# 测试
@property
def get_sb_idx(self) -> Optional[int]:
return self._sb_idx
@property
def get_bb_idx(self) -> Optional[int]:
return self._bb_idx
@property
def get_active_players(self) -> List[PlayerState]:
return [p for p in self.players if p.in_hand and not p.folded]
@property
def get_sb_amt(self) -> int:
return self.small_blind if self._sb_idx is not None else 0
@property
def get_bb_amt(self) -> int:
return self.big_blind if self._bb_idx is not None else 0
@property
def get_dealer_idx(self) -> int:
return self.dealer_idx
@property
def get_winners(self) -> Optional[List[PlayerId]]:
if not self.terminal:
return None
if self.hand_rankings:
best_strength = max(r.get_strength() for r in self.hand_rankings.values())
winners = [pid for pid, r in self.hand_rankings.items()
if r.get_strength() == best_strength]
return winners
return None
@property
def get_next_act_idx(self) -> Optional[int]:
return self.next_to_act_idx
@property
def get_deck(self) -> Optional[DeckManager]:
return self.deck_manager

11
gametree/model.py
View File

@@ -29,13 +29,6 @@ class PlayerId:
def from_dict(cls, d: Dict[str, Any]) -> "PlayerId":
return cls(int(d["id"]), d.get("name"))
@dataclass
class Player:
pid: PlayerId
name: str
# seat: int
# address: str
class Street(Enum):
PREFLOP = 0
FLOP = 1
@@ -48,7 +41,7 @@ class ActionType(Enum):
CHECK = "check"
RAISE = "raise"
BET = "bet"
ALL_IN = "all_in"
ALLIN = "allin"
@dataclass
@@ -103,5 +96,5 @@ def act_raise(pid: PlayerId, to_amount: int) -> Action:
return Action(actor=pid, type=ActionType.RAISE, amount=to_amount)
def act_all_in(pid: PlayerId, amount: int) -> Action:
return Action(actor=pid, type=ActionType.ALL_IN, amount=amount)
return Action(actor=pid, type=ActionType.ALLIN, amount=amount)

67
test/p_creat.py Normal file
View File

@@ -0,0 +1,67 @@
import sys
from pathlib import Path
import pytest
from gametree import Game, PlayerId, Street, ActionType
def make_players(n, stack):
return [(PlayerId(i, f"p{i}"), stack) for i in range(n)]
def test_player_states():
sb = 5
bb = 10
d_idx = 1
n_player = 4
stack = 500
players_init = make_players(n_player, stack)
g = Game(players_init=players_init, small_blind=sb, big_blind=bb, dealer_idx=d_idx)
assert g.current_street == Street.PREFLOP
sb_idx = (d_idx + 1) % n_player
bb_idx = (d_idx + 2) % n_player
next_idx = (d_idx + 3) % n_player
# init_bet
assert g.get_sb_amt == sb
assert g.get_bb_amt == bb
assert g.get_sb_idx == sb_idx
assert g.get_bb_idx == bb_idx
assert g.get_next_act_idx == next_idx
# init_players
players = g.get_active_players
assert len(players) == n_player
for idx, p in enumerate(players):
assert p.hand_cards is not None and len(p.hand_cards) == 2
if idx == sb_idx:
assert p.stack == stack - sb
assert p.committed == sb
elif idx == bb_idx:
assert p.stack == stack - bb
assert p.committed == bb
else:
assert p.stack == stack
assert p.committed == 0
# init_deck
deck = getattr(g, 'deck_manager', None)
rem_cards = getattr(deck, 'remaining_cards', None)
assert rem_cards is not None
assert rem_cards() == 28 # 36-8
# first_player
legal_actions = g.legal_actions(g.players[next_idx].pid)
assert legal_actions is not None and len(legal_actions) == 4
assert set(legal_actions) == {ActionType.FOLD, ActionType.CALL,
ActionType.RAISE, ActionType.ALLIN}
# n_players = len(players_init)
# if n_players < 2:
# expected_sb = None
# expected_bb = None
# elif n_players == 2:
# expected_sb = d_idx
# expected_bb = (d_idx + 1) % n_players
# else:
# expected_sb = (d_idx + 1) % len(players_init)
# expected_bb = (d_idx + 2) % len(players_init)
# assert g.get_sb_idx == expected_sb
# assert g.get_bb_idx == expected_bb

64
test/pg.json
View File

@@ -1,64 +0,0 @@
{
"player_names": {
"0": "a",
"1": "b",
"2": "c",
"3": "d",
"4": "e"
},
"game_state": {
"players_init": [
[
0,
500
],
[
1,
500
],
[
2,
500
],
[
3,
500
],
[
4,
500
]
],
"dealer_idx": 0,
"small_blind": 5,
"big_blind": 10,
"current_street": "PREFLOP",
"all_actions": [
{
"type": "CALL",
"actor": 3,
"amount": null
},
{
"type": "CALL",
"actor": 4,
"amount": null
},
{
"type": "CALL",
"actor": 0,
"amount": null
},
{
"type": "ALL_IN",
"actor": 1,
"amount": null
},
{
"type": "ALL_IN",
"actor": 2,
"amount": null
}
]
}
}

13
test/pg.py
View File

@@ -15,6 +15,7 @@ from gametree.model import act_fold, act_call, act_check, act_bet, act_raise, ac
GAME_FILE = "pg.json"
def create_file_idx():
Path("data").mkdir(exist_ok=True)
files = list(Path("data").glob("pg_*.json"))
game_id = len(files) + 1
return f"data/pg_{game_id:03d}.json"
@@ -64,7 +65,7 @@ def display_game_status(game: Game, player_names: Dict[int,str], show_cards_for:
marks.append("->")
if pstate.folded:
marks.append("F")
if pstate.all_in:
if pstate.allin:
marks.append("A")
info = f"{i}:{name} stack={pstate.stack} bet={pstate.committed} [{' '.join(marks)}]"
if show_cards_for == "all" or show_cards_for == name:
@@ -144,8 +145,8 @@ def display_player_turn():
elif a == ActionType.RAISE:
min_raise, max_raise = GAME.get_raise_bounds(pid)
print(f" - raise (min: {min_raise}, max: {max_raise})")
elif a == ActionType.ALL_IN:
print(f" - all_in (amount: {GAME.get_allin_amt(pid)})")
elif a == ActionType.ALLIN:
print(f" - allin (amount: {GAME.get_allin_amt(pid)})")
print("----------------------------------------------------")
@@ -200,7 +201,7 @@ def main():
return
input = shlex.split(args)
if len(input) < 2:
print("act <player_name> <fold|call|check|bet|raise|all_in> [amount]")
print("act <player_name> <fold|call|check|bet|raise|allin> [amount]")
return
pname = input[0]
action = input[1]
@@ -213,7 +214,7 @@ def main():
legal = GAME.legal_actions(pid)
if action.lower() == 'allin':
atype = ActionType.ALL_IN
atype = ActionType.ALLIN
elif action.lower() in ('fold', 'call', 'check', 'bet', 'raise'):
atype = ActionType[action.upper()]
else:
@@ -273,7 +274,7 @@ def main():
def cmd_help(args):
print("可用命令：")
print(" set <small>/<big> <p1> [p2 ...] [--stack N] ")
print(" act <player> <fold|call|check|bet|raise|all_in> [amount]")
print(" act <player> <fold|call|check|bet|raise|allin> [amount]")
print(" status [player|all] ")
print(" save ")
print(" reset ")