Import source: Shayan1/Prod/nash/simulation

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

Shayan1/Prod/nash/simulation.py

@@ -0,0 +1,110 @@
+import numpy as np
+import random
+
+from nash.players import make_player
+
+
+class Simulation:
+    def __init__(self, game, num_rounds, memory=True, rng=None, convergence_window=5,
+                 player_kind="best_response", debug_logger=None):
+        self.game = game
+        self.num_rounds = num_rounds
+        self.memory = memory
+        self.convergence_window = convergence_window
+        self.rng = rng or random.Random()
+        self.players = [
+            make_player(player_kind, game, i, memory=memory, rng=self.rng,
+                        debug_logger=debug_logger)
+            for i in range(game.num_players)
+        ]
+        self.history = []
+        self.payoff_history = []
+
+    def run(self):
+        for _ in range(self.num_rounds):
+            profile = tuple(player.choose(self.history) for player in self.players)
+            payoffs = tuple(
+                self.game.payoff(i, profile) for i in range(self.game.num_players)
+            )
+            self.history.append(profile)
+            self.payoff_history.append(payoffs)
+            for player in self.players:
+                player.update(profile, payoffs)
+        return self.history
+
+    def converged_round(self):
+        """First round index where the profile repeats for convergence_window rounds
+        straight, or None if it never stabilizes."""
+        window = self.convergence_window
+        if len(self.history) < window:
+            return None
+        for start in range(len(self.history) - window + 1):
+            segment = self.history[start:start + window]
+            if all(profile == segment[0] for profile in segment):
+                return start
+        return None
+
+    def converged_profile(self):
+        round_idx = self.converged_round()
+        return None if round_idx is None else self.history[round_idx]
+
+    def strategy_frequencies(self):
+        """Per player, per strategy, the running frequency of that strategy having
+        been played, one value per round -- used to chart usage over time."""
+        strategy_counts = self.game.strategy_counts()
+        running_counts = [[0] * n for n in strategy_counts]
+        frequencies = [[[] for _ in range(n)] for n in strategy_counts]
+
+        for round_num, profile in enumerate(self.history, start=1):
+            for player_idx, strategy in enumerate(profile):
+                running_counts[player_idx][strategy] += 1
+            for player_idx, n in enumerate(strategy_counts):
+                for strategy_idx in range(n):
+                    freq = running_counts[player_idx][strategy_idx] / round_num
+                    frequencies[player_idx][strategy_idx].append(freq)
+
+        return frequencies
+
+class EvolutionarySimulation:
+    def __init__(self, game, num_rounds):
+        self.game = game
+        self.num_rounds = num_rounds
+        self.populations = [
+            np.full(len(game.strategies[i]), 1.0 / len(game.strategies[i]))
+            for i in range(game.num_players)
+        ]
+        self.history = []
+
+    def _expected_payoffs(self, player_idx):
+        payoff_tensor = self.game.payoffs[player_idx]
+        num_players = self.game.num_players
+        expr = payoff_tensor.copy()
+
+        for j in range(num_players):
+            if j == player_idx:
+                continue
+            shape = [1] * num_players
+            shape[j] = len(self.populations[j])
+            x_j_reshaped = self.populations[j].reshape(shape)
+            expr = expr * x_j_reshaped
+
+        axes_to_sum = tuple(j for j in range(num_players) if j != player_idx)
+        if axes_to_sum:
+            return expr.sum(axis=axes_to_sum)
+        else:
+            return expr
+
+    def run(self):
+        min_payoff = min(np.min(p) for p in self.game.payoffs)
+        for _ in range(self.num_rounds):
+            next_populations = []
+            for p in range(self.game.num_players):
+                u = self._expected_payoffs(p)
+                u_shifted = u - min_payoff + 1e-9
+                bar_u = np.sum(self.populations[p] * u_shifted)
+                new_pop = self.populations[p] * (u_shifted / bar_u)
+                new_pop = new_pop / np.sum(new_pop)
+                next_populations.append(new_pop)
+            self.populations = next_populations
+            self.history.append([pop.copy() for pop in self.populations])
+        return self.history