source: CSE3210/agent41/agent41.py

import decimal
import time
from random import randint
from typing import cast

from geniusweb.actions.Accept import Accept
from geniusweb.actions.Action import Action
from geniusweb.actions.Offer import Offer
from geniusweb.bidspace.AllBidsList import AllBidsList
from geniusweb.inform.ActionDone import ActionDone
from geniusweb.inform.Finished import Finished
from geniusweb.inform.Inform import Inform
from geniusweb.inform.Settings import Settings
from geniusweb.inform.YourTurn import YourTurn
from geniusweb.issuevalue.Bid import Bid
from geniusweb.opponentmodel.FrequencyOpponentModel import FrequencyOpponentModel
from geniusweb.party.Capabilities import Capabilities
from geniusweb.party.DefaultParty import DefaultParty
from geniusweb.profileconnection.ProfileConnectionFactory import (
    ProfileConnectionFactory,
)
from geniusweb.progress.ProgressRounds import ProgressRounds
from tudelft_utilities_logging.Reporter import Reporter

reservation_progress = 0.995
search_numb = 2500


class Agent41(DefaultParty):
    """
    Template agent that offers random bids until a bid with sufficient utility is offered.
    """

    def __init__(self, reporter: Reporter = None):
        super().__init__(reporter)
        # self.getReporter().log(logging.INFO, "party is initialized")
        self._profile = None
        self._last_received_bid: Bid = None
        self._best_past_bid: Bid = None
        self.opponentModel = FrequencyOpponentModel.create()


        #  maximal advantage that our opponent might have over us to accept the bid in the later phase
        self.util_adv_from_accept = decimal.Decimal(0.2)
        # maximal advantage that our opponent might have over us
        # to propose the bid in the later phase
        self.util_adv_from_offer = decimal.Decimal(0.1)
        # maximal advantage that our agent might have over the opponent
        # to propose the bid in the later phase (potential to be removed)
        self.util_adv_to_offer = decimal.Decimal(0.8)
        # progress point at which the intermediate strategy starts being applied
        self.progress_mid = 0.6
        # progress point at which the late strategy starts being applied
        self.progress_fast = 0.95
        # starting utility range for which the offers are considered
        self.utility_range = [decimal.Decimal(0.9),
                              decimal.Decimal(1.1)]
        # linear decrease factors in different stages of the negotiations
        self.slow_decrease = decimal.Decimal(0.0005)
        self.mid_decrease = decimal.Decimal(0.001)
        self.fast_decrease = decimal.Decimal(0.002)
        # initial reservation value under which bids are denied
        self.minimal_reservation_val = decimal.Decimal(0.6)

    def notifyChange(self, info: Inform):
        """This is the entry point of all interaction with your agent after is has been initialised.

        Args:
            info (Inform): Contains either a request for action or information.
        """

        # a Settings message is the first message that will be send to your
        # agent containing all the information about the negotiation session.
        if isinstance(info, Settings):
            self._settings: Settings = cast(Settings, info)
            self._me = self._settings.getID()

            # progress towards the deadline has to be tracked manually through the use of the Progress object
            self._progress: ProgressRounds = self._settings.getProgress()

            # the profile contains the preferences of the agent over the domain
            self._profile = ProfileConnectionFactory.create(
                info.getProfile().getURI(), self.getReporter()
            )

            domain = self._profile.getProfile().getDomain()
            self.opponentModel = self.opponentModel.With(newDomain=domain, newResBid=0)

        # ActionDone is an action send by an opponent (an offer or an accept)
        elif isinstance(info, ActionDone):
            action: Action = cast(ActionDone, info).getAction()

            self.opponentModel = self.opponentModel.WithAction(action=action,
                                                               progress=self._progress)

            # if it is an offer, set the last received bid
            if isinstance(action, Offer):
                self._last_received_bid = cast(Offer, action).getBid()


        # YourTurn notifies you that it is your turn to act
        elif isinstance(info, YourTurn):
            action = self._my_turn()
            if isinstance(self._progress, ProgressRounds):
                self._progress = self._progress.advance()
            self.getConnection().send(action)

        # Finished will be send if the negotiation has ended (through agreement or deadline)
        elif isinstance(info, Finished):
            # terminate the agent MUST BE CALLED
            self.terminate()
        # else:
        # self.getReporter().log(
        #     logging.WARNING, "Ignoring unknown info " + str(info)
        # )

    # lets the geniusweb system know what settings this agent can handle
    # leave it as it is for this competition
    def getCapabilities(self) -> Capabilities:
        return Capabilities(
            set(["SAOP"]),
            set(["geniusweb.profile.utilityspace.LinearAdditive"]),
        )

    # terminates the agent and its connections
    # leave it as it is for this competition
    def terminate(self):
        # self.getReporter().log(logging.INFO, "party is terminating:")
        super().terminate()
        if self._profile is not None:
            self._profile.close()
            self._profile = None

    

    # give a description of your agent
    def getDescription(self) -> str:
        return "Agent orange for Collaborative AI course"

    def _my_turn(self):
        """
        Executes a turn.
        """
        # update the utility range
        self._update_range()

        # check if the last received offer if the opponent is good enough
        if self._is_good(self._last_received_bid):
            # if so, accept the offer
            action = Accept(self._me, self._last_received_bid)
        else:
            # if not, find a bid to propose as counter offer
            bid = self._find_bid()
            action = Offer(self._me, bid)

        # send the action
        return action

    # method that checks if we would agree with an offer
    def _is_good(self, bid: Bid) -> bool:
        """
        Checks if the received offer is good enough to be accepted.
        """
        if bid is None:
            return False

        profile = self._profile.getProfile()

        # Update the best past bid effectively making it a reservation bid
        if self._best_past_bid is None:
            self._best_past_bid = bid
        elif profile.getUtility(bid) > profile.getUtility(self._best_past_bid):
            self._best_past_bid = bid

        # get the utility of us and our opponent (frequency model)
        our_util = profile.getUtility(bid)
        opp_util = self.opponentModel.getUtility(bid)
        progress = self._progress.get(time.time() * 1000)

        # immediately return the offer it is below our lower utility threshold
        if our_util < self.utility_range[0]:
            return False
        # if the progress reached a certain point,
        # accept the ofer only if it is not overwhelmingly better than the opponents
        if progress > self.progress_mid:
            return our_util + self.util_adv_from_accept >= opp_util
        # if this progress has not been reached yet,
        # simply accept the offer if it is better than our utility
        return True

    def _find_bid(self) -> Bid:
        """
        Find the most suitable bid to offer given the current state of negotiations.
        """
        domain = self._profile.getProfile().getDomain()
        profile = self._profile.getProfile()

        # verify if the reservation bid should be used
        if self._verify_reservation_val():
            return self._best_past_bid

        # store the best bid found so far
        best_bid: Bid = self._best_past_bid
        # store the difference between the utility of the best bid
        # and the upper threshold of the range
        best_dist = decimal.Decimal(1.0)

        all_bids = AllBidsList(domain)
        # search through a number of random bids to find a suitable bid
        for _ in range(search_numb):
            # get the random bid and the utility of it
            bid = all_bids.get(randint(0, all_bids.size() - 1))
            our_util = profile.getUtility(bid)
            # verify if the bid is suitable to be offered to the opponent
            if self._verify_bid(bid):
                best_bid = bid
                break
            # If the best bid was not found, check if the current bid is closer to idea value
            curr_dist = abs(our_util - self.utility_range[1])
            if curr_dist < best_dist:
                best_bid = bid
                best_dist = curr_dist

        return best_bid

    def _update_range(self):
        """
        Update the acceptance range depending on the current progress of the negotiation.
        The further we are into the negotiation, the faster the utility range drops.
        """
        progress = self._progress.get(time.time() * 1000)
        if progress < self.progress_mid:
            self.utility_range[1] = self.utility_range[1] - self.slow_decrease
            self.utility_range[0] = self.utility_range[0] - self.slow_decrease
        elif progress > self.progress_fast:
            self.utility_range[1] = self.utility_range[1] - self.fast_decrease
            self.utility_range[0] = self.utility_range[0] - self.fast_decrease
        else:
            self.utility_range[1] = self.utility_range[1] - self.mid_decrease
            self.utility_range[0] = self.utility_range[0] - self.mid_decrease

    def _verify_reservation_val(self) -> bool:
        """
        Return the best bid received so far, if the time is almost up and the bid is good enough.
        The reservation value is equal to the best offer received from the opponent.
        It is sent if our utility coming from this offer exceeds 0.7
        or if the utility of the reservation bid exceeds our upper utility threshold.
        """
        if self._best_past_bid is not None:
            reservation_val = self._profile.getProfile().getUtility(
                self._best_past_bid)
            if self._progress.get(
                    0) >= reservation_progress and reservation_val >= self.minimal_reservation_val:
                return True
            if self._profile.getProfile().getUtility(self._best_past_bid) >= self.utility_range[1]:
                return True
        return False

    def _verify_bid(self, bid: Bid) -> bool:
        """
        Verifies if the bid that our agent came up with is good enough to be offered to another agent.
        """
        progress = self._progress.get(time.time() * 1000)
        profile = self._profile.getProfile()
        our_util = profile.getUtility(bid)

        # do not consider the bid if it is worse than our reservation bid
        if self._best_past_bid is not None and our_util < profile.getUtility(self._best_past_bid):
            return False
        # if a certain point in the negotiation has not been reached,
        # propose the bid if it falls into the accepted utility range
        if progress < self.progress_mid:
            return self.utility_range[0] < our_util < self.utility_range[1]

        # otherwise, in addition to considering the utility range, also consider the utility of our opponent, make sure
        # that the utility difference between us and the opponent is not too high
        opp_util = self.opponentModel.getUtility(bid)
        return our_util + self.util_adv_from_offer > opp_util > our_util - self.util_adv_to_offer \
               and self.utility_range[0] < our_util < self.utility_range[1]