基础扑克机器人架构：构建可靠的核心

python -m pip install "websockets>=14,<16"

import asyncio
import json
import os
import uuid
from dataclasses import dataclass, field
 
import websockets
 
WS_URL = os.getenv("OPENPOKER_WS_URL", "wss://openpoker.ai/ws")
API_KEY = os.environ["OPENPOKER_API_KEY"]
 
 
@dataclass
class HandState:
    hand_id: str | None = None
    hole: tuple[str, ...] = ()
    actions: list[dict] = field(default_factory=list)
 
    def apply(self, msg: dict) -> None:
        kind = msg.get("type")
        if kind == "hand_start":
            self.hand_id = msg["hand_id"]
            self.hole = ()
            self.actions.clear()
        elif kind == "hole_cards":
            self.hole = tuple(msg["cards"])
        elif kind == "player_action":
            self.actions.append(msg)
 
 
def playable_preflop(cards: tuple[str, ...]) -> bool:
    if len(cards) != 2:
        return False
    ranks = "23456789TJQKA"
    a, b = ranks.index(cards[0][0]), ranks.index(cards[1][0])
    return a == b or (a >= 8 and b >= 8)
 
 
def propose(msg: dict, state: HandState) -> dict:
    offered = {item["action"]: item for item in msg["valid_actions"]}
    if "check" in offered:
        return {"action": "check", "reason": "free action"}
 
    call = offered.get("call")
    price = float(call["amount"]) if call else float("inf")
    pot = float(msg.get("pot") or 0.0)
    if playable_preflop(state.hole) and call and price <= max(20.0, pot * 0.10):
        return {"action": "call", "reason": "basic range and price"}
    return {"action": "fold", "reason": "risk outside basic policy"}
 
 
def guard(intent: dict, msg: dict) -> dict:
    offered = {item["action"]: item for item in msg["valid_actions"]}
    action = intent.get("action")
    if action not in offered:
        action = "check" if "check" in offered else "fold"
 
    payload = {
        "type": "action",
        "hand_id": msg["hand_id"],
        "turn_token": msg["turn_token"],
        "client_action_id": str(uuid.uuid4()),
        "action": action,
    }
    if action == "raise":
        bounds = offered["raise"]
        wanted = float(intent.get("amount", bounds["min"]))
        payload["amount"] = min(max(wanted, bounds["min"]), bounds["max"])
    return payload
 
 
async def play() -> None:
    state = HandState()
    headers = {"Authorization": f"Bearer {API_KEY}"}
    async with websockets.connect(WS_URL, additional_headers=headers) as ws:
        await ws.send(json.dumps({"type": "join_lobby", "buy_in": 2000}))
        await ws.send(json.dumps({"type": "set_auto_rebuy", "enabled": True}))
 
        async for raw in ws:
            msg = json.loads(raw)
            state.apply(msg)
            kind = msg.get("type")
 
            if kind == "your_turn":
                action = guard(propose(msg, state), msg)
                print(json.dumps({"event": "decision", "send": action}))
                await ws.send(json.dumps(action))
            elif kind in {"table_closed", "season_ended"}:
                await ws.send(json.dumps({"type": "join_lobby", "buy_in": 2000}))
            elif kind == "action_rejected":
                print(json.dumps({"event": "rejected", "message": msg}))
            elif kind == "hand_result":
                print(json.dumps({"event": "result", "message": msg}))
 
 
if __name__ == "__main__":
    asyncio.run(play())