negmas.mechanisms

Provides interfaces for defining negotiation mechanisms.

class negmas.mechanisms.Mechanism(initial_state: TState | None = None, outcome_space: OutcomeSpace | None = None, issues: Sequence[Issue] | None = None, outcomes: Sequence[Outcome] | int | None = None, n_steps: int | float | None = None, time_limit: float | None = None, pend: float = 0, pend_per_second: float = 0, step_time_limit: float | None = None, negotiator_time_limit: float | None = None, hidden_time_limit: float = inf, max_n_negotiators: int | None = None, dynamic_entry=False, annotation: dict[str, Any] | None=None, nmi_factory: type[TNMI] = <class 'negmas.common.NegotiatorMechanismInterface'>, extra_callbacks=False, checkpoint_every: int = 1, checkpoint_folder: PathLike | None = None, checkpoint_filename: str | None = None, extra_checkpoint_info: dict[str, Any] | None=None, single_checkpoint: bool = True, exist_ok: bool = True, name=None, genius_port: int = 25337, id: str | None = None, type_name: str | None = None, verbosity: int = 0, ignore_negotiator_exceptions=False)

Bases: NamedObject, EventSource, CheckpointMixin, Generic[TNMI, TState, TAction, TNegotiator], ABC

Base class for all negotiation Mechanisms.

Override the round function of this class to implement a round of your mechanism

Parameters:

• outcome_space – The negotiation agenda

• outcomes – list of outcomes (optional as you can pass issues). If an int then it is the number of outcomes

• n_steps – Number of rounds allowed (None means infinity)

• time_limit – Number of real seconds allowed (None means infinity)

• pend – Probability of ending the negotiation at any step

• pend_per_second – Probability of ending the negotiation every second

• hidden_time_limit – Number of real seconds allowed but not visilbe to the negotiators

• max_n_negotiators – Maximum allowed number of negotiators.

• dynamic_entry – Allow negotiators to enter/leave negotiations between rounds

• cache_outcomes – If true, a list of all possible outcomes will be cached

• max_cardinality – The maximum allowed number of outcomes in the cached set

• annotation – Arbitrary annotation

• checkpoint_every – The number of steps to checkpoint after. Set to <= 0 to disable

• checkpoint_folder – The folder to save checkpoints into. Set to None to disable

• checkpoint_filename – The base filename to use for checkpoints (multiple checkpoints will be prefixed with step number).

• single_checkpoint – If true, only the most recent checkpoint will be saved.

• extra_checkpoint_info – Any extra information to save with the checkpoint in the corresponding json file as a dictionary with string keys

• exist_ok – IF true, checkpoints override existing checkpoints with the same filename.

• name – Name of the mechanism session. Should be unique. If not given, it will be generated.

• genius_port – the port used to connect to Genius for all negotiators in this mechanism (0 means any).

• id – An optional system-wide unique identifier. You should not change the default value except in special circumstances like during serialization and should always guarantee system-wide uniquness if you set this value explicitly

abort() → TState

Aborts the negotiation.

add(negotiator: TNegotiator, *, preferences: Preferences | None = None, role: str | None = None, ufun: BaseUtilityFunction | None = None) → bool | None

Add an negotiator to the negotiation.

Parameters:

• negotiator – The negotiator to be added.

• preferences – The utility function to use. If None, then the negotiator must already have a stored utility function otherwise it will fail to enter the negotiation.

• ufun – [depricated] same as preferences but must be a UFun object.

• role – The role the negotiator plays in the negotiation mechanism. It is expected that mechanisms inheriting from this class will check this parameter to ensure that the role is a valid role and is still possible for negotiators to join on that role. Roles may include things like moderator, representative etc based on the mechanism

Returns:

• True if the negotiator was added.

• False if the negotiator was already in the negotiation.

• None if the negotiator cannot be added. This can happen in the following cases:

  1. The capabilities of the negotiator do not match the requirements of the negotiation

  2. The outcome-space of the negotiator’s preferences do not contain the outcome-space of the negotiation

  3. The negotiator refuses to join (by returning False from its join method) see Negotiator.join for possible reasons of that

add_requirements(requirements: dict) → None

Adds requirements.

property agent_ids: list[str | None]

Agent ids.

Returns:

The result.

Return type:

list[str | None]

property agent_names: list[str | None]

Agent names.

Returns:

The result.

Return type:

list[str | None]

property agreement

Agreement.

property atomic_steps: bool

Is every step corresponding to a single action by a single negotiator

can_accept_more_negotiators() → bool

Whether the mechanism can currently accept more negotiators.

can_enter(negotiator: TNegotiator) → bool

Whether the negotiator can enter the negotiation now.

can_leave(negotiator: Negotiator) → bool

Can the negotiator leave now?

can_participate(negotiator: TNegotiator) → bool

Checks if the negotiator can participate in this type of negotiation in general.

Returns:

True if the negotiator can participate

Return type:

bool

Remarks:

The only reason this may return False is if the mechanism requires some requirements that are not within the capabilities of the negotiator.

When evaluating compatibility, the negotiator is considered incapable of participation if any of the following conditions hold: * A mechanism requirement is not in the capabilities of the negotiator * A mechanism requirement is in the capabilities of the negotiator by the values required for it

is not in the values announced by the negotiator.

An negotiator that lists a None value for a capability is announcing that it can work with all its values. On the other hand, a mechanism that lists a requirement as None announces that it accepts any value for this requirement as long as it exist in the negotiator

property completed

Ended without timing out (either with agreement or broken by a negotiator)

property current_step

Current step.

discrete_outcome_space(levels: int = 5, max_cardinality: int = 10000000000) → DiscreteOutcomeSpace

Returns a stable discrete version of the given outcome-space.

discrete_outcomes(levels: int = 5, max_cardinality: int | float = inf) → list[tuple]

A discrete set of outcomes that spans the outcome space.

Parameters:

• max_cardinality – The maximum number of outcomes to return. If None, all outcomes will be returned for discrete issues

• continuous (and 10_000 if any of the issues was)

Returns:

list of n or less outcomes

Return type:

list[Outcome]

property dynamic_entry

Dynamic entry.

property ended

Ended in any way

property expected_relative_time: float

Returns a positive number indicating elapsed relative time or steps.

Remarks:

• This is relative to the expected time/step at which the negotiation ends given all timing conditions (time_limit, n_step, pend, pend_per_second).

property expected_remaining_steps: int | None

Returns the expected remaining number of steps until the end of the mechanism run.

None if unlimited

property expected_remaining_time: float | None

Returns remaining time in seconds (expectation).

None if no time limit or pend_per_second is given.

genius_id(id: str | None) → str | None

Gets the Genius ID corresponding to the given negotiator if known otherwise its normal ID

property genius_negotiator_ids: list[str]

Genius negotiator ids.

Returns:

The result.

Return type:

list[str]

get_negotiator(source: str) → Negotiator | None

Returns the negotiator with the given ID if present in the negotiation.

get_negotiator_raise(source: str) → Negotiator

Returns the negotiator with the given ID if present in the negotiation otherwise it raises an exception.

property history: list[TState]

History.

Returns:

The result.

Return type:

list[TState]

is_satisfying(capabilities: dict) → bool

Checks if the given capabilities are satisfying mechanism requirements.

Parameters:

capabilities – capabilities to check

Returns:

bool are the requirements satisfied by the capabilities.

Remarks:

• Requirements are also a dict with the following meanings:

  • tuple: Min and max acceptable values

  • list/set: Any value in the iterable is acceptable

  • Single value: The capability must match this value

• Capabilities can also have the same three possibilities.

is_valid(outcome: tuple)

Checks whether the outcome is valid given the issues.

property issues: list[Issue]

Returns the issues of the outcome space (if defined).

Will raise an exception if the outcome space has no defined issues

kalai_points(max_cardinality: float = inf, frontier: tuple[tuple[float, ...], ...] | None = None, frontier_outcomes: list[tuple] | None = None) → tuple[tuple[tuple[float, ...], tuple], ...]

Kalai points.

Parameters:

• max_cardinality – Max cardinality.

• frontier – Frontier.

• frontier_outcomes – Frontier outcomes.

Returns:

The result.

Return type:

tuple[tuple[tuple[float, …], Outcome], …]

log(nid: str, data: dict[str, Any], level: str) → None

Saves a log for a negotiator

log_critical(nid: str, data: dict[str, Any]) → None

Logs at critical level

log_debug(nid: str, data: dict[str, Any]) → None

Logs at debug level

log_error(nid: str, data: dict[str, Any]) → None

Logs at error level

log_info(nid: str, data: dict[str, Any]) → None

Logs at info level

log_warning(nid: str, data: dict[str, Any]) → None

Logs at warning level

property max_n_negotiators

Max n negotiators.

max_relative_welfare_points(max_cardinality: float = inf, frontier: tuple[tuple[float, ...], ...] | None = None, frontier_outcomes: list[tuple] | None = None) → tuple[tuple[tuple[float, ...], tuple], ...]

Max relative welfare points.

Parameters:

• max_cardinality – Max cardinality.

• frontier – Frontier.

• frontier_outcomes – Frontier outcomes.

Returns:

The result.

Return type:

tuple[tuple[tuple[float, …], Outcome], …]

max_welfare_points(max_cardinality: float = inf, frontier: tuple[tuple[float, ...], ...] | None = None, frontier_outcomes: list[tuple] | None = None) → tuple[tuple[tuple[float, ...], tuple], ...]

Max welfare points.

Parameters:

• max_cardinality – Max cardinality.

• frontier – Frontier.

• frontier_outcomes – Frontier outcomes.

Returns:

The result.

Return type:

tuple[tuple[tuple[float, …], Outcome], …]

modified_kalai_points(max_cardinality: float = inf, frontier: tuple[tuple[float, ...], ...] | None = None, frontier_outcomes: list[tuple] | None = None) → tuple[tuple[tuple[float, ...], tuple], ...]

Modified kalai points.

Parameters:

• max_cardinality – Max cardinality.

• frontier – Frontier.

• frontier_outcomes – Frontier outcomes.

Returns:

The result.

Return type:

tuple[tuple[tuple[float, …], Outcome], …]

property n_outcomes

N outcomes.

property n_steps

N steps.

nash_points(max_cardinality: float = inf, frontier: tuple[tuple[float, ...], ...] | None = None, frontier_outcomes: list[tuple] | None = None) → tuple[tuple[tuple[float, ...], tuple], ...]

Nash points.

Parameters:

• max_cardinality – Max cardinality.

• frontier – Frontier.

• frontier_outcomes – Frontier outcomes.

Returns:

The result.

Return type:

tuple[tuple[tuple[float, …], Outcome], …]

property negotiator_ids: list[str]

Negotiator ids.

Returns:

The result.

Return type:

list[str]

negotiator_index(source: str) → int | None

Gets the negotiator index.

Parameters:

source (str) – source

Return type:

int | None

property negotiator_names: list[str]

Negotiator names.

Returns:

The result.

Return type:

list[str]

property negotiator_times: dict[str, float]

The total time consumed by every negotiator.

Each mechanism class is responsible of updating this for any activities of the negotiator it controls.

property negotiators: list[TNegotiator]

Negotiators.

Returns:

The result.

Return type:

list[TNegotiator]

on_mechanism_error() → None

Called when there is a mechanism error.

Remarks:

• When overriding this function you MUST call the base class version

on_negotiation_end() → None

Called at the end of each negotiation.

Remarks:

• When overriding this function you MUST call the base class version

on_negotiation_start() → bool

Called before starting the negotiation.

If it returns False then negotiation will end immediately

property outcome_space: OutcomeSpace

Outcome space.

Returns:

The result.

Return type:

OutcomeSpace

property outcomes

Outcomes.

pareto_frontier(max_cardinality: float = inf, sort_by_welfare=True) → tuple[tuple[tuple[float, ...], ...], list[tuple]]

Pareto frontier.

Parameters:

• max_cardinality – Max cardinality.

• sort_by_welfare – Sort by welfare.

Returns:

The result.

Return type:

tuple[tuple[tuple[float, …], …], list[Outcome]]

pareto_frontier_bf(max_cardinality: float = inf, sort_by_welfare=True) → tuple[list[tuple[float, ...]], list[tuple]]

Pareto frontier bf.

Parameters:

• max_cardinality – Max cardinality.

• sort_by_welfare – Sort by welfare.

Returns:

The result.

Return type:

tuple[list[tuple[float, …]], list[Outcome]]

property participants: list[NegotiatorInfo]

Returns a list of all participant names.

plot(**kwargs) → Any

A method for plotting a negotiation session.

random_outcome() → tuple

Returns a single random offer

random_outcomes(n: int = 1, with_replacement: bool = False) → list[tuple]

Returns random outcomes.

Parameters:

• n – Number of outcomes to generate

• with_replacement – If true, outcomes may be repeated

Returns:

A list of outcomes of at most n outcomes.

Remarks:

• If the number of outcomes n cannot be satisfied, a smaller number will be returned

• Sampling is done without replacement (i.e. returned outcomes are unique).

property relative_time: float

Returns a number between 0 and 1 indicating elapsed relative time or steps.

Remarks:

• If pend or pend_per_second are defined in the NegotiatorMechanismInterface, and time_limit/n_steps are not given, this becomes an expectation that is limited above by one.

property remaining_steps: int | None

Returns the remaining number of steps until the end of the mechanism run.

None if unlimited

property remaining_time: float | None

Returns remaining time in seconds.

None if no time limit is given.

remove(negotiator: TNegotiator) → bool | None

Remove the negotiator from the negotiation.

Returns:

• True if the negotiator was removed.

• False if the negotiator was not in the negotiation already.

• None if the negotiator cannot be removed.

remove_requirements(requirements: Iterable) → None

Adds requirements.

property requirements

A dictionary specifying the requirements that must be in the capabilities of any negotiator to join the mechanism.

run(timeout=None) → TState

Run.

Parameters:

timeout – Timeout.

Returns:

The result.

Return type:

TState

run_with_progress(timeout=None) → TState

Run with progress.

Parameters:

timeout – Timeout.

Returns:

The result.

Return type:

TState

classmethod runall(mechanisms: list[Mechanism] | tuple[Mechanism, ...], keep_order=True, method: str = 'ordered', ordering: tuple[int, ...] | None = None, ordering_fun: Callable[[int, list[TState | None]], int] | None = None, completion_callback: Callable[[int, Mechanism], None] | None = None, ignore_mechanism_exceptions: bool = False) → list[TState | None]

Runs all mechanisms.

Parameters:

• mechanisms – list of mechanisms

• keep_order – if True, the mechanisms will be run in order every step otherwise the order will be randomized at every step. This is only allowed if the method is ordered

• method – the method to use for running all the sessions. Acceptable options are: sequential, ordered, threads, processes

• ordering – Controls the order of advancing the negotiations with the “ordered” method.

• ordering_fun – A function to implement dynamic ordering for the “ordered” method. This function receives a list of states and returns the index of the next mechanism to step. Note that a state may be None if the corresponding mechanism was None and it should never be stepped

• completion_callback – A callback to be called at the moment each mechanism is ended

• ignore_mechanism_exceptions – If given, mechanisms with exceptions will be treated as ending and running will continue

Returns:

• list of states of all mechanisms after completion

• None for any such states indicates disagreements

Remarks:

• sequential means running each mechanism until completion before going to the next

• ordered means stepping mechanisms in some order which can be controlled by ordering. If no ordering is given, the ordering is just round-robin

property running

Running.

property state: TState

Returns the current state.

Override extra_state if you want to keep extra state

property state4history: Any

Returns the state as it should be stored in the history.

property stats

Stats.

step(action: dict[str, TAction] | None = None) → TState

Runs a single step of the mechanism.

Returns:

The state of the negotiation after the round is conducted action: An optional action (value) for the next negotiator (key). If given, the call

should just execute the action without calling the next negotiator.

Return type:

MechanismState

Remarks:

• Every call yields the results of one round (see round())

• If the mechanism was yet to start, it will start it and runs one round

• There is another function (run()) that runs the whole mechanism in blocking mode

classmethod stepall(mechanisms: list[Mechanism] | tuple[Mechanism, ...], keep_order=True) → list[TState]

Step all mechanisms.

Parameters:

• mechanisms – list of mechanisms

• keep_order – if True, the mechanisms will be run in order every step otherwise the order will be randomized at every step

Returns:

• list of states of all mechanisms after completion

property time: float

Elapsed time since mechanism started in seconds.

0.0 if the mechanism did not start running

property time_limit

Time limit.

property verbosity: int

Verbosity level.

• Children of this class should only print if verbosity > 1

class negmas.mechanisms.MechanismStepResult(state: TState, completed: bool = True, broken: bool = False, timedout: bool = False, agreement: tuple | None = None, error: bool = False, error_details: str = '', waiting: bool = False, exceptions: dict[str, list[str]] | None = None, times: dict[str, float] | None = None)

Bases: Generic[TState]

Represents the results of a negotiation step.

This is what round() should return.

agreement: tuple | None

The agreement if any. Allows for a single outcome or a collection of outcomes.

broken: bool

True only if END_NEGOTIATION was selected by one negotiator.

completed: bool

Whether the current round is completed or not.

error: bool

True if an error occurred in the mechanism.

error_details: str

Error message.

exceptions: dict[str, list[str]] | None

A mapping from negotiator ID to a list of exceptions raised by that negotiator in this round.

state: TState

The returned state.

timedout: bool

True if a timeout occurred.

times: dict[str, float] | None

A mapping from negotiator ID to the time it consumed during this round.

waiting: bool

Whether to consider that the round is still running and call the round method again without increasing the step number.

class negmas.mechanisms.Traceable(*args, **kwargs)

Bases: Protocol

A mechanism that can generate a trace

property full_trace: list[TraceElement]

Returns the negotiation history as a list of relative_time/step/negotiator/offer tuples

negotiator_full_trace(negotiator_id: str) → list[TraceElement]

Returns the (time/relative-time/step/outcome/response) given by a negotiator (in order)