Import source: Shayan1/Prod/nash/players

Import the existing local project snapshot. This commit organizes existing files; it does not represent newly developed functionality.

Shayan1/Prod/nash/players.py

@@ -0,0 +1,391 @@
+"""Player decision models for the simulation.
+
+Every player exposes a uniform interface:
+
+    choose(history) -> int          # strategy index for the next round
+    update(profile, payoffs)        # optional learning hook (no-op by default)
+
+`BestResponsePlayer` is the original model (best response / fictitious play). The
+additional models -- regret matching, softmax best response, and Q-learning -- are
+learning agents that implement `update` so they improve over the course of a run.
+All randomness flows through an injected `random.Random` instance so runs are
+reproducible given a fixed seed.
+"""
+
+import itertools
+import json
+import math
+import random
+from abc import ABC, abstractmethod
+from collections import Counter
+
+import numpy as np
+
+
+class BasePlayer(ABC):
+    """Common interface and shared helpers for all player models."""
+
+    def __init__(self, game, player_index, rng=None, debug_logger=None):
+        self.game = game
+        self.player_index = player_index
+        self.rng = rng or random.Random()
+        self.debug_logger = debug_logger
+
+    def _log(self, message):
+        """Emit a debug message if a logger is attached."""
+        if self.debug_logger is not None:
+            self.debug_logger(self.player_index, message)
+
+    @abstractmethod
+    def choose(self, history):
+        """Return this player's strategy index for the next round."""
+        raise NotImplementedError
+
+    def update(self, round_profile, round_payoffs):
+        """Learning hook, called after each round. Default: stateless (no-op)."""
+        return None
+
+    # --- shared geometry helpers (depend only on self.player_index) ---
+
+    def _opponents_of(self, profile):
+        """Tuple of everyone else's strategy index, in player order."""
+        return tuple(s for idx, s in enumerate(profile) if idx != self.player_index)
+
+    def _build_profile(self, my_strategy, opponent_profile):
+        """Re-insert this player's strategy into an opponent-only profile."""
+        profile = list(opponent_profile)
+        profile.insert(self.player_index, my_strategy)
+        return tuple(profile)
+
+
+class BestResponsePlayer(BasePlayer):
+    """Best response to opponents' play.
+
+    memory=False: best response to only the immediately preceding round (classic
+    best-response dynamics -- can cycle indefinitely).
+    memory=True:  best response to the empirical distribution of opponents' entire
+    history (fictitious play -- converges more reliably).
+    Ties among equally good strategies are broken via the injected RNG.
+    """
+
+    def __init__(self, game, player_index, memory=True, rng=None, debug_logger=None):
+        super().__init__(game, player_index, rng=rng, debug_logger=debug_logger)
+        self.memory = memory
+
+    def choose(self, history):
+        n = self.game.strategy_counts()[self.player_index]
+        if not history:
+            return self.rng.randrange(n)
+
+        relevant_history = history if self.memory else [history[-1]]
+        opponent_profiles = [self._opponents_of(p) for p in relevant_history]
+        return self._best_response(opponent_profiles)
+
+    def _best_response(self, opponent_profiles, tol=1e-9):
+        counts = Counter(opponent_profiles)
+        total = len(opponent_profiles)
+        n = self.game.strategy_counts()[self.player_index]
+
+        expected_payoffs = []
+        for my_strategy in range(n):
+            expected = sum(
+                (count / total)
+                * self.game.payoff(
+                    self.player_index, self._build_profile(my_strategy, opponents)
+                )
+                for opponents, count in counts.items()
+            )
+            expected_payoffs.append(expected)
+
+        self._log(f"expected payoffs: {expected_payoffs}")
+
+        best_value = max(expected_payoffs)
+        best_strategies = [
+            s for s, v in enumerate(expected_payoffs) if v >= best_value - tol
+        ]
+        return self.rng.choice(best_strategies)
+
+
+class SoftmaxBestResponsePlayer(BasePlayer):
+    """Best response with Boltzmann (softmax) exploration over expected payoff.
+
+    Computes the same expected-payoff-vs-opponents signal as BestResponsePlayer
+    (honoring the memory switch) but samples an action from a softmax rather than
+    taking the argmax. `temperature` controls exploration: high T -> near uniform,
+    low T -> near greedy. `temperature <= 0` falls back to greedy tie-breaking.
+    """
+
+    def __init__(self, game, player_index, memory=True, temperature=1.0, rng=None, debug_logger=None):
+        super().__init__(game, player_index, rng=rng, debug_logger=debug_logger)
+        self.memory = memory
+        self.temperature = float(temperature)
+
+    def choose(self, history):
+        n = self.game.strategy_counts()[self.player_index]
+        if not history:
+            return self.rng.randrange(n)
+
+        relevant_history = history if self.memory else [history[-1]]
+        counts = Counter(self._opponents_of(p) for p in relevant_history)
+        total = len(relevant_history)
+
+        expected = [
+            sum(
+                (c / total)
+                * self.game.payoff(
+                    self.player_index, self._build_profile(a, opp)
+                )
+                for opp, c in counts.items()
+            )
+            for a in range(n)
+        ]
+
+        self._log(f"expected payoffs: {expected}")
+
+        if self.temperature <= 0:
+            best_value = max(expected)
+            best_strategies = [
+                i for i, v in enumerate(expected) if v >= best_value - 1e-9
+            ]
+            return self.rng.choice(best_strategies)
+
+        # Numerical stability: shift by max before exponentiating.
+        m = max(expected)
+        weights = [math.exp((e - m) / self.temperature) for e in expected]
+        return int(self.rng.choices(range(n), weights=weights)[0])
+
+
+class RegretMatchingPlayer(BasePlayer):
+    """Hart & Mas-Colell regret matching (no-external-regret learning).
+
+    Each round the player plays proportionally to its cumulative positive regret
+    against its own actions, given the opponents' realized actions. This is a
+    no-regret dynamics: the empirical joint distribution of play converges to a
+    *coarse correlated equilibrium* (which contains every correlated equilibrium).
+    Works for any number of players and general-sum games.
+    """
+
+    def __init__(self, game, player_index, rng=None, debug_logger=None):
+        super().__init__(game, player_index, rng=rng, debug_logger=debug_logger)
+        self.n = self.game.strategy_counts()[self.player_index]
+        self.cumulative_regret = np.zeros(self.n, dtype=float)
+
+    def choose(self, history):
+        pos = np.clip(self.cumulative_regret, 0, None)
+        total = float(pos.sum())
+        if total <= 0:
+            probs = [1.0 / self.n] * self.n
+        else:
+            probs = (pos / total).tolist()
+
+        self._log(f"cumulative regret: {self.cumulative_regret.tolist()}, probabilities: {probs}")
+
+        return int(self.rng.choices(range(self.n), weights=probs)[0])
+
+    def update(self, round_profile, round_payoffs):
+        my_action = round_profile[self.player_index]
+        opponents = self._opponents_of(round_profile)
+        realized = round_payoffs[self.player_index]
+        for a in range(self.n):
+            payoff_if_a = self.game.payoff(
+                self.player_index, self._build_profile(a, opponents)
+            )
+            self.cumulative_regret[a] += payoff_if_a - realized
+
+
+class QLearningPlayer(BasePlayer):
+    """Independent Q-learning with epsilon-greedy exploration.
+
+    State = opponents' realized joint action; action = this player's strategy.
+    Standard tabular Q-update with learning rate `alpha` and discount `gamma`.
+    Each player learns independently, so this models decentralized adaptive play
+    rather than equilibrium solving.
+    """
+
+    def __init__(
+        self,
+        game,
+        player_index,
+        alpha=0.1,
+        gamma=0.9,
+        epsilon=0.1,
+        rng=None,
+        debug_logger=None,
+    ):
+        super().__init__(game, player_index, rng=rng, debug_logger=debug_logger)
+        self.alpha = float(alpha)
+        self.gamma = float(gamma)
+        self.epsilon = float(epsilon)
+        self.n = self.game.strategy_counts()[self.player_index]
+        self.q = {}  # (state, action) -> float
+        self.last_state = None
+
+    def _state_from(self, profile):
+        return self._opponents_of(profile) if profile is not None else ()
+
+    def _q(self, state, action):
+        return self.q.get((state, action), 0.0)
+
+    def choose(self, history):
+        state = self._state_from(history[-1]) if history else ()
+        self.last_state = state
+        values = [self._q(state, a) for a in range(self.n)]
+        self._log(f"Q-values for state {state}: {values}")
+        if self.rng.random() < self.epsilon:
+            return self.rng.randrange(self.n)
+        best_value = max(values)
+        best_actions = [a for a, v in enumerate(values) if v >= best_value - 1e-9]
+        return self.rng.choice(best_actions)
+
+    def update(self, round_profile, round_payoffs):
+        if self.last_state is None:
+            return
+        action = round_profile[self.player_index]
+        reward = round_payoffs[self.player_index]
+        next_state = self._opponents_of(round_profile)
+        best_next = max(self._q(next_state, a) for a in range(self.n))
+        old = self._q(self.last_state, action)
+        self.q[(self.last_state, action)] = (
+            old + self.alpha * (reward + self.gamma * best_next - old)
+        )
+        self.last_state = next_state
+
+
+class LLMPlayer(BasePlayer):
+    """An LLM-driven player that queries Claude (Anthropic API)."""
+
+    def __init__(self, game, player_index, rng=None, debug_logger=None, model="claude-3-5-sonnet-20240620"):
+        super().__init__(game, player_index, rng=rng, debug_logger=debug_logger)
+        import anthropic
+        self.client = anthropic.Anthropic()
+        self.model = model
+
+    def choose(self, history):
+        game = self.game
+        player_name = game.players[self.player_index]
+        
+        my_strategies = game.strategies[self.player_index]
+        strategy_options_str = "\n".join(
+            f"  Index {i}: {strat}" for i, strat in enumerate(my_strategies)
+        )
+        
+        payoff_lines = []
+        for profile in itertools.product(*(range(n) for n in game.strategy_counts())):
+            labels = " | ".join(
+                f"{game.strategy_name(i, s)}" for i, s in enumerate(profile)
+            )
+            payoffs = ", ".join(
+                f"{game.payoff(i, profile):.2f}" for i in range(game.num_players)
+            )
+            payoff_lines.append(f"  Profile: ({labels}) -> Payoffs: ({payoffs})")
+        payoff_matrix_str = "\n".join(payoff_lines)
+        
+        history_lines = []
+        for r_idx, profile in enumerate(history):
+            round_str = ", ".join(
+                f"{game.players[i]}: {game.strategy_name(i, s)}" for i, s in enumerate(profile)
+            )
+            history_lines.append(f"  Round {r_idx + 1}: {round_str}")
+        history_str = "\n".join(history_lines) if history_lines else "  No rounds played yet."
+
+        prompt = f"""You are playing a normal-form game as Player '{player_name}'.
+Your strategy options and their indices:
+{strategy_options_str}
+
+The payoff matrix for all combinations (order of players: {", ".join(game.players)}):
+{payoff_matrix_str}
+
+History of recent rounds:
+{history_str}
+
+Analyze the game and the history of play. Decide on your next move.
+You must output a raw JSON object with EXACTLY the following keys:
+- "reasoning": A string explaining your strategic decision.
+- "strategy_index": An integer representing the strategy index of your chosen move.
+
+Output ONLY the raw JSON object. Do not include any other text or markdown formatting outside the JSON."""
+
+        response = self.client.messages.create(
+            model=self.model,
+            max_tokens=1000,
+            messages=[{"role": "user", "content": prompt}],
+        )
+        text = response.content[0].text
+        
+        parsed_json = self._extract_json(text)
+        reasoning = parsed_json.get("reasoning", "")
+        strategy_index = int(parsed_json.get("strategy_index"))
+        
+        if self.debug_logger is not None:
+            self.debug_logger(self.player_index, reasoning)
+            
+        return strategy_index
+
+    def _extract_json(self, text):
+        cleaned = text.strip()
+        if cleaned.startswith("```"):
+            lines = cleaned.splitlines()
+            if lines[0].startswith("```"):
+                lines = lines[1:]
+            if lines and lines[-1].strip() == "```":
+                lines = lines[:-1]
+            cleaned = "\n".join(lines).strip()
+            
+        try:
+            return json.loads(cleaned)
+        except json.JSONDecodeError:
+            start = cleaned.find("{")
+            end = cleaned.rfind("}")
+            if start == -1 or end == -1:
+                raise ValueError(f"No JSON object found in model response:\n{text}")
+            return json.loads(cleaned[start:end+1])
+
+
+PLAYER_KINDS = [
+    "best_response",
+    "regret_matching",
+    "softmax",
+    "q_learning",
+    "llm",
+]
+
+
+def make_player(
+    kind,
+    game,
+    player_index,
+    *,
+    rng=None,
+    memory=True,
+    temperature=1.0,
+    alpha=0.1,
+    gamma=0.9,
+    epsilon=0.1,
+    debug_logger=None,
+    model="claude-3-5-sonnet-20240620",
+):
+    """Construct a player of the requested kind with the given hyperparameters."""
+    if kind == "best_response":
+        return BestResponsePlayer(
+            game, player_index, memory=memory, rng=rng, debug_logger=debug_logger
+        )
+    if kind == "softmax":
+        return SoftmaxBestResponsePlayer(
+            game, player_index, memory=memory, temperature=temperature, rng=rng,
+            debug_logger=debug_logger,
+        )
+    if kind == "regret_matching":
+        return RegretMatchingPlayer(
+            game, player_index, rng=rng, debug_logger=debug_logger
+        )
+    if kind == "q_learning":
+        return QLearningPlayer(
+            game, player_index, alpha=alpha, gamma=gamma, epsilon=epsilon, rng=rng,
+            debug_logger=debug_logger,
+        )
+    if kind == "llm":
+        return LLMPlayer(
+            game, player_index, rng=rng, debug_logger=debug_logger, model=model
+        )
+    raise ValueError(f"Unknown player kind: {kind!r}")
+