From ef9c3528a7be7d9222e891fcef53b80bcf7b84f2 Mon Sep 17 00:00:00 2001 From: Richard Wang Date: Wed, 4 Sep 2024 18:17:28 +0800 Subject: [PATCH 1/7] Create .gitignore --- envs/ieds/.gitignore | 1 + 1 file changed, 1 insertion(+) create mode 100644 envs/ieds/.gitignore diff --git a/envs/ieds/.gitignore b/envs/ieds/.gitignore new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/envs/ieds/.gitignore @@ -0,0 +1 @@ + From de146e6522f02f03ecc2eb92c0222cbc0d0d7976 Mon Sep 17 00:00:00 2001 From: Richard Wang Date: Wed, 4 Sep 2024 03:18:17 -0700 Subject: [PATCH 2/7] Add files for IEDS environment --- envs/ieds/env.py | 267 +++++++++++++++++++++++++++++++++ envs/ieds/generate_examples.py | 69 +++++++++ envs/ieds/program_agent.py | 96 ++++++++++++ envs/ieds/tools.py | 173 +++++++++++++++++++++ 4 files changed, 605 insertions(+) create mode 100644 envs/ieds/env.py create mode 100644 envs/ieds/generate_examples.py create mode 100644 envs/ieds/program_agent.py create mode 100644 envs/ieds/tools.py diff --git a/envs/ieds/env.py b/envs/ieds/env.py new file mode 100644 index 0000000..758e64d --- /dev/null +++ b/envs/ieds/env.py @@ -0,0 +1,267 @@ +import sys +import os +root_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.realpath(__file__)))) +sys.path.append(root_dir) + +import numpy as np +from itertools import product +from copy import deepcopy +from pydantic import BaseModel, NonNegativeInt +from typing import List, Optional + +from envs.env_helper import BaseEnv, get_env_param + +description_of_problem_class_ieds = """ +Iterated Elimination of Dominated Strategies (IEDS) is a process used in game theory to simplify the analysis of strategic interactions by iteratively removing strategies that are strictly dominated. + +Components: +Players: A finite set of players, each with a finite set of strategies. +Payoff Matrix: A matrix that represents the payoffs for each player given the strategies chosen by all players. + +Interaction protocol: +1. Identify strictly dominated strategies for each player and remove them. +2. Update the game by considering only the remaining strategies. +3. Repeat steps 1 and 2 until no more strictly dominated strategies can be identified. + +Goal of the players: +Each player aims to maximize their payoff, and the IEDS process helps in narrowing down the set of strategies to those that could be rationally chosen by eliminating those that are dominated. +""" + +class State(BaseModel): + iteration: int + remaining_strategies: List[List[NonNegativeInt]] + payoff_matrix: List[List[List[int]]] + textual_descript: str + mathematical_descript: str + + def is_valid_action(self, action): + return True + + def generate_remaining_payoff_matrix(self): + '''Generate the payoff matrix with only the remaining strategies''' + cur_payoff_matrix = [] + for player in range(len(self.payoff_matrix)): + player_matrix = [] + for i in self.remaining_strategies[0]: + row = [] + for j in self.remaining_strategies[1]: + row.append(self.payoff_matrix[player][i][j]) + player_matrix.append(row) + cur_payoff_matrix.append(player_matrix) + print("Ending Payoff for Player 1:\n{}".format(cur_payoff_matrix[0])) + print("Ending Payoff for Player 2:\n{}".format(cur_payoff_matrix[1])) + print("==============================================================") + return cur_payoff_matrix + + def update_textual_descript(self): + '''Update the textual description to include remaining payoff matrices''' + self.textual_descript = "ITERATION: {}\nStarting Strategies for Player {}:\n{}\nStarting Strategies for Player {}:\n{}\n".format( + self.iteration, 1, self.remaining_strategies[0], 2, self.remaining_strategies[1] + ) + +class IteratedEliminationDominatedStrategies(BaseEnv): + def __init__(self, env_param): + self.name = "ieds" + self.required_agents = ["player"] + self.env_param = env_param + self.num_players = env_param["num_players"] + self.cur_player = 0 + self.strategies_per_player = env_param["strategies_per_player"] + self.payoff_matrix = env_param["payoff_matrix"] + self.dominated_strategies = [] + + self.reset() + self.description_of_problem_class = description_of_problem_class_ieds + + def get_description(self, agent_role): + description = "This is the beginning of a new game instance, where you will play as a player using the Iterated Elimination of Dominated Strategies (IEDS) method. The game has {} players, each with a set of strategies. The payoff matrix is provided. Your goal is to iteratively eliminate strictly dominated strategies and simplify the game.\n".format(self.num_players) + return description + + def reset(self): + '''Reset the environment''' + self.state = State(iteration=1, remaining_strategies=[list(range(s)) for s in self.strategies_per_player], payoff_matrix=self.payoff_matrix, textual_descript="This is iteration {}. The remaining strategies are {}.".format(0, [list(range(s)) for s in self.strategies_per_player]), mathematical_descript="Initial state") + self.is_done = False + self.state.update_textual_descript() + + def step(self): + ''' + Perform one step of the IEDS process + + Returns: + newState - State - new state after elimination + ''' + self.state.iteration += 1 + elimination_occured = False + + for player in range(self.num_players): + self.cur_player = player + # self.eliminate_dominated_strategies() + dominated_strategy, dominating_strategy = self.find_strictly_dominated_strategy(self.cur_player) + + if dominated_strategy is not None and dominating_strategy is not None: + # self.eliminate_dominated_strategies(dominated_strategies) + print("Strategy {} strictly dominates strategy {} for Player {}.".format( + dominating_strategy, dominated_strategy, player+1)) + self.eliminate_strictly_dominated_strategy(dominated_strategy) + self.payoff_matrix = self.state.generate_remaining_payoff_matrix() + elimination_occured = True + if self.check_termination(): + self.is_done = True + break + + break + + if not elimination_occured: + self.check_termination() + self.is_done = True + + self.state.update_textual_descript() + + return self.state + + def eliminate_strictly_dominated_strategy(self, strategy): + '''Eliminate a single strictly dominated strategy for the current player''' + self.state.remaining_strategies[self.cur_player].remove(strategy) + # +1 to strategy and player to match 1-indexing for output clarity + print("Eliminating strictly dominated strategy {} for Player {}...\n".format(strategy, self.cur_player + 1)) + print("Ending strategies for Player 1:\n{}".format(self.state.remaining_strategies[0])) + print("Ending strategies for Player 2:\n{}\n".format(self.state.remaining_strategies[1])) + + + def find_strictly_dominated_strategy(self, player): + '''Find a strictly dominated strategy for a given player''' + remaining_strategies = self.state.remaining_strategies[player] + dominated_strategy = None + dominating_strategy = None + + for i in range(len(remaining_strategies)): + s1 = remaining_strategies[i] + for j in range(i + 1, len(remaining_strategies)): + s2 = remaining_strategies[j] + dominated_result = self.check_dominance(player, s1, s2) + + if dominated_result == s1: + dominated_strategy = s1 + dominating_strategy = s2 + + elif dominated_result == s2: + dominated_strategy = s2 + dominating_strategy = s1 + + if dominated_strategy is not None and dominating_strategy is not None: + break + + return dominated_strategy, dominating_strategy + + def check_dominance(self, player, s1, s2): + '''Check if strategy s1 is strictly dominated by strategy s2 or vice versa for player''' + opponent_indices = [i for i in range(self.num_players) if i != player] + opp_remaining_strategies = [self.state.remaining_strategies[i] for i in opponent_indices] + + s1_dominates = True + s2_dominates = True + + for opp_strategies in product(*opp_remaining_strategies): + # Build the full index list for accessing the payoff matrix + s1_index = [opp_remaining_strategies[i].index(opp_strategies[i]) for i in range(len(opponent_indices))] + s2_index = [opp_remaining_strategies[i].index(opp_strategies[i]) for i in range(len(opponent_indices))] + + # Map s1 and s2 to their positions in the remaining strategies list + mapped_s1 = self.state.remaining_strategies[player].index(s1) + mapped_s2 = self.state.remaining_strategies[player].index(s2) + + s1_index.insert(player, mapped_s1) + s2_index.insert(player, mapped_s2) + + payoff_s1 = self.payoff_matrix[player] + payoff_s2 = self.payoff_matrix[player] + + for idx in s1_index: + payoff_s1 = payoff_s1[idx] + for idx in s2_index: + payoff_s2 = payoff_s2[idx] + + if payoff_s1 >= payoff_s2: + s2_dominates = False + if payoff_s2 >= payoff_s1: + s1_dominates = False + + # Early exit if neither strictly dominates + if not s1_dominates and not s2_dominates: + return None + + if s1_dominates: + return s2 + elif s2_dominates: + return s1 # s1 strictly dominates s2 + else: + return None + + def is_dominated(self, player, s1, s2): + '''Check if strategy s1 is strictly dominated by strategy s2 for player''' + opponent_indices = [i for i in range(self.num_players) if i != player] + + opp_remaining_strategies = [] + for i in opponent_indices: + opp_remaining_strategies.append(self.state.remaining_strategies[i]) + + for opp_strategies in product(*opp_remaining_strategies): + # Build the full index list for accessing the payoff matrix + s1_index = [opp_remaining_strategies[i].index(opp_strategies[i]) for i in range(len(opponent_indices))] + s2_index = [opp_remaining_strategies[i].index(opp_strategies[i]) for i in range(len(opponent_indices))] + + # Map s1 and s2 to their positions in the remaining strategies list + mapped_s1 = self.state.remaining_strategies[player].index(s1) + mapped_s2 = self.state.remaining_strategies[player].index(s2) + + s1_index.insert(player, mapped_s1) + s2_index.insert(player, mapped_s2) + + payoff_s1 = self.payoff_matrix[player] + payoff_s2 = self.payoff_matrix[player] + + for idx in s1_index: + payoff_s1 = payoff_s1[idx] + for idx in s2_index: + payoff_s2 = payoff_s2[idx] + + if payoff_s1 >= payoff_s2: + return False + return True + + def generate_opponent_strategies(self, player): + '''Generate all combinations of opponent strategies for a given player''' + opponent_strategies = [self.state.remaining_strategies[i] for i in range(self.num_players) if i != player] + return opponent_strategies[0] + + def check_termination(self): + '''Check if no more strictly dominated strategies can be found''' + for player in range(self.num_players): + dominated_strategy, dominating_strategy = self.find_strictly_dominated_strategy(player) + if dominated_strategy is not None and dominating_strategy is not None: + return False + + remaining_strategies_count = [len(strategies) for strategies in self.state.remaining_strategies] + + if all(count == 1 for count in remaining_strategies_count): + print("Game ended: Unique PNE found.\n") + elif self.strategies_per_player[0] == remaining_strategies_count[0] and self.strategies_per_player[1] == remaining_strategies_count[1]: + print("Game ended early: No PNE found.\n") + elif 1 < remaining_strategies_count[0] < self.strategies_per_player[0] and 1 < remaining_strategies_count[1] < self.strategies_per_player[1]: + print("Game ended early: Multiple PNE found.\n") + + return True + + +if __name__ == "__main__": + ## num_pne: 0 for no PNE, 1 for unique PNE, 2+ for multiple PNE + env_param = get_env_param("ieds", random_param=False) + env = IteratedEliminationDominatedStrategies(env_param=env_param) + + state = env.state + print(state.textual_descript) + while not env.is_done: + state = env.step() + print(state.textual_descript) + + print("\nThe game has ended! \nFinal remaining strategies are: {}. \nFinal remaining payoff is: {}".format(state.remaining_strategies, env.payoff_matrix)) \ No newline at end of file diff --git a/envs/ieds/generate_examples.py b/envs/ieds/generate_examples.py new file mode 100644 index 0000000..b3075b5 --- /dev/null +++ b/envs/ieds/generate_examples.py @@ -0,0 +1,69 @@ +import sys +import os +root_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.realpath(__file__)))) +sys.path.append(root_dir) + +from datetime import datetime +import numpy as np +import argparse + +from env import IteratedEliminationDominatedStrategies +from program_agent import IEDSAgent +from utils import Logger +from envs.env_helper import get_env_param + +def play_through(game, agents, logger, exmps_file=None): + assert game.check_agents(agents) # check if all agents required by the game are specified + logger.write(game.description_of_problem_class) + # describe the decision making problem instance to each agent + for role in agents: + instance_description = game.get_description(agent_role=role) + logger.write(instance_description) + with open(exmps_file, "a") as f: + f.write("==== ASSISTANT ====\n") + f.write(instance_description + "\n") + + game.reset() + state = game.state + + while not game.is_done: + logger.write(state.textual_descript, color="green") + agents['player'].eliminate_dominated_strategies() + state = game.step() + + logger.write("The IEDS game has ended!\nFinal strategies for player 1 are:\n{}\nFinal strategies for player 2 are:\n{}".format(np.array(state.remaining_strategies[0]), np.array(state.remaining_strategies[1])), color="red") + logger.write("Final payoffs for player 1 are:\n{}\nFinal payoffs for player 2 are:\n{}".format(np.array(game.payoff_matrix[0]), np.array(game.payoff_matrix[1])), color="red") + +if __name__ == "__main__": + parser = argparse.ArgumentParser() + parser.add_argument('--game', type=str, default="ieds") + parser.add_argument('--random_param', action='store_true', default=False, help="Whether to use random parameters for the game") + parser.add_argument('--num_strategies', nargs="+", type=int, help="Number of strategies for each player") + parser.add_argument('--num_pne', type=int, default=1, help="Number of PNE in the game") + parser.add_argument('--seed', type=int, default=0, help="Random seed") + args = parser.parse_args() + + logger_output_path = "envs/ieds/outputs/" + os.makedirs(logger_output_path, exist_ok=True) + now = datetime.now() + time_string = now.strftime('%Y%m%d%H%M%S') + logger = Logger(logger_output_path + args.game + "-" + time_string + ".html", verbose=True, writeToFile=False) + + exmps_output_path = "envs/ieds/prompts/" + os.makedirs(exmps_output_path, exist_ok=True) + exmps_file = exmps_output_path + "ieds_exmps_" + time_string + ".txt" + + ### initialize game and agents ### + env_param = get_env_param(env_name="ieds", random_param=args.random_param, num_pne=args.num_pne, num_strategies=args.num_strategies) + game = IteratedEliminationDominatedStrategies(env_param=env_param) + working_memory = { + "payoff_matrix": game.payoff_matrix, + "current_player": game.cur_player, + "remaining_strategies": game.state.remaining_strategies, + "dominated_strategies": game.dominated_strategies + } + agent = IEDSAgent(working_memory=working_memory, exmps_file=exmps_file) + agents = {"player": agent} + + ### start play ### + play_through(game=game, agents=agents, logger=logger, exmps_file=exmps_file) diff --git a/envs/ieds/program_agent.py b/envs/ieds/program_agent.py new file mode 100644 index 0000000..90e89be --- /dev/null +++ b/envs/ieds/program_agent.py @@ -0,0 +1,96 @@ +from tools import * +import sys +import os +root_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.realpath(__file__)))) +sys.path.append(root_dir) + +class IEDSAgent: + def __init__(self, working_memory, exmps_file): + ''' + Agent for iterated elimination of dominated strategies. + + Args: + working_memory + exmps_file + ''' + self.working_memory = working_memory + self.exmps_file = exmps_file + + def eliminate_dominated_strategies(self): + with open(self.exmps_file, "a") as f: + f.write("==== USER ====\n") + f.write("Question: How do I perform iterated elimination of dominated strategies?\n") + f.write("Thought: To perform iterated elimination of dominated strategies, I need to find and eliminate strictly dominated strategies for each player until no more dominated strategies can be found.\n") + + while not self.check_termination(): + for player in range(len(self.working_memory["remaining_strategies"])): + dominated_strategy = self.find_dominated_strategy(player) + if dominated_strategy is not None: + self.update_strategies(player, dominated_strategy) + + with open(self.exmps_file, "a") as f: + f.write("==== USER ====\n") + f.write("Player {} has the following strictly dominated strategy eliminated: {}\n".format(player, dominated_strategy)) + + with open(self.exmps_file, "a") as f: + f.write("Thought: No more strictly dominated strategies can be found. The iterated elimination process is complete.\n") + + def find_dominated_strategy(self, player): + op = FindDominatedStrategy(player=player) + return op.execute(self.working_memory) + + def update_strategies(self, player, dominated_strategy): + self.working_memory["remaining_strategies"][player].remove(dominated_strategy) + with open(self.exmps_file, "a") as f: + f.write("Thought: Eliminated dominated strategy {} for player {}.\n".format(dominated_strategy, player)) + + def check_termination(self): + op = CheckTermination() + return op.execute(self.working_memory) + + def move(self, state_struct): + with open(self.exmps_file, "a") as f: + f.write("==== USER ====\n") + f.write(state_struct.textual_descript + "\n") + f.write("Question: Which action should I choose?\n") + f.write("Thought: I should choose an action that maximizes my payoff given the remaining strategies of the other players.\n") + + remaining_strategies = self.working_memory["remaining_strategies"][state_struct.cur_agent] + payoffs = [] + + for strategy in remaining_strategies: + payoff = self.compute_payoff(state_struct.cur_agent, strategy) + payoffs.append(payoff) + + with open(self.exmps_file, "a") as f: + f.write("Operation: compute payoffs for remaining strategies.\n") + f.write("Result: {}.\n".format(payoffs)) + + best_strategy_index = np.argmax(payoffs) + + with open(self.exmps_file, "a") as f: + f.write("Operation: choose the strategy with the highest payoff.\n") + f.write("Result: {}.\n".format(best_strategy_index)) + f.write("Thought: I will choose the strategy {} which has the highest payoff.\n".format(remaining_strategies[best_strategy_index])) + + return remaining_strategies[best_strategy_index] + + def compute_payoff(self, player, strategy): + """ + Compute the expected payoff for a given player and strategy given the remaining strategies of the other players. + """ + total_payoff = 0 + opponent_strategies = self.generate_opponent_strategies(player) + + for opp_strategy in opponent_strategies: + payoff = self.working_memory["payoff_matrix"][player][strategy][tuple(opp_strategy)] + total_payoff += payoff + + return total_payoff / len(opponent_strategies) + + def generate_opponent_strategies(self, player): + """ + Generate all combinations of opponent strategies for a given player. + """ + opponent_strategies = [self.working_memory["remaining_strategies"][i] for i in range(len(self.working_memory["remaining_strategies"])) if i != player] + return list(product(*[range(len(s)) for s in opponent_strategies])) diff --git a/envs/ieds/tools.py b/envs/ieds/tools.py new file mode 100644 index 0000000..178ba04 --- /dev/null +++ b/envs/ieds/tools.py @@ -0,0 +1,173 @@ +from pydantic import BaseModel, Field +from typing import List +import numpy as np +from itertools import product + +tool_names_ieds = [ + "EliminateDominatedStrategy", + "FindDominatedStrategy", + "CheckDominance", + "GenerateOpponentStrategies", + "CheckTermination" +] + +class EliminateDominatedStrategy(BaseModel): + """ + Eliminate a strictly dominated strategy for the current player + """ + def execute(self, working_memory): + required_params = ["remaining_strategies", "current_player", "payoff_matrix", "dominated_strategies", "num_players"] + for param in required_params: + if param not in working_memory: + return f"Parameter {param} is missing in the working memory." + + player = working_memory["current_player"] + dominated_strategies = FindDominatedStrategy(player=player).execute(working_memory) + if dominated_strategies: + # for strategy in dominated_strategies: + strategy = dominated_strategies[0] + working_memory["remaining_strategies"][player].remove(strategy) + working_memory["dominated_strategies"][player].append(strategy) + + updated_payoff_matrix = self.generate_remaining_payoff_matrix(working_memory) + working_memory["payoff_matrix"] = updated_payoff_matrix + + return "Dominated strategies eliminated and remaining strategies updated in working memory." + + def generate_remaining_payoff_matrix(self, working_memory): + '''Generate the payoff matrix with only the remaining strategies''' + cur_payoff_matrix = [] + remaining_strategies = working_memory["remaining_strategies"] + payoff_matrix = working_memory["payoff_matrix"] + for player in range(len(payoff_matrix)): + player_matrix = [] + for i in remaining_strategies[0]: + row = [] + for j in remaining_strategies[1]: + row.append(payoff_matrix[player][i][j]) + player_matrix.append(row) + cur_payoff_matrix.append(player_matrix) + + print("Remaining Payoff for Player 1:", cur_payoff_matrix[0]) + print("Remaining Payoff for Player 2:", cur_payoff_matrix[1]) + return cur_payoff_matrix + +class FindDominatedStrategy(BaseModel): + """ + Find a strictly dominated strategy for a given player. + """ + player: int = Field(..., description="The player for whom to find dominated strategies") + + def execute(self, working_memory): + required_params = ["remaining_strategies", "payoff_matrix"] + for param in required_params: + if param not in working_memory: + return f"Parameter {param} is missing in the working memory." + + remaining_strategies = working_memory["remaining_strategies"][self.player] + dominated_strategy = None + for i in range(len(remaining_strategies)): + s1 = remaining_strategies[i] + for j in range(i + 1, len(remaining_strategies)): + s2 = remaining_strategies[j] + # if s1 != s2 and CheckDominance(player=self.player, s1=s1, s2=s2).execute(working_memory): + # dominated_strategies.append(s1) + # break + dominated_result = CheckDominance(player=self.player, s1=s1, s2=s2).execute(working_memory) + if dominated_result == s1: + dominated_strategy = s1 + elif dominated_result == s2: + dominated_strategy = s2 + + return dominated_strategy + +class CheckDominance(BaseModel): + """ + Check if a strategy s1 is strictly dominated by strategy s2 or vice versa for a given player. + """ + player: int = Field(..., description="The player for whom to check dominated strategies") + s1: int = Field(..., description="Strategy s1 to check") + s2: int = Field(..., description="Strategy s2 to compare against") + + def execute(self, working_memory): + required_params = ["payoff_matrix", "remaining_strategies", "num_players"] + missing_params = [param for param in required_params if param not in working_memory] + if missing_params: + return f"Parameters {missing_params} are missing in the working memory." + + opponent_strategies = GenerateOpponentStrategies(player=self.player).execute(working_memory) + + # Indices of other players + opponent_indices = [i for i in range(working_memory["num_players"]) if i != self.player] + + # Initialize dominance checks + s1_dominated = True + s2_dominated = True + + for opp_strategies in opponent_strategies: + # Build the full index list for accessing the payoff matrix + s1_index = [opponent_strategies[i].index(opp_strategies[i]) for i in range(len(opponent_indices))] + s2_index = [opponent_strategies[i].index(opp_strategies[i]) for i in range(len(opponent_indices))] + + # Map s1 and s2 to their positions in the remaining strategies list + mapped_s1 = working_memory["remaining_strategies"][self.player].index(self.s1) + mapped_s2 = working_memory["remaining_strategies"][self.player].index(self.s2) + + s1_index.insert(self.player, mapped_s1) + s2_index.insert(self.player, mapped_s2) + + # Retrieve payoffs for the current strategy profiles + payoff_s1 = working_memory["payoff_matrix"][self.player] + payoff_s2 = working_memory["payoff_matrix"][self.player] + + for idx in s1_index: + payoff_s1 = payoff_s1[idx] + for idx in s2_index: + payoff_s2 = payoff_s2[idx] + + # Check for dominance conditions + if payoff_s1 >= payoff_s2: + s2_dominated = False + if payoff_s2 >= payoff_s1: + s1_dominated = False + + # Early exit if neither strategy is strictly dominated + if not s1_dominated and not s2_dominated: + return None + + if s1_dominated: + return self.s1 + elif s2_dominated: + return self.s2 + else: + return None + +class GenerateOpponentStrategies(BaseModel): + """ + Generate all combinations of opponent strategies for a given player. + """ + player: int = Field(..., description="The player for whom to generate opponent strategies") + + def execute(self, working_memory): + required_params = ["remaining_strategies", "current_player"] + for param in required_params: + if param not in working_memory: + return f"Parameter {param} is missing in the working memory." + + opponent_strategies = [working_memory["remaining_strategies"][i] for i in range(len(working_memory["remaining_strategies"])) if i != self.player] + return list(product(*opponent_strategies)) + +class CheckTermination(BaseModel): + """ + Check if no more strictly dominated strategies can be found. + """ + def execute(self, working_memory): + required_params = ["remaining_strategies", "payoff_matrix", "current_player"] + for param in required_params: + if param not in working_memory: + return f"Parameter {param} is missing in the working memory." + + for player in range(len(working_memory["remaining_strategies"])): + if FindDominatedStrategy(player=player).execute(working_memory) is not None: + return False + return True From 47ccf7c4d4f16708a78fa4e8193b80700210c250 Mon Sep 17 00:00:00 2001 From: Richard Wang Date: Wed, 4 Sep 2024 18:20:15 +0800 Subject: [PATCH 3/7] Update env_helper.py --- envs/env_helper.py | 362 ++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 359 insertions(+), 3 deletions(-) diff --git a/envs/env_helper.py b/envs/env_helper.py index b362656..eaf2c42 100644 --- a/envs/env_helper.py +++ b/envs/env_helper.py @@ -19,7 +19,7 @@ def reset(self): def step(self, action): pass -def get_env_param(env_name, random_param=False): +def get_env_param(env_name, random_param=True, num_pne=1, num_strategies=[5, 5]): if env_name == "bargain_alternate_singleissue": if random_param: # random.randint(3,4) @@ -114,10 +114,366 @@ def get_env_param(env_name, random_param=False): P_true[0, 1][1] = 0.6 P_true[nState - 1, 1][nState - 1] = 0.6 P_true[nState - 1, 1][nState - 2] = 0.4 - return {"nState":nState, "nAction":nAction, "epLen":epLen, "R":R_true, "P":P_true} + return {"nState": nState, "nAction": nAction, "epLen": epLen, "R": R_true, "P": P_true} + + elif env_name == "ieds": + if random_param: + # Unique PNE + if num_pne == 1: + payoff1, payoff2 = generate_payoff_bimatrix_with_unique_pne(num_strategies[0], num_strategies[1]) + return { + "num_players": 2, + "strategies_per_player": [num_strategies[0], num_strategies[1]], + "payoff_matrix": [payoff1.tolist(), payoff2.tolist()] + } + # No PNE + elif num_pne == 0: + payoff1, payoff2 = generate_payoff_bimatrix_with_no_pne(num_strategies[0], num_strategies[1]) + return { + "num_players": 2, + "strategies_per_player": [num_strategies[0], num_strategies[1]], + "payoff_matrix": [payoff1.tolist(), payoff2.tolist()] + } + # Multiple PNE + else: + payoff1, payoff2 = generate_payoff_bimatrix_with_multiple_pne(num_strategies[0], num_strategies[1], num_pne) + return { + "num_players": 2, + "strategies_per_player": [num_strategies[0], num_strategies[1]], + "payoff_matrix": [payoff1.tolist(), payoff2.tolist()] + } + else: + # fixed payoff matrix + # payoff_matrix = [[[10, 10], [9, 11]], [[10, 9], [10, 8]]] + # payoff_matrix = [[[10, 10, 8], [9, 11, 11], [8, 9, 9]], [[10, 9, 8], [10, 9, 8], [8, 11, 7]]] + # payoff_matrix = [[[0, 0], [0, 1]], [[0, 0], [0, 1]]] + # payoff_matrix = [[[5, 6], [8, 2]], [[5, 4], [7, 8]]] + # payoff_matrix = [[[-5, 2, 3], [1, 1, 1], [0, 0, 0]], [[-1, 2, 3], [-3, 2, 1], [10, 0, -10]]] + # payoff_matrix = [[ + # [2, 3, 2, 1], # Row A + # [8, 5, 4, 7], # Row B + # [100, 2, 1, 3], # Row C + # [3, 6, 2, 0] # Row D + # ] + # ,[ + # [1, 10, 9, 3], # Row A + # [7, 8, 9, 11], # Row B + # [9, 11, 2, 3], # Row C + # [0, 1, 2, 0] # Row D + # ]] + payoff_matrix = [ + [[12, 6, 7, 8], + [17, 3, 4, 5], + [18, 4, 5, 6], + [19, 5, 6, 7]], + [[19, 4, 5, 6], + [ 1, 8, 9, 10], + [ 2, 9, 10, 11], + [ 3, 10, 11, 12]]] + num_players = len(payoff_matrix) + strategies_per_player = [len(payoff_matrix[0]), len(payoff_matrix[0][0])] + return { + "num_players": num_players, + "strategies_per_player": strategies_per_player, + "payoff_matrix": payoff_matrix + } else: raise ValueError("Unknown game {}".format(env_name)) +def generate_payoff_bimatrix_with_no_pne(num_strategies_player1: int, num_strategies_player2: int): + """ + Generates a random game with no pure Nash Equilibrium using iterated elimination of dominated strategies. + + Args: + num_strategies_player1: Number of strategies for Player 1. + num_strategies_player2: Number of strategies for Player 2. + + Returns: + A tuple containing two numpy arrays representing the payoff matrices for Player 1 and Player 2. + """ + if num_strategies_player1 <= 0 or num_strategies_player2 <= 0: + raise ValueError("Number of strategies must be positive integers") + + elif num_strategies_player1 == 1 and num_strategies_player2 == 1: + raise ValueError("Game must have at least two strategies for each player") + + else: + # Generate distinct scores for each strategy + # initial_value_a = np.random.randint(5, 20) + # initial_value_b = np.random.randint(5, 20) + initial_value_a = 1 + initial_value_b = 0 + + m = num_strategies_player1 + n = num_strategies_player2 + A = np.zeros((m, n), dtype=int) + B = np.zeros((m, n), dtype=int) + + # Fill in the matrices using the antisymmetric property + for i in range(m): + for j in range(n): + if (i + j) % 2 == 0: + A[i, j] = initial_value_a + B[i, j] = initial_value_b + else: + A[i, j] = initial_value_a + B[i, j] = initial_value_b + + print("Payoff Matrix for Player 1: \n", A) + print("Payoff Matrix for Player 2: \n", B) + return A, B + +def generate_payoff_bimatrix_with_unique_pne(num_strategies_player1: int, num_strategies_player2: int): + """ + Generates a random bi-matrix game with a unique pure Nash Equilibrium using iterated elimination of dominated strategies. + + Args: + num_strategies_player1: Number of strategies for Player 1. + num_strategies_player2: Number of strategies for Player 2. + + Returns: + A tuple containing two numpy arrays representing the payoff matrices for Player 1 and Player 2. + """ + def add_dominated_row(A, B, increment): + min_values_A = np.min(A, axis=0) + new_row_A = min_values_A - increment + A = np.vstack((A, new_row_A)) + + new_row_B = np.zeros(A.shape[1], dtype=int) + for i in range(A.shape[1]): + if i == 0: + new_row_B[i] = np.min(B[:, i]) - increment + else: + new_row_B[i] = np.max(B[:, i-1]) + increment + + B = np.vstack((B, new_row_B)) + return A, B + + def add_dominated_column(A, B, increment): + min_values_B = np.min(B, axis=1) + new_col_B = min_values_B - increment + B = np.column_stack((B, new_col_B)) + + new_col_A = np.zeros(B.shape[0], dtype=int) + for i in range(B.shape[0]): + if i == 0: + new_col_A[i] = np.min(A[i, :]) - increment + else: + new_col_A[i] = np.max(A[i-1, :]) + increment + + A = np.column_stack((A, new_col_A)) + return A, B + + if num_strategies_player1 <= 0 or num_strategies_player2 <= 0: + raise ValueError("Number of strategies must be positive integers") + + elif num_strategies_player1 == 1 and num_strategies_player2 == 1: + raise ValueError("Game must have at least two strategies for each player") + + elif num_strategies_player1 == num_strategies_player2: + # Scenario 1: n x n bi-matrix game with unique PNE + n = num_strategies_player1 + print("Number of strategies for each player: ", n) + + initial_value = np.random.randint(30, 50) + increment = np.random.randint(1, 5) + # increment = 1 + A = np.array([[initial_value, initial_value], + [initial_value - increment, initial_value + increment]]) + B = np.array([[initial_value, initial_value - increment], [initial_value, initial_value - increment]]) + + # Expand the matrices alternately to ensure unique IEDS order + for size in range(3, n + 1): + if A.shape[0] < n: # Add row if needed + A, B = add_dominated_row(A, B, increment) + + if B.shape[1] < n: # Add column if needed + A, B = add_dominated_column(A, B, increment) + + A, B, _, _ = permute_payoff_matrices([A, B]) + + print("Payoff Matrix for Player 1: \n", A) + print("Payoff Matrix for Player 2: \n", B) + return A, B + + elif num_strategies_player1 > num_strategies_player2: + # Scenario 2: m x n (m > n) bi-matrix game with unique IEDS order + m = num_strategies_player1 + n = num_strategies_player2 + print("WARNING: The order of IEDS will not be unique for this case.") + print("Number of strategies for Player 1: ", m) + print("Number of strategies for Player 2: ", n) + + initial_value = np.random.randint(30, 50) + increment = np.random.randint(1, 5) + + A = np.array([[initial_value, initial_value], [initial_value - increment, initial_value + increment]]) + B = np.array([[initial_value, initial_value - increment], [initial_value, initial_value - increment]]) + + if n < len(B) and n == 1: + B = np.delete(B, 1, axis=0) + A = np.delete(A, 1, axis=0) + + # Building the base n x n matrix + for size in range(3, n + 1): + A, B = add_dominated_row(A, B, increment) + A, B = add_dominated_column(A, B, increment) + + # Expanding to m x n by adding rows + while A.shape[0] < m: + A, B = add_dominated_row(A, B, increment) + + A, B, _, _ = permute_payoff_matrices([A, B]) + + print("Payoff Matrix for Player 1: \n", A) + print("Payoff Matrix for Player 2: \n", B) + + return A, B + + elif num_strategies_player1 < num_strategies_player2: + # Scenario 3: m x n (m < n) bi-matrix game with unique IEDS order + m = num_strategies_player1 + n = num_strategies_player2 + print("WARNING: The order of IEDS will not be unique for this case.") + print("Number of strategies for Player 1: ", m) + print("Number of strategies for Player 2: ", n) + + initial_value = np.random.randint(30, 50) + increment = np.random.randint(1, 5) + + A = np.array([[initial_value, initial_value], [initial_value - increment, initial_value + increment]]) + B = np.array([[initial_value, initial_value - increment], [initial_value, initial_value - increment]]) + + temp = A + A = np.transpose(B) + B = np.transpose(temp) + + if m < len(A) and m == 1: + A = np.delete(A, 1, axis=0) + B = np.delete(B, 1, axis=0) + + # Building the base m x m matrix + for size in range(3, m + 1): + A, B = add_dominated_row(A, B, increment) + A, B = add_dominated_column(A, B, increment) + + # Expanding to m x n by adding columns + while B.shape[1] < n: + A, B = add_dominated_column(A, B, increment) + + A, B, _, _ = permute_payoff_matrices([A, B]) + + print("Payoff Matrix for Player 1: \n", A) + print("Payoff Matrix for Player 2: \n", B) + + return A, B + +def generate_payoff_bimatrix_with_multiple_pne(num_strategies_player1: int, num_strategies_player2: int, num_pne: int): + """ + Generates a random game with multiple pure Nash Equilibria to be solvable using iterated elimination of dominated strategies. + + Args: + - num_strategies_player1 (int): Number of strategies for Player 1. + - num_strategies_player2 (int): Number of strategies for Player 2. + - num_pne (int): Number of pure Nash Equilibria desired. + + Returns: + - A (np.array): Payoff matrix for Player 1. + - B (np.array): Payoff matrix for Player 2. + """ + m = num_strategies_player1 + n = num_strategies_player2 + + # Ensure number of PNEs does not exceed the minimum number of strategies for both players + assert num_pne <= min(m, n), "Number of PNEs must be less than or equal to the minimum number of strategies for both players." + + # Initialize empty payoff matrices for Player 1 (A) and Player 2 (B) + A = np.zeros((num_pne, num_pne), dtype=int) + B = np.zeros((num_pne, num_pne), dtype=int) + + # Generate the initial l x l base matrix with PNEs on the diagonal + initial_value = 10 + increment = 1 + + for i in range(num_pne): + for j in range(num_pne): + if i == j: + A[i, j] = initial_value + B[i, j] = initial_value + else: + A[i, j] = A[i, i] - abs(i - j) + B[i, j] = B[i, i] - abs(i - j) + + def add_dominated_row(A, B, increment): + min_values_A = np.min(A, axis=0) + new_row_A = min_values_A - increment + A = np.vstack((A, new_row_A)) + + new_row_B = np.zeros(A.shape[1], dtype=int) + for i in range(A.shape[1]): + if i == 0: + new_row_B[i] = np.min(B[:, i]) - increment + else: + new_row_B[i] = np.max(B[:, i-1]) + increment + + B = np.vstack((B, new_row_B)) + return A, B + + def add_dominated_column(A, B, increment): + min_values_B = np.min(B, axis=1) + new_col_B = min_values_B - increment + B = np.column_stack((B, new_col_B)) + + new_col_A = np.zeros(B.shape[0], dtype=int) + for i in range(B.shape[0]): + if i == 0: + new_col_A[i] = np.min(A[i, :]) - increment + else: + new_col_A[i] = np.max(A[i-1, :]) + increment + + A = np.column_stack((A, new_col_A)) + return A, B + + # Expand the l x l base matrix to m x n matrix + for size in range(num_pne, max(m, n)): + if A.shape[0] < m: # Add a row if needed + A, B = add_dominated_row(A, B, increment) + if B.shape[1] < n: # Add a column if needed + A, B = add_dominated_column(A, B, increment) + + print("Payoff Matrix for Player 1: \n", A) + print("Payoff Matrix for Player 2: \n", B) + + return A, B + +def permute_matrix(matrix, row_permutation, col_permutation): + """Permutes the rows and columns of a matrix using given permutations.""" + permuted_matrix = matrix.copy() + + # Apply the provided permutations + permuted_matrix = permuted_matrix[row_permutation, :] + permuted_matrix = permuted_matrix[:, col_permutation] + return permuted_matrix + +def permute_payoff_matrices(payoff_matrices): + """Permutes the rows and columns of the payoff matrices for all players using the same permutations.""" + + # Get the shape of the matrices + n_rows, n_cols = payoff_matrices[0].shape + + # Generate the same random permutations for both matrices + row_permutation = np.random.permutation(n_rows) + col_permutation = np.random.permutation(n_cols) + + # Apply the same permutation to both payoff matrices + permuted_matrices = [] + for matrix in payoff_matrices: + permuted_matrix = permute_matrix(np.array(matrix), row_permutation, col_permutation) + permuted_matrices.append(permuted_matrix) + + # Return the permuted matrices along with the applied permutations + return permuted_matrices[0], permuted_matrices[1], row_permutation, col_permutation + def _randDense(states, actions, mask): """Generate random dense ``P`` and ``R``. See ``rand`` for details. @@ -246,4 +602,4 @@ def randmdp(S, A, is_sparse=False, mask=None): else: P = np.ones((A, S, S)) R = np.random.uniform(low=-1, high=1, size=(S,A)) - return P, R \ No newline at end of file + return P, R From 652403144d723baf7784bfbd9df44e39c315c9a2 Mon Sep 17 00:00:00 2001 From: Richard Wang Date: Wed, 4 Sep 2024 18:22:23 +0800 Subject: [PATCH 4/7] Update env_helper.py --- envs/env_helper.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/envs/env_helper.py b/envs/env_helper.py index eaf2c42..87bdfd0 100644 --- a/envs/env_helper.py +++ b/envs/env_helper.py @@ -385,7 +385,7 @@ def generate_payoff_bimatrix_with_multiple_pne(num_strategies_player1: int, num_ n = num_strategies_player2 # Ensure number of PNEs does not exceed the minimum number of strategies for both players - assert num_pne <= min(m, n), "Number of PNEs must be less than or equal to the minimum number of strategies for both players." + assert num_pne < min(m, n), "Number of PNEs must be less than the minimum number of strategies for both players." # Initialize empty payoff matrices for Player 1 (A) and Player 2 (B) A = np.zeros((num_pne, num_pne), dtype=int) From 20ef61989e159f32a5a294adec653b50aa808a33 Mon Sep 17 00:00:00 2001 From: Richard Wang Date: Wed, 4 Sep 2024 18:31:02 +0800 Subject: [PATCH 5/7] Update generate_examples.py --- envs/ieds/generate_examples.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/envs/ieds/generate_examples.py b/envs/ieds/generate_examples.py index b3075b5..895efe6 100644 --- a/envs/ieds/generate_examples.py +++ b/envs/ieds/generate_examples.py @@ -38,7 +38,7 @@ def play_through(game, agents, logger, exmps_file=None): parser = argparse.ArgumentParser() parser.add_argument('--game', type=str, default="ieds") parser.add_argument('--random_param', action='store_true', default=False, help="Whether to use random parameters for the game") - parser.add_argument('--num_strategies', nargs="+", type=int, help="Number of strategies for each player") + parser.add_argument('--num_strategies', nargs="+", type=int, default=[5,5], help="Number of strategies for each player") parser.add_argument('--num_pne', type=int, default=1, help="Number of PNE in the game") parser.add_argument('--seed', type=int, default=0, help="Random seed") args = parser.parse_args() From 582a8d92e204670553d158189999db20f50c8fda Mon Sep 17 00:00:00 2001 From: Richard Wang Date: Wed, 4 Sep 2024 18:31:33 +0800 Subject: [PATCH 6/7] Update env_helper.py --- envs/env_helper.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/envs/env_helper.py b/envs/env_helper.py index 87bdfd0..7e9cc89 100644 --- a/envs/env_helper.py +++ b/envs/env_helper.py @@ -19,7 +19,7 @@ def reset(self): def step(self, action): pass -def get_env_param(env_name, random_param=True, num_pne=1, num_strategies=[5, 5]): +def get_env_param(env_name, random_param=True, num_pne=1, num_strategies=None): if env_name == "bargain_alternate_singleissue": if random_param: # random.randint(3,4) From 10ff72aff4cf4eb370d5545617d3f31bbc215137 Mon Sep 17 00:00:00 2001 From: Richard Wang Date: Wed, 4 Sep 2024 18:32:18 +0800 Subject: [PATCH 7/7] Update env_helper.py --- envs/env_helper.py | 19 +------------------ 1 file changed, 1 insertion(+), 18 deletions(-) diff --git a/envs/env_helper.py b/envs/env_helper.py index 7e9cc89..bd08cd8 100644 --- a/envs/env_helper.py +++ b/envs/env_helper.py @@ -143,24 +143,7 @@ def get_env_param(env_name, random_param=True, num_pne=1, num_strategies=None): "payoff_matrix": [payoff1.tolist(), payoff2.tolist()] } else: - # fixed payoff matrix - # payoff_matrix = [[[10, 10], [9, 11]], [[10, 9], [10, 8]]] - # payoff_matrix = [[[10, 10, 8], [9, 11, 11], [8, 9, 9]], [[10, 9, 8], [10, 9, 8], [8, 11, 7]]] - # payoff_matrix = [[[0, 0], [0, 1]], [[0, 0], [0, 1]]] - # payoff_matrix = [[[5, 6], [8, 2]], [[5, 4], [7, 8]]] - # payoff_matrix = [[[-5, 2, 3], [1, 1, 1], [0, 0, 0]], [[-1, 2, 3], [-3, 2, 1], [10, 0, -10]]] - # payoff_matrix = [[ - # [2, 3, 2, 1], # Row A - # [8, 5, 4, 7], # Row B - # [100, 2, 1, 3], # Row C - # [3, 6, 2, 0] # Row D - # ] - # ,[ - # [1, 10, 9, 3], # Row A - # [7, 8, 9, 11], # Row B - # [9, 11, 2, 3], # Row C - # [0, 1, 2, 0] # Row D - # ]] + # fixed payoff bi-matrix payoff_matrix = [ [[12, 6, 7, 8], [17, 3, 4, 5],