negmas.mechanisms

Provides interfaces for defining negotiation mechanisms.

class negmas.mechanisms.CompletedRun(history: list[TState] | list[TraceElement] | list[tuple], history_type: str, scenario: Scenario | None, agreement: Outcome | None, agreement_stats: OutcomeOptimality | None, outcome_stats: dict[str, Any], config: dict[str, Any], metadata: dict[str, Any])

Bases: Generic[TState]

Represents a completed negotiation run with all its data.

This class encapsulates the results of a negotiation including the history, scenario, agreement, and various statistics. It can be saved to and loaded from disk in various formats.

history

The negotiation history/trace data.

Type:

list[TState] | list[TraceElement] | list[tuple]

history_type

Type of history stored (“history”, “full_trace”, “trace”, “extended_trace”).

Type:

str

scenario

The negotiation scenario with outcome space and utility functions.

Type:

Scenario | None

agreement

The final agreement reached, or None if no agreement.

Type:

Outcome | None

agreement_stats

Optimality statistics for the agreement.

Type:

OutcomeOptimality | None

outcome_stats

Basic outcome statistics (agreement, broken, timedout, utilities).

Type:

dict[str, Any]

config

Configuration parameters used for the negotiation.

Type:

dict[str, Any]

metadata

Arbitrary metadata associated with the run.

Type:

dict[str, Any]

agreement: Outcome | None

agreement_stats: OutcomeOptimality | None

config: dict[str, Any]

history: list[TState] | list[TraceElement] | list[tuple]

history_type: str

classmethod load(path: Path | str, load_scenario: bool = True, load_scenario_stats: bool = False, load_agreement_stats: bool = True, load_config: bool = True) → CompletedRun[TState]

Loads a completed run from the given path.

Parameters:

• path – Path to a file (single-file mode) or directory (multi-file mode).

• load_scenario – If True, load the scenario from the scenario directory.

• load_scenario_stats – If True, load scenario statistics when loading the scenario.

• load_agreement_stats – If True, load agreement optimality statistics.

• load_config – If True, load the configuration from config.yaml.

Returns:

A CompletedRun instance with the loaded data.

Raises:

• FileNotFoundError – If the path does not exist.

• ValueError – If the path format is not recognized.

metadata: dict[str, Any]

outcome_stats: dict[str, Any]

save(parent: Path | str, name: str, single_file: bool = False, per_negotiator: bool = False, save_scenario: bool = True, save_scenario_stats: bool = False, save_agreement_stats: bool = True, save_config: bool = True, overwrite: bool = True, warn_if_existing: bool = True, storage_format: Literal['csv', 'gzip', 'parquet'] | None = 'csv') → Path

Saves the completed run to disk.

Parameters:

• parent – Parent directory where to save the run.

• name – Name for the saved run (directory name or file name without extension).

• single_file – If True, save only the trace as a single file.

• per_negotiator – If True, save traces per negotiator in a subdirectory. Files are named {type}@{index}_{name}.{ext}. Only works with full_trace history type.

• save_scenario – If True, save the scenario information.

• save_scenario_stats – If True, save scenario statistics.

• save_agreement_stats – If True, save agreement statistics.

• save_config – If True, save the configuration.

• overwrite – If True, overwrite existing files/directories.

• warn_if_existing – If True, warn when overwriting.

• storage_format – Format for table storage (“csv”, “gzip”, “parquet”).

Returns:

Path to the saved file or directory.

scenario: Scenario | None

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

Merges new requirements into the existing mechanism requirements.

Parameters:

requirements – Dict mapping requirement names to acceptable values

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

Returns the current negotiation step number (0-indexed).

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

Returns whether negotiators can join/leave during negotiation.

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]

Complete history of mechanism states throughout the negotiation.

Returns:

Chronological list of all negotiation states from start to current

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

make_config() → dict[str, Any]

Create a YAML-serializable configuration dict from the mechanism state.

Returns:

A dictionary containing mechanism configuration that can be safely serialized to YAML.

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

Returns the total number of possible outcomes in the outcome space.

property n_steps

Returns the maximum number of negotiation steps allowed, or None if unlimited.

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]

All negotiators participating in this negotiation mechanism.

Returns:

List of negotiator objects currently registered

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

The space of all possible negotiation outcomes.

Returns:

Defines valid outcomes including issues, values, and constraints

Return type:

OutcomeSpace

property outcomes

All possible outcomes for discrete spaces, or None for continuous spaces.

params: dict[str, Any]

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

Removes specified requirements from the mechanism.

Parameters:

requirements – Iterable of requirement names to remove

property requirements

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

run(timeout=None) → TState

Execute the negotiation mechanism until completion or timeout.

Parameters:

timeout – Maximum time in seconds to run, or None for no limit

Returns:

Final negotiation state after completion

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.

save(parent: Path | str, name: str, single_file: bool = False, per_negotiator: bool = False, save_scenario: bool = True, save_scenario_stats: bool = False, save_agreement_stats: bool = True, save_config: bool = True, source: str = 'history', metadata: dict[str, Any] | None = None, overwrite: bool = True, warn_if_existing: bool = True, storage_format: Literal['csv', 'gzip', 'parquet'] | None = 'csv') → Path

Saves the negotiation in a standard NegMAS format.

Parameters:

• parent – Where to save the negotiation

• name – Name to save to. It can be a directory name or a file name (without extension)

• single_file – If True, a single file with the negotiation history/trace will be saved. all save_* arguments will be ignored. The extension will be decided based on storage_format

• per_negotiator – If True, save traces per negotiator in a subdirectory. Files are named {type}@{index}_{name}.{ext}. Only works with full_trace source.

• save_scenario – If True (and single_file is False), save the negotiation scenario information with history (under a scenario folder)

• save_scenario_stats – If True (and single_file is False), save scenario stats and info (using Scenario.dumpas)

• save_agreement_stats – If True (and single_file is False), save agreement stats, the agreement itself and its utilities under outcome_stats.yaml. If utility functions are not found in negotiators, only the agreement will be saved there

• save_config – If True (and single_file is False), save the configuration paramters of the mechanism run.

• source – The source of the history information to save. Default is “history” which means using the history attribute.

• metadata – arbitrary data to save under meta_data.yaml. Only supported if single_file is False

• overwrite – Overwrite existing files/folders

• warn_if_existing – Warn if existing files/folders are found

• storage_format – The format for storing tables.

Returns:

The path to the saved file or directory.

Return type:

Path

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

Mechanism statistics collected during negotiation (e.g., step counts, agreement metrics).

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

Returns the maximum negotiation time in seconds, or infinity if unlimited.

to_completed_run(source: str = 'history', metadata: dict[str, Any] | None = None) → CompletedRun

Creates a CompletedRun object from the current mechanism state.

Parameters:

• source – The source of the history information. Options: - “history”: Use the history attribute (list of states) - “full_trace”: Use the full_trace attribute (if available) - “trace”: Use the trace attribute (if available) - “extended_trace”: Use the extended_trace attribute (if available)

• metadata – Arbitrary metadata to include in the CompletedRun.

Returns:

A CompletedRun object containing the negotiation data.

Return type:

CompletedRun

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

Detailed description of any error that occurred during the step.

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

extended_trace() → list[tuple[int, str, tuple]]

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

property full_trace: list[TraceElement]

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

property full_trace_with_utils: list[tuple]

Returns the full trace and the utility of the negotiators at each step.

negotiator_full_trace(negotiator_id: str) → list[tuple[float, float, int, tuple, str]]

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

property trace: list[tuple[str, tuple]]

Basic trace keeping only outcomes