diff --git a/Shayan1/Prod/nash/players.py b/Shayan1/Prod/nash/players.py new file mode 100644 index 0000000..fba93db --- /dev/null +++ b/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}") +