negmas.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]

convert(target: str, scenario: Scenario | None = None) → CompletedRun[TState]

Convert the history to a different trace format.

This method converts the current history to a different format. The conversion follows these rules: - Any format can be converted to a simpler format (losing information) - Converting to a more detailed format will set missing fields to None - Converting to full_trace_with_utils requires a scenario with utility functions

(either from self.scenario or the provided scenario parameter)

Format hierarchy (most detailed to least detailed):

full_trace_with_utils > full_trace > extended_trace > trace > history

Parameters:

• target – The target format. One of: “full_trace_with_utils”, “full_trace”, “extended_trace”, “trace”, “history”.

• scenario – Optional scenario with utility functions. If provided, it will be used for computing utilities when converting to full_trace_with_utils. If not provided, self.scenario will be used.

Returns:

A new CompletedRun with the converted history.

Raises:

• ValueError – If target is not a valid format.

• ValueError – If converting to full_trace_with_utils without a scenario with ufuns.

Examples

>>> from negmas.sao import SAOMechanism, RandomNegotiator
>>> m = SAOMechanism(outcomes=[(i,) for i in range(5)], n_steps=5)
>>> _ = m.add(RandomNegotiator(name="n1"))
>>> _ = m.add(RandomNegotiator(name="n2"))
>>> _ = m.run()
>>> # Create a CompletedRun with full_trace
>>> run = m.to_completed_run(source="full_trace")
>>> run.history_type
'full_trace'
>>> # Convert to trace format
>>> converted = run.convert("trace")
>>> converted.history_type
'trace'
>>> # Convert to extended_trace
>>> converted2 = run.convert("extended_trace")
>>> converted2.history_type
'extended_trace'

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

history_type: str

classmethod infer_source(path: Path | str) → str

Infer the source type from a saved negotiation file or folder.

Parameters:

path – Path to a file or directory containing saved negotiation data.

Returns:

“history”, “trace”, “extended_trace”, “full_trace”, or “full_trace_with_utils”.

Return type:

The inferred source type

Raises:

FileNotFoundError – If the path does not exist or no trace file is found.

Examples

>>> from pathlib import Path
>>> from negmas.sao import SAOMechanism, RandomNegotiator
>>> import tempfile
>>> import shutil
>>> # Create and run a simple negotiation
>>> m = SAOMechanism(outcomes=[(i,) for i in range(5)], n_steps=5)
>>> _ = m.add(RandomNegotiator(name="n1"))
>>> _ = m.add(RandomNegotiator(name="n2"))
>>> _ = m.run()
>>> # Save as full_trace
>>> tmpdir = Path(tempfile.mkdtemp())
>>> completed = m.to_completed_run(source="full_trace")
>>> save_path = completed.save(tmpdir, "test", single_file=False)
>>> # Infer the source type
>>> inferred = CompletedRun.infer_source(save_path)
>>> inferred == "full_trace"
True
>>> shutil.rmtree(tmpdir)

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.

Remarks:

When loading scenarios, we will look at a scenario folder in the given path and if not for a scenario_path in the metadata

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]

plot(plotting_negotiators: tuple[int, int] | tuple[str, str] = (0, 1), save_fig: bool = False, path: str | None = None, fig_name: str | None = None, image_format: str = 'webp', ignore_none_offers: bool = True, with_lines: bool = True, show_agreement: bool = False, show_pareto_distance: bool = True, show_nash_distance: bool = True, show_kalai_distance: bool = True, show_ks_distance: bool = True, show_max_welfare_distance: bool = True, show_max_relative_welfare_distance: bool = False, show_end_reason: bool = True, show_annotations: bool = False, show_reserved: bool = True, show_total_time: bool = True, show_relative_time: bool = True, show_n_steps: bool = True, colors: list | None = None, markers: list[str] | None = None, colormap: str = 'tab10', ylimits: tuple[float, float] | None = None, common_legend: bool = True, extra_annotation: str = '', xdim: str = 'relative_time', colorizer: Any | None = None, only2d: bool = False, no2d: bool = False, fast: bool = False, simple_offers_view: bool = False, mark_offers_view: bool = True, mark_pareto_points: bool = True, mark_all_outcomes: bool = True, mark_nash_points: bool = True, mark_kalai_points: bool = True, mark_ks_points: bool = True, mark_max_welfare_points: bool = True, show: bool = True)

Visualize the completed negotiation run showing offers and utilities in 2D space.

This method produces the same visualization as SAOMechanism.plot() but works with saved/completed runs. It requires that the CompletedRun has a scenario with utility functions, and that the history is in ‘full_trace’ format.

Parameters:

• plotting_negotiators – Indices or IDs of two negotiators whose utilities form the axes.

• save_fig – Whether to save the figure to disk.

• path – Directory path for saving the figure.

• fig_name – Filename for the saved figure.

• image_format – Image format to use (default: “webp”). Supported: webp, png, jpg, svg, pdf.

• ignore_none_offers – Whether to skip None offers in the plot.

• with_lines – Whether to connect offers with lines showing progression.

• show_agreement – Whether to highlight the final agreement point.

• show_pareto_distance – Whether to display distance to Pareto frontier.

• show_nash_distance – Whether to display distance to Nash solution.

• show_kalai_distance – Whether to display distance to Kalai-Smorodinsky solution.

• show_ks_distance – Whether to display distance to KS point.

• show_max_welfare_distance – Whether to display distance to max welfare point.

• show_max_relative_welfare_distance – Whether to display distance to max relative welfare.

• show_end_reason – Whether to annotate why the negotiation ended.

• show_annotations – Whether to show offer annotations on the plot.

• show_reserved – Whether to show reservation values.

• show_total_time – Whether to display total negotiation time.

• show_relative_time – Whether to display relative time progress.

• show_n_steps – Whether to display the number of steps.

• colors – Custom color list for negotiators.

• markers – Custom marker list for negotiators.

• colormap – Matplotlib colormap name for coloring offers.

• ylimits – Y-axis limits as (min, max) tuple.

• common_legend – Whether to use a shared legend for all subplots.

• extra_annotation – Additional text to annotate on the plot.

• xdim – Dimension for x-axis (“relative_time”, “step”, “time”).

• colorizer – Optional colorizer function for offers.

• only2d – Whether to show only the 2D utility space plot.

• no2d – Whether to skip the 2D utility space plot.

• fast – Whether to use faster but less detailed rendering.

• simple_offers_view – Whether to use simplified offer visualization.

• mark_offers_view – Whether to mark offers in the utility space view.

• mark_pareto_points – Whether to mark Pareto optimal points.

• mark_all_outcomes – Whether to mark all possible outcomes.

• mark_nash_points – Whether to mark Nash bargaining solution.

• mark_kalai_points – Whether to mark Kalai-Smorodinsky solution.

• mark_ks_points – Whether to mark KS points.

• mark_max_welfare_points – Whether to mark maximum welfare points.

• show – Whether to display the figure immediately.

Returns:

Plotly figure object if show=False, None otherwise.

Raises:

ValueError – If scenario is None, history_type is not ‘full_trace’, or utility functions are missing.

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, include_pareto_frontier: bool = False, storage_format: Literal['csv', 'gzip', 'parquet'] | None = 'parquet') → 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 – Deprecated. This parameter is ignored. Per-negotiator traces are available in the main trace file grouped by negotiator.

• 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.

• include_pareto_frontier – If True the pareto frontier will be included in the scenario (if save_scenario)

• 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:

• initial_state – Initial mechanism state. If None, a default MechanismState will be created.

• outcome_space – The negotiation agenda as an OutcomeSpace object. Use this for complex outcome spaces.

• issues – List of Issue objects defining the negotiation dimensions. Alternative to outcome_space.

• outcomes – List of valid outcomes or an integer specifying the number of outcomes. Alternative to outcome_space.

• n_steps – Maximum number of negotiation rounds/steps. None means unlimited. This is a shared limit visible to all negotiators.

• time_limit – Maximum negotiation duration in seconds. None means unlimited. This is a shared limit visible to all negotiators.

• pend – Probability (0-1) of ending negotiation at each step. 0 means disabled.

• pend_per_second – Probability (0-1) of ending negotiation each second. 0 means disabled.

• step_time_limit – Maximum time in seconds allowed for each negotiation step/round. None means unlimited.

• negotiator_time_limit – Maximum cumulative time in seconds for each negotiator’s responses. None means unlimited.

• hidden_time_limit – Secret time limit not visible to negotiators. Used for testing or forcing timeouts.

• max_n_negotiators – Maximum number of negotiators allowed to join. None means unlimited.

• dynamic_entry – If True, negotiators can join/leave after negotiation starts. If False, all must join before start.

• annotation – Dictionary of arbitrary metadata attached to this mechanism session.

• nmi_factory – Class to use for creating NegotiatorMechanismInterface instances. Defaults to NegotiatorMechanismInterface.

• extra_callbacks – If True, call additional negotiator callbacks like on_round_start/end, on_leave, etc.

• checkpoint_every – Save checkpoint every N steps. Set to <=0 to disable checkpointing.

• checkpoint_folder – Directory path for saving checkpoints. None disables checkpointing.

• checkpoint_filename – Base name for checkpoint files. Step numbers will be prefixed.

• extra_checkpoint_info – Additional data to save in checkpoint metadata as a dictionary.

• single_checkpoint – If True, only keep the most recent checkpoint (overwrite previous ones).

• exist_ok – If True, allow overwriting existing checkpoint files.

• name – Human-readable name for this mechanism session. Auto-generated if not provided.

• genius_port – Port number for Genius bridge connection. Use 0 for automatic port selection.

• id – Unique system-wide identifier. Usually auto-generated; only set explicitly during deserialization.

• type_name – Custom type name for this mechanism. Auto-generated from class name if not provided.

• verbosity – Logging verbosity level (0=quiet, higher=more verbose).

• ignore_negotiator_exceptions – If True, mechanism continues even if negotiators raise exceptions.

Remarks:

• You can specify per-negotiator time limits and step limits when calling add() to add negotiators.

• The mechanism tracks three types of limits: * shared_*: Limits visible to all negotiators (passed here) * private_*: Per-negotiator limits (passed to add()) * internal_*: Effective limits (minimum of shared and all private limits)

• Negotiators see their own relative_time based on their private limits

• History and internal tracking use internal limits (strictest across all negotiators)

abort() → TState

Aborts the negotiation.

add(negotiator: TNegotiator, *, preferences: Preferences | None = None, role: str | None = None, ufun: BaseUtilityFunction | None = None, time_limit: float | None = inf, n_steps: int | None = None, annotation: dict[str, Any] | None = None) → bool | None

Add a 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 – [deprecated] 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.

• time_limit – Per-negotiator time limit in seconds. This creates a private time limit for this negotiator. The negotiator will see nmi.time_limit = min(shared_time_limit, time_limit) and their relative_time will be calculated based on this effective limit. If None or inf, only the shared time limit applies. This allows different negotiators to have different time constraints in the same negotiation.

• n_steps – Per-negotiator step limit. This creates a private step limit for this negotiator. The negotiator will see nmi.n_steps = min(shared_n_steps, n_steps) and their relative_time will be calculated based on this effective limit. If None, only the shared step limit applies. This allows different negotiators to have different step constraints in the same negotiation.

• annotation – Additional metadata to attach to this negotiator’s NMI.

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

Notes

Per-negotiator limits enable scenarios where different negotiators have different time/step constraints:

• A negotiator with time_limit=30 in a mechanism with time_limit=60 will see nmi.time_limit=30

• A negotiator with time_limit=90 in a mechanism with time_limit=60 will see nmi.time_limit=60

• A negotiator with time_limit=None or time_limit=inf will see the shared limit

• The mechanism’s internal tracking uses the strictest limit across all negotiators

• Each negotiator’s relative_time is calculated based on their individual effective limit

This maintains backward compatibility: negotiators added without private limits behave exactly as before.

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

available_save_sources() → list[str]

Returns the available sources for saving history.

Returns:

The available sources for save().

Return type:

list[str]

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.

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.

This uses internal time limits (the strictest across all negotiators).

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.

get_nmi_for(negotiator: TNegotiator) → TNMI

Returns the NMI instance for the given negotiator.

By default, all negotiators share the same NMI instance, but mechanisms can override this to provide different NMIs to different negotiators.

Parameters:

negotiator – The negotiator for whom to get the NMI.

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]

property internal_nmi: TNMI

The Negotiation Mechanism Interface (NMI) with combined deadline information from all negotiators.

Returns:

The NMI instance

Return type:

TNMI

is_satisfying(capabilities: dict) → bool

Check if given capabilities satisfy the mechanism’s requirements.

Parameters:

capabilities – Dictionary of capabilities to check

Returns:

True if capabilities satisfy all requirements, False otherwise

Remarks:

• None requirement value means any capability value is acceptable

• Set/list requirement means capability must match one of the values

• Tuple of 2 numbers is treated as a range [min, max] - capability can be value in range or overlapping range

• For exact equality check after set conversion if both are collections

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 taking into account individual negotiator limits, or None if unlimited.

n_steps_for(nid: str)

Returns the maximum number of negotiation steps allowed as seen by a given negotiator, 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]

property nmi: TNMI

The Negotiation Mechanism Interface (NMI) with shared information available to all negotiators.

Returns:

The NMI instance

Return type:

TNMI

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.

This uses shared time limits (visible to all negotiators).

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: dict

Protocol requirements dictionary.

Returns:

Dictionary mapping requirement names to acceptable values

run(timeout=None, start_callback: Callable[[TState], None] | None = None, progress_callback: Callable[[TState], None] | None = None, completion_callback: Callable[[TState], None] | None = None) → TState

Execute the negotiation mechanism until completion or timeout.

Parameters:

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

• start_callback – Optional callback called once at the start of negotiation with the initial state

• progress_callback – Optional callback called after each negotiation step with the current state

• completion_callback – Optional callback called once at the end of negotiation with the final state

Returns:

Final negotiation state after completion

Return type:

TState

run_with_progress(timeout=None, start_callback: Callable[[TState], None] | None = None, progress_callback: Callable[[TState], None] | None = None, completion_callback: Callable[[TState], None] | None = None) → TState

Run with a progress bar (using rich).

Parameters:

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

• start_callback – Optional callback called once at the start of negotiation with the initial state

• progress_callback – Optional callback called after each negotiation step with the current state

• completion_callback – Optional callback called once at the end of negotiation with the final state

Returns:

The final negotiation state after completion

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, start_callback: Callable[[int, Mechanism], None] | None = None, progress_callback: Callable[[int, Mechanism], None] | 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

• start_callback – Optional callback called once at the start of each mechanism with (negotiation_id, mechanism)

• progress_callback – Optional callback called after each step of each mechanism with (negotiation_id, mechanism)

• completion_callback – Optional callback called once at the end of each mechanism with (negotiation_id, mechanism)

• 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 | None = None, metadata: dict[str, Any] | None = None, agreement_stats: OutcomeOptimality | None = None, calc_agreement_stats: bool = False, overwrite: bool = True, warn_if_existing: bool = True, storage_format: Literal['csv', 'gzip', 'parquet'] | None = 'parquet') → 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. If None (default), auto-detect the best available source in priority order: full_trace_with_utils > full_trace > extended_trace > trace > history.

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

• agreement_stats – Pre-calculated agreement optimality statistics. If provided, these will be saved instead of calculating them (which requires scenario stats).

• calc_agreement_stats – If True and agreement_stats is None, calculate agreement_stats from the scenario. This is expensive as it requires calculating scenario stats.

• 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 shared_n_steps

Returns the maximum number of negotiation steps allowed according to shared information between negotiators, or None if unlimited.

property shared_nmi: TNMI

The Negotiation Mechanism Interface (NMI) with shared information available to all negotiators.

Returns:

The NMI instance

Return type:

TNMI

property shared_time_limit

Returns the maximum negotiation time in seconds according to shared information between negotiators, or infinity if unlimited.

property state: TState

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.

state_for(negotiator: TNegotiator) → TState

Returns a state object for the given negotiator with per-negotiator relative_time.

The state is a copy of the current state with only relative_time modified based on the negotiator’s private time limits.

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 (nanosecond precision).

0.0 if the mechanism did not start running

property time_limit

Returns the maximum negotiation time in seconds taking into account individual negotiator limits, or infinity if unlimited.

time_limit_for(nid: str)

Returns the maximum negotiation time in seconds as seen by a given negotiator, or infinity if unlimited.

to_completed_run(source: str | None = None, metadata: dict[str, Any] | None = None, agreement_stats: OutcomeOptimality | None = None, calc_agreement_stats: bool = False) → CompletedRun

Creates a CompletedRun object from the current mechanism state.

Parameters:

• source –

  The source of the history information. Options: - None: Auto-detect the best available source in priority order:

  full_trace_with_utils > full_trace > extended_trace > trace > history

  • ”history”: Use the history attribute (list of states)

  • ”full_trace”: Use the full_trace attribute (if available)

  • ”full_trace_with_utils”: Use full_trace_with_utils (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.

• agreement_stats – Pre-calculated agreement optimality statistics. If provided, these will be used directly.

• calc_agreement_stats – If True and agreement_stats is None, calculate agreement_stats from the scenario (requires ufuns to be available). This involves calculating scenario stats which can be expensive.

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