ESROppoMod.py

import random
from collections import Counter
from QLearnerESR import scalarize
import numpy as np
from scipy.optimize import minimize

class ESROppoMod:
    def __init__(self, agent_id, alpha, gamma, epsilon, num_oppo_actions, num_actions, opt=False, rand_prob=False):
        self.agent_id = agent_id
        self.alpha = alpha
        self.gamma = gamma
        self.epsilon = epsilon
        self.num_oppo_actions = num_oppo_actions
        self.num_actions = num_actions
        self.num_objectives = 1

        # Opponent ID used to pull opponent actions from action history in MONFG.py
        # while updating q_table.
        if agent_id == 0:
            self.oppo_id = 1
        elif agent_id == 1:
            self.oppo_id = 0

        if opt:  # init with optimistic q_values
            self.q_table = np.ones((num_oppo_actions, num_actions, self.num_objectives)) * 20
        else:  # init with zeroed q_values
            self.q_table = np.zeros((num_oppo_actions, num_actions, self.num_objectives))

        self.current_state = -1
        self.rand_prob = rand_prob
        self.history_window = 50
        self.oppo_action_history = []

    def update_q_table(self, oppo_action, action, reward):
        # Appending opponents actions to oppo_action_history
        if len(self.oppo_action_history) >= self.history_window:
            self.oppo_action_history.pop(0)  # removing oldest action
            self.oppo_action_history.append(oppo_action)  # adding newest action to maintain 50 action window
        else:
            self.oppo_action_history.append(oppo_action)
        payoff = scalarize(self.agent_id, reward)  # scalarizing payoff vector
        self.q_table[oppo_action][action] += self.alpha * (payoff - self.q_table[oppo_action][action])

    # epsilon greedy based on nonlinear optimiser mixed strategy search
    def select_action_mixed_nonlinear(self, state):
        self.current_state = state
        if random.uniform(0.0, 1.0) < self.epsilon:
            return self.select_random_action()
        else:
            return self.select_action_greedy_mixed_nonlinear(state)

    # random action selection
    def select_random_action(self):
        random_action = np.random.randint(self.num_actions)
        return random_action

    # greedy action selection based on nonlinear optimiser mixed strategy search
    def select_action_greedy_mixed_nonlinear(self, state):
        strategy = self.calc_mixed_strategy_nonlinear(state)
        if isinstance(strategy, int) or isinstance(strategy, np.int64):
            return strategy
        else:
            if np.sum(strategy) != 1:
                strategy = strategy / np.sum(strategy)
            return np.random.choice(range(self.num_actions), p=strategy)

    def calc_mixed_strategy_nonlinear(self, state):
        if self.rand_prob:
            s0 = np.random.random(self.num_actions)
            s0 /= np.sum(s0)
        else:
            s0 = np.full(self.num_actions, 1.0 / self.num_actions)  # initial guess set to equal prob over all actions
        b = (0.0, 1.0)
        bnds = (b,) * self.num_actions
        con1 = {'type': 'eq', 'fun': lambda x: np.sum(x) - 1}
        cons = ([con1])
        solution = minimize(self.objective, s0, bounds=bnds, constraints=cons)
        strategy = solution.x
        return strategy

    # this is the objective function to be minimised by the nonlinear optimiser
    # (therefore it returns the negative of ESR)
    def objective(self, strategy):
        return - self.calc_esr_from_strategy(strategy)

    # Calculates the ESR for a given strategy using the agent's own Q val
    def calc_esr_from_strategy(self, strategy):
        esr = self.calc_expected_scalar(self.current_state, strategy)
        return esr

    def calc_expected_scalar(self, state, strategy):
        opponent_probs = self.calc_oppo_probs()
        expectations = opponent_probs @ self.q_table
        return np.dot(expectations.T, np.array(strategy))  # expected scalars

    # returns probability distribution using historical data for relevant agent.
    # used to determine policies.
    def calc_oppo_probs(self):
        if len(self.oppo_action_history) > 0:
            count = Counter(self.oppo_action_history)
            probs = [count[i] / sum(count.values()) for i in range(self.num_oppo_actions)]
            probs = np.array(probs)
        else:
            probs = np.zeros(self.num_oppo_actions)
        return probs