from .extended_util_space import ExtendedUtilSpace from .utils.opponent_model import OpponentModel from decimal import Decimal from geniusweb.actions.Accept import Accept from geniusweb.actions.Action import Action from geniusweb.actions.Offer import Offer from geniusweb.actions.PartyId import PartyId 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.issuevalue.Domain import Domain from geniusweb.party.Capabilities import Capabilities from geniusweb.party.DefaultParty import DefaultParty from geniusweb.profile.utilityspace.LinearAdditive import LinearAdditive from geniusweb.profileconnection.ProfileConnectionFactory import ProfileConnectionFactory from geniusweb.profileconnection.ProfileInterface import ProfileInterface from geniusweb.progress.Progress import Progress from geniusweb.references.Parameters import Parameters from json import dump, load from random import randint from statistics import mean from time import time as clock from tudelft.utilities.immutablelist.ImmutableList import ImmutableList from tudelft_utilities_logging.Reporter import Reporter import logging class ChargingBoul(DefaultParty): def __init__(self, reporter: Reporter = None): super().__init__(reporter) self.best_received_bid: Bid = None self.best_received_util: Decimal = Decimal(0) self.domain: Domain = None self.e: float = 0.1 self.extended_space: ExtendedUtilSpace = None self.filepath: str = None self.final_rounds: int = 90 self.last_received_bid: Bid = None self.last_received_util: Decimal = None self.max_util = Decimal(1) self.me: PartyId = None self.min_util = Decimal(0.5) self.opponent_model: OpponentModel = None self.opponent_strategy: str = None self.other: str = None self.parameters: Parameters = None self.profile_int: ProfileInterface = None self.progress: Progress = None self.received_bids: list = [] self.received_utils: list = [] self.settings: Settings = None self.storage_dir: str = None self.summary: dict = None self.util_space: LinearAdditive = None self.getReporter().log(logging.INFO, "party is initialized") def notifyChange(self, info: Inform): """MUST BE IMPLEMENTED This is the entry point of all interaction with your agent after is has been initialised. How to handle the received data is based on its class type. Args: info (Inform): Contains either a request for action or information. """ try: if isinstance(info, Settings): self.settings = info self.me = self.settings.getID() self.parameters = self.settings.getParameters() self.profile_int = ProfileConnectionFactory.create( self.settings.getProfile().getURI(), self.getReporter() ) self.progress = self.settings.getProgress() self.storage_dir = self.parameters.get("storage_dir") self.util_space = self.profile_int.getProfile() self.domain = self.util_space.getDomain() self.extended_space = ExtendedUtilSpace(self.util_space) self.detect_strategy() elif isinstance(info, ActionDone): other_act: Action = info.getAction() actor = other_act.getActor() if actor != self.me: self.other = str(actor).rsplit("_", 1)[0] self.filepath = f"{self.storage_dir}/{self.other}.json" if isinstance(other_act, Offer): # create opponent model if it was not yet initialised if self.opponent_model is None: self.opponent_model = OpponentModel(self.domain) self.last_received_bid = other_act.getBid() self.last_received_util = self.util_space.getUtility(self.last_received_bid) # update opponent model with bid self.opponent_model.update(self.last_received_bid) elif isinstance(info, YourTurn): self.my_turn() elif isinstance(info, Finished): self.getReporter().log(logging.INFO, "Final outcome:" + str(info)) self.terminate() # stop this party and free resources. except Exception as ex: self.getReporter().log(logging.CRITICAL, "Failed to handle info", ex) def getCapabilities(self) -> Capabilities: """MUST BE IMPLEMENTED Method to indicate to the protocol what the capabilities of this agent are. Leave it as is for the ANL 2022 competition Returns: Capabilities: Capabilities representation class """ return Capabilities( set(["SAOP"]), set(["geniusweb.profile.utilityspace.LinearAdditive"]), ) def getDescription(self) -> str: """MUST BE IMPLEMENTED Returns a description of your agent. 1 or 2 sentences. Returns: str: Agent description """ return ( "Increasingly random Boulwarish agent. Last second concessions based on opponent's strategy." ) ##################### private support funcs ######################### def detect_strategy(self): if self.filepath is not None: with open(self.filepath, "w") as f: self.summary = load(f) if self.summary["ubi"] >= 5: self.opponent_strategy = "boulware" self.e = 0.2 * 2**(5 - self.summary["ubi"]) elif self.summary["aui"] <= 2: self.opponent_strategy = "hardline" else: self.opponent_strategy = "concede" self.min_util = Decimal(0.4) def my_turn(self): # Keep history of received bids and best alternative if self.last_received_bid is not None: self.received_bids.append(self.last_received_bid) self.received_utils.append(self.last_received_util) if self.last_received_util > self.best_received_util: self.best_received_bid = self.last_received_bid self.best_received_util = self.last_received_util # Create new bid based on the point in the negotiation and opponent's strategy t = self.progress.get(clock() * 1000) if self.summary is not None and self.opponent_strategy == "boulware" and t > 1 - (1/2)**min(self.summary["ubi"], 10): # If Boulware opponent is not going to concede much more, try to make a reasonable concession bid = self.make_concession() else: bid = self.make_bid() # Check if we've previously gotten a better bid already if self.best_received_util >= self.util_space.getUtility(bid): i = self.received_utils.index(self.best_received_util) bid = self.received_bids.pop(i) self.received_utils.pop(i) # Find next bests self.best_received_util = max(self.received_utils) i = self.received_utils.index(self.best_received_util) self.best_received_bid = self.received_bids[i] # Take action my_action: Action if bid == None or ( self.last_received_bid != None and self.util_space.getUtility(self.last_received_bid) >= self.util_space.getUtility(bid) ): # if bid==null we failed to suggest next bid. my_action = Accept(self.me, self.last_received_bid) else: my_action = Offer(self.me, bid) self.getConnection().send(my_action) def make_concession(self): self.min_util = Decimal(0.3) opponent_util = self.opponent_model.get_predicted_utility(self.best_received_util) if self.best_received_util > self.min_util and opponent_util < 2*self.min_util: bid = self.best_received_bid else: bid = self.make_bid() return bid def make_bid(self) -> Bid: time = self.progress.get(clock() * 1000) utility_goal = self.get_utility_goal(time) options: ImmutableList[Bid] = self.extended_space.getBids(utility_goal, time) if options.size() == 0: # if we can't find good bid, get max util bid.... options = self.extended_space.getBids(self.max_util, time) # pick a random one. return options.get(randint(0, options.size() - 1)) def get_utility_goal(self, t: float) -> Decimal: ft1 = Decimal(1) if self.e != 0: ft1 = round(Decimal(1 - pow(t, 1 / self.e)), 6) # defaults ROUND_HALF_UP return max( min((self.min_util + (self.max_util - self.min_util) * ft1), self.max_util), self.min_util ) def terminate(self): self.save_data() self.getReporter().log(logging.INFO, "party is terminating:") super().terminate() if self.profile_int != None: self.profile_int.close() self.profile_int = None def save_data(self): ubi, aui = self.summarize_opponent() with open(self.filepath, "w") as f: dump({ "ubi": ubi, "aui": aui }, f) def summarize_opponent(self): # Detect how much the number of unique bids is increasing unique_bid_index = 0 s = round(len(self.received_bids)/2) left = self.received_bids[:s] right = self.received_bids[s:] while len(set(left)) > 0 and len(set(right)) > 0 and len(set(left)) < len(set(right)): unique_bid_index += 1 s = round(len(right)/2) left = right[:s] right = right[s:] # Detect how much average utility is increasing avg_utility_index = 0 s = round(len(self.received_utils)/2) left = self.received_utils[:s] right = self.received_utils[s:] while len(set(left)) > 0 and len(set(right)) > 0 and mean(left) < mean(right): avg_utility_index += 1 s = round(len(right)/2) left = right[:s] right = right[s:] return unique_bid_index, avg_utility_index