Skip to content

ANL 2025 Reference

This package provides a wrapper around NegMAS functionality to generate and run tournaments a la ANL 2025 competition. You mostly only need to use anl2025_tournament in your code. The other helpers are provided to allow for a finer control over the scenarios used.

Negotiators (Agents)

The package provides few example negotiators. Of special importance is the MiCRO negotiator which provides a full implementation of a recently proposed behavioral strategy. Other negotiators are just wrappers over negotiators provided by NegMAS.

anl2025.negotiator.ANL2025Negotiator

Bases: SAOController

Base class of all participant code.

See the next two examples of how to implement it (Boulware2025, Random2025).

Parameters:

Name Type Description Default
n_edges int

Number of edges for this negotiator. You can access it using self._n_edges

 0
update_side_ufuns_on_end bool

Updates the expected outcome for each thread at the end of each negotiation threads. These expected values are used when a side-negotiator calculates its side-utility.

 True
update_side_ufuns_after_receiving_offers bool

Updates expected outcome for each thread whenever an offer is received from a partner

 False
update_side_ufuns_after_offering bool

Updates the expected outcome for each thread after it is offered (i.e. assume the opponent will accept).

 False
auto_kill bool

Removes negotiators from the self.negotiators list whenever the negotiation ends.

 False
Remarks
  • This class provides some useful members that can be used when developing the negotiation strategy:
    • self.ufun : The CenterUFun for the center negotiator (for edge negotiators, it will be a normal negmas ufun).
    • self.negotiators: Returns a dict mapping negotiator-IDs (for side negotiators) to the corresponding negotiator and context. You can use the returned negotiator to access the NMI (Negotiator-Mechanism-Interface) for this side negotiator which in turn have access to the state and other services provided by negmas NMIs. The context has the following members:
    • ufun: The side ufun for the thread.
    • center: A boolean indicating whether this is a side negotiator for a center or is an edge negotiator.
    • index: The thread index for this thread.
    • self.active_negotiators / self.started_negotiators / self.to_start_negotiators / self.finished_negotiators / self.unfinished_negotiators Same as self.negotiators but with the corresponding subset of negotiators only.

Methods:

Name Description
init

Called after all mechanisms are created to initialize
propose

Called to propose an offer for one of the edge negotiators
respond

Called to respond to an offer from an edge negotiator
set_expected_outcome

Sets the expected value for a negotiation thread.
thread_finalize

Called when a negotiation thread ends
thread_init

Called when a negotiation thread starts
Source code in anl2025/negotiator.py 
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
class ANL2025Negotiator(SAOController):
    """
    Base class of all participant code.

    See the next two examples of how to implement it (`Boulware2025`, `Random2025`).

    Args:
        n_edges: Number of edges for this negotiator. You can access it using self._n_edges
        update_side_ufuns_on_end: Updates the expected outcome for each thread at the end of
                                  each negotiation threads. These expected values are used
                                  when a side-negotiator calculates its side-utility.
        update_side_ufuns_after_receiving_offers: Updates expected outcome for each thread whenever
                                                  an offer is received from a partner
        update_side_ufuns_after_offering: Updates the expected outcome for each thread after it is
                                          offered (i.e. assume the opponent will accept).
        auto_kill: Removes negotiators from the self.negotiators list whenever the negotiation ends.

    Remarks:
        - This class provides some useful members that can be used when developing the negotiation strategy:
            - `self.ufun` : The `CenterUFun` for the center negotiator (for edge negotiators, it will be a normal negmas ufun).
            - `self.negotiators`: Returns a dict mapping negotiator-IDs (for side negotiators) to the corresponding negotiator
                                  and context. You can use the returned negotiator to access the `NMI` (Negotiator-Mechanism-Interface)
                                  for this side negotiator which in turn have access to the state and other services provided by
                                  negmas NMIs. The context has the following members:
               - `ufun`: The side ufun for the thread.
               - `center`: A boolean indicating whether this is a side negotiator for a center or is an edge negotiator.
               - `index`: The thread index for this thread.
            - `self.active_negotiators` / `self.started_negotiators` / `self.to_start_negotiators` / `self.finished_negotiators` / `self.unfinished_negotiators`
              Same as `self.negotiators` but with the corresponding subset of negotiators only.
    """

    def __init__(
        self,
        *args,
        n_edges: int = 0,
        update_side_ufuns_on_end: bool = True,
        update_side_ufuns_after_offering: bool = False,
        update_side_ufuns_after_receiving_offers: bool = False,
        auto_kill: bool = False,
        **kwargs,
    ):
        super().__init__(*args, auto_kill=auto_kill, **kwargs)
        self._n_edges = n_edges
        self._update_side_ufuns_on_end = update_side_ufuns_on_end
        self._update_side_ufuns_after_offering = update_side_ufuns_after_offering
        self._update_side_ufuns_after_receiving_offers = (
            update_side_ufuns_after_receiving_offers
        )

    def init(self):
        """Called after all mechanisms are created to initialize

        Remarks:
            - self.negotiators can be used to access the threads.
            - Each has a negotiator object and a cntxt object.
            - We can pass anything in the cntxt. Currently, we pass the side ufun
            - Examples:
                1. Access the CenterUFun associated with the agent. For edge agents, this will be the single ufun it uses.
                    my_ufun = self.ufun
                2. Access the side ufun associated with each thread. For edge agents this will be the single ufun it uses.
                    my_side_ufuns = [info.context["ufun"] for neg_id, info in self.negotiators.items()]
                    my_side_indices = [info.context["index"] for neg_id, info in self.negotiators.items()]
                    my_side_is_center = [info.context["center"] for neg_id, info in self.negotiators.items()]
                2. Access the side negotiators connected to different negotiation threads
                    my_side_negotiators = [info.negotiator for neg_id, info in self.negotiators.items()]
        """

    def propose(
        self, negotiator_id: str, state: SAOState, dest: str | None = None
    ) -> Outcome | ExtendedOutcome | None:
        """Called to propose an offer for one of the edge negotiators

        Args:
            negotiator_id: The ID of the connection to the edge negotiator.
            state: The state of the negotiation with this edge negotiator.
            dest: The ID of the edge negotiator


        Returns:
            An outcome to offer. In ANL2025, `None` and `ExtendedOutcome` are not allowed
        """
        return super().propose(negotiator_id, state, dest)

    def respond(
        self, negotiator_id: str, state: SAOState, source: str | None = None
    ) -> ResponseType:
        """Called to respond to an offer from an edge negotiator

        Args:
            negotiator_id: The ID of the connection to the edge negotiator.
            state: The state of the negotiation with this edge negotiator.
            dest: The ID of the edge negotiator

        Returns:
            A response (Accept, Reject, or End_Negotiation)

        Remarks:
            - The current offer on the negotiation thread with this edge
              negotiator can be accessed as `state.current_offer`.
        """
        return super().respond(negotiator_id, state, source)

    def set_expected_outcome(self, negotiator_id: str, outcome: Outcome | None) -> None:
        """
        Sets the expected value for a negotiation thread.
        """
        if not isinstance(self.ufun, CenterUFun):
            return
        _, cntxt = self.negotiators[negotiator_id]
        index = cntxt["index"]
        self.ufun.set_expected_outcome(index, outcome)

    def thread_init(self, negotiator_id: str, state: SAOState) -> None:
        """Called when a negotiation thread starts

        Args:
            negotiator_id: Connection ID to this negotiation thread
            state: The state of the negotiation thread at the start of the negotiation.
        """

    def thread_finalize(self, negotiator_id: str, state: SAOState) -> None:
        """Called when a negotiation thread ends

        Args:
            negotiator_id: Connection ID to this negotiation thread
            state: The state of the negotiation thread at the end of the negotiation.
        """

    def on_negotiation_start(self, negotiator_id: str, state: SAOState) -> None:
        super().on_negotiation_start(negotiator_id, state)
        self.thread_init(negotiator_id, state)

    def on_negotiation_end(self, negotiator_id: str, state: SAOState) -> None:
        if self._update_side_ufuns_on_end:
            self.set_expected_outcome(negotiator_id, state.agreement)
        super().on_negotiation_end(negotiator_id, state)
        self.thread_finalize(negotiator_id, state)

init

Called after all mechanisms are created to initialize

Remarks
  • self.negotiators can be used to access the threads.
  • Each has a negotiator object and a cntxt object.
  • We can pass anything in the cntxt. Currently, we pass the side ufun
  • Examples:
    1. Access the CenterUFun associated with the agent. For edge agents, this will be the single ufun it uses. my_ufun = self.ufun
    2. Access the side ufun associated with each thread. For edge agents this will be the single ufun it uses. my_side_ufuns = [info.context["ufun"] for neg_id, info in self.negotiators.items()] my_side_indices = [info.context["index"] for neg_id, info in self.negotiators.items()] my_side_is_center = [info.context["center"] for neg_id, info in self.negotiators.items()]
    3. Access the side negotiators connected to different negotiation threads my_side_negotiators = [info.negotiator for neg_id, info in self.negotiators.items()]
Source code in anl2025/negotiator.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
def init(self):
    """Called after all mechanisms are created to initialize

    Remarks:
        - self.negotiators can be used to access the threads.
        - Each has a negotiator object and a cntxt object.
        - We can pass anything in the cntxt. Currently, we pass the side ufun
        - Examples:
            1. Access the CenterUFun associated with the agent. For edge agents, this will be the single ufun it uses.
                my_ufun = self.ufun
            2. Access the side ufun associated with each thread. For edge agents this will be the single ufun it uses.
                my_side_ufuns = [info.context["ufun"] for neg_id, info in self.negotiators.items()]
                my_side_indices = [info.context["index"] for neg_id, info in self.negotiators.items()]
                my_side_is_center = [info.context["center"] for neg_id, info in self.negotiators.items()]
            2. Access the side negotiators connected to different negotiation threads
                my_side_negotiators = [info.negotiator for neg_id, info in self.negotiators.items()]
    """

propose

Called to propose an offer for one of the edge negotiators

Parameters:

Name Type Description Default
negotiator_id str

The ID of the connection to the edge negotiator.

 required
state SAOState

The state of the negotiation with this edge negotiator.

 required
dest str | None

The ID of the edge negotiator

 None

Returns:

Type Description
Outcome | ExtendedOutcome | None

An outcome to offer. In ANL2025, None and ExtendedOutcome are not allowed
Source code in anl2025/negotiator.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
def propose(
    self, negotiator_id: str, state: SAOState, dest: str | None = None
) -> Outcome | ExtendedOutcome | None:
    """Called to propose an offer for one of the edge negotiators

    Args:
        negotiator_id: The ID of the connection to the edge negotiator.
        state: The state of the negotiation with this edge negotiator.
        dest: The ID of the edge negotiator


    Returns:
        An outcome to offer. In ANL2025, `None` and `ExtendedOutcome` are not allowed
    """
    return super().propose(negotiator_id, state, dest)

respond

Called to respond to an offer from an edge negotiator

Parameters:

Name Type Description Default
negotiator_id str

The ID of the connection to the edge negotiator.

 required
state SAOState

The state of the negotiation with this edge negotiator.

 required
dest

The ID of the edge negotiator

 required

Returns:

Type Description
ResponseType

A response (Accept, Reject, or End_Negotiation)
Remarks
  • The current offer on the negotiation thread with this edge negotiator can be accessed as state.current_offer.
Source code in anl2025/negotiator.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
def respond(
    self, negotiator_id: str, state: SAOState, source: str | None = None
) -> ResponseType:
    """Called to respond to an offer from an edge negotiator

    Args:
        negotiator_id: The ID of the connection to the edge negotiator.
        state: The state of the negotiation with this edge negotiator.
        dest: The ID of the edge negotiator

    Returns:
        A response (Accept, Reject, or End_Negotiation)

    Remarks:
        - The current offer on the negotiation thread with this edge
          negotiator can be accessed as `state.current_offer`.
    """
    return super().respond(negotiator_id, state, source)

set_expected_outcome

Sets the expected value for a negotiation thread.

Source code in anl2025/negotiator.py 
145
146
147
148
149
150
151
152
153
def set_expected_outcome(self, negotiator_id: str, outcome: Outcome | None) -> None:
    """
    Sets the expected value for a negotiation thread.
    """
    if not isinstance(self.ufun, CenterUFun):
        return
    _, cntxt = self.negotiators[negotiator_id]
    index = cntxt["index"]
    self.ufun.set_expected_outcome(index, outcome)

thread_finalize

Called when a negotiation thread ends

Parameters:

Name Type Description Default
negotiator_id str

Connection ID to this negotiation thread

 required
state SAOState

The state of the negotiation thread at the end of the negotiation.

 required
Source code in anl2025/negotiator.py 
163
164
165
166
167
168
169
def thread_finalize(self, negotiator_id: str, state: SAOState) -> None:
    """Called when a negotiation thread ends

    Args:
        negotiator_id: Connection ID to this negotiation thread
        state: The state of the negotiation thread at the end of the negotiation.
    """

thread_init

Called when a negotiation thread starts

Parameters:

Name Type Description Default
negotiator_id str

Connection ID to this negotiation thread

 required
state SAOState

The state of the negotiation thread at the start of the negotiation.

 required
Source code in anl2025/negotiator.py 
155
156
157
158
159
160
161
def thread_init(self, negotiator_id: str, state: SAOState) -> None:
    """Called when a negotiation thread starts

    Args:
        negotiator_id: Connection ID to this negotiation thread
        state: The state of the negotiation thread at the start of the negotiation.
    """

anl2025.negotiator.Random2025

Bases: ANL2025Negotiator

The most general way to implement an agent is to implement propose and respond.

Methods:

Name Description
propose

Proposes to the given partner (dest) using the side negotiator (negotiator_id).
respond

Responds to the given partner (source) using the side negotiator (negotiator_id).
Source code in anl2025/negotiator.py 
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
class Random2025(ANL2025Negotiator):
    """
    The most general way to implement an agent is to implement propose and respond.
    """

    p_end = 0.000003
    p_reject = 0.99

    def propose(
        self, negotiator_id: str, state: SAOState, dest: str | None = None
    ) -> Outcome | None:
        """Proposes to the given partner (dest) using the side negotiator (negotiator_id).

        Remarks:
        """
        nmi = self.negotiators[negotiator_id].negotiator.nmi
        os: DiscreteCartesianOutcomeSpace = nmi.outcome_space
        return list(os.sample(1))[0]

    def respond(
        self, negotiator_id: str, state: SAOState, source: str | None = None
    ) -> ResponseType:
        """Responds to the given partner (source) using the side negotiator (negotiator_id).

        Remarks:
            - source: is the ID of the partner.
            - the mapping from negotiator_id to source is stable within a negotiation.

        """

        if random() < self.p_end:
            return ResponseType.END_NEGOTIATION

        if (
            random() < self.p_reject
            or float(self.ufun(state.current_offer)) < self.ufun(None)  # type: ignore
        ):
            return ResponseType.REJECT_OFFER
        return ResponseType.ACCEPT_OFFER

propose

Proposes to the given partner (dest) using the side negotiator (negotiator_id).

Remarks:

Source code in anl2025/negotiator.py 
241
242
243
244
245
246
247
248
249
250
def propose(
    self, negotiator_id: str, state: SAOState, dest: str | None = None
) -> Outcome | None:
    """Proposes to the given partner (dest) using the side negotiator (negotiator_id).

    Remarks:
    """
    nmi = self.negotiators[negotiator_id].negotiator.nmi
    os: DiscreteCartesianOutcomeSpace = nmi.outcome_space
    return list(os.sample(1))[0]

respond

Responds to the given partner (source) using the side negotiator (negotiator_id).

Remarks
  • source: is the ID of the partner.
  • the mapping from negotiator_id to source is stable within a negotiation.
Source code in anl2025/negotiator.py 
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
def respond(
    self, negotiator_id: str, state: SAOState, source: str | None = None
) -> ResponseType:
    """Responds to the given partner (source) using the side negotiator (negotiator_id).

    Remarks:
        - source: is the ID of the partner.
        - the mapping from negotiator_id to source is stable within a negotiation.

    """

    if random() < self.p_end:
        return ResponseType.END_NEGOTIATION

    if (
        random() < self.p_reject
        or float(self.ufun(state.current_offer)) < self.ufun(None)  # type: ignore
    ):
        return ResponseType.REJECT_OFFER
    return ResponseType.ACCEPT_OFFER

anl2025.negotiator.TimeBased2025

Bases: ANL2025Negotiator

A time-based conceding agent

Methods:

Name Description
ensure_inverter

Ensures that utility inverter is available
propose

Proposes to the given partner (dest) using the side negotiator (negotiator_id).
respond

Responds to the given partner (source) using the side negotiator (negotiator_id).
Source code in anl2025/negotiator.py 
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
class TimeBased2025(ANL2025Negotiator):
    """
    A time-based conceding agent
    """

    def __init__(
        self,
        *args,
        aspiration_type: Literal["boulware"]
        | Literal["conceder"]
        | Literal["linear"]
        | Literal["hardheaded"]
        | float = "boulware",
        deltas: tuple[float, ...] = (1e-3, 1e-1, 2e-1, 4e-1, 8e-1, 1.0),
        reject_exactly_as_reserved: bool = False,
        **kwargs,
    ):
        super().__init__(*args, **kwargs)
        self._curve = PolyAspiration(1.0, aspiration_type)
        self._inverter: dict[str, InverseUFun] = dict()
        self._best: list[Outcome] = None  # type: ignore
        self._mx: float = 1.0
        self._mn: float = 0.0
        self._deltas = deltas
        self._best_margin = 1e-8
        self.reject_exactly_as_reserved = reject_exactly_as_reserved

    def ensure_inverter(self, negotiator_id) -> InverseUFun:
        """Ensures that utility inverter is available"""
        if self._inverter.get(negotiator_id, None) is None:
            _, cntxt = self.negotiators[negotiator_id]
            ufun = cntxt["ufun"]
            inverter = PresortingInverseUtilityFunction(ufun, rational_only=True)
            inverter.init()
            # breakpoint()
            self._mx, self._mn = inverter.max(), inverter.min()
            self._mn = max(self._mn, ufun(None))
            self._best = inverter.some(
                (
                    max(0.0, self._mn, ufun(None), self._mx - self._best_margin),
                    self._mx,
                ),
                normalized=True,
            )
            if not self._best:
                self._best = [inverter.best()]  # type: ignore
            self._inverter[negotiator_id] = inverter

        return self._inverter[negotiator_id]

    def calc_level(self, nmi: SAONMI, state: SAOState, normalized: bool):
        if state.step == 0:
            level = 1.0
        elif (
            # not self.reject_exactly_as_reserved
            # and
            nmi.n_steps is not None and state.step >= nmi.n_steps - 1
        ):
            level = 0.0
        else:
            level = self._curve.utility_at(state.relative_time)
        if not normalized:
            level = level * (self._mx - self._mn) + self._mn
        return level

    def propose(
        self, negotiator_id: str, state: SAOState, dest: str | None = None
    ) -> Outcome | None:
        """Proposes to the given partner (dest) using the side negotiator (negotiator_id).

        Remarks:
        """
        assert self.ufun
        negotiator, cntxt = self.negotiators[negotiator_id]
        inverter = self.ensure_inverter(negotiator_id)
        nmi = negotiator.nmi
        level = self.calc_level(nmi, state, normalized=True)
        ufun: SideUFun = cntxt["ufun"]
        outcome = None
        if self._mx < float(ufun(None)):
            return None
        for d in self._deltas:
            mx = min(1.0, level + d)
            outcome = inverter.one_in((level, mx), normalized=True)
            # print(f"{self.id} found {outcome} at level {(level, mx)}")
            if outcome:
                break
        if not outcome:
            return choice(self._best)
        return outcome

    def respond(
        self, negotiator_id: str, state: SAOState, source: str | None = None
    ) -> ResponseType:
        """Responds to the given partner (source) using the side negotiator (negotiator_id).

        Remarks:
            - source: is the ID of the partner.
            - the mapping from negotiator_id to source is stable within a negotiation.

        """
        assert self.ufun
        _, cntxt = self.negotiators[negotiator_id]
        ufun: SideUFun = cntxt["ufun"]
        nmi = self.negotiators[negotiator_id][0].nmi
        self.ensure_inverter(negotiator_id)
        # end the negotiation if there are no rational outcomes
        level = self.calc_level(nmi, state, normalized=False)

        if self._mx < float(ufun(None)):
            return ResponseType.END_NEGOTIATION

        # print(f"{self.id} got {ufun(state.current_offer)} at level {level}")
        if (self.reject_exactly_as_reserved and level >= ufun(state.current_offer)) or (
            not self.reject_exactly_as_reserved and level > ufun(state.current_offer)
        ):
            return ResponseType.REJECT_OFFER
        return ResponseType.ACCEPT_OFFER

    def thread_finalize(self, negotiator_id: str, state: SAOState) -> None:
        for side in self.negotiators.keys():
            if side == negotiator_id:
                continue
            if side in self._inverter:
                del self._inverter[side]

ensure_inverter

Ensures that utility inverter is available

Source code in anl2025/negotiator.py 
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
def ensure_inverter(self, negotiator_id) -> InverseUFun:
    """Ensures that utility inverter is available"""
    if self._inverter.get(negotiator_id, None) is None:
        _, cntxt = self.negotiators[negotiator_id]
        ufun = cntxt["ufun"]
        inverter = PresortingInverseUtilityFunction(ufun, rational_only=True)
        inverter.init()
        # breakpoint()
        self._mx, self._mn = inverter.max(), inverter.min()
        self._mn = max(self._mn, ufun(None))
        self._best = inverter.some(
            (
                max(0.0, self._mn, ufun(None), self._mx - self._best_margin),
                self._mx,
            ),
            normalized=True,
        )
        if not self._best:
            self._best = [inverter.best()]  # type: ignore
        self._inverter[negotiator_id] = inverter

    return self._inverter[negotiator_id]

propose

Proposes to the given partner (dest) using the side negotiator (negotiator_id).

Remarks:

Source code in anl2025/negotiator.py 
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
def propose(
    self, negotiator_id: str, state: SAOState, dest: str | None = None
) -> Outcome | None:
    """Proposes to the given partner (dest) using the side negotiator (negotiator_id).

    Remarks:
    """
    assert self.ufun
    negotiator, cntxt = self.negotiators[negotiator_id]
    inverter = self.ensure_inverter(negotiator_id)
    nmi = negotiator.nmi
    level = self.calc_level(nmi, state, normalized=True)
    ufun: SideUFun = cntxt["ufun"]
    outcome = None
    if self._mx < float(ufun(None)):
        return None
    for d in self._deltas:
        mx = min(1.0, level + d)
        outcome = inverter.one_in((level, mx), normalized=True)
        # print(f"{self.id} found {outcome} at level {(level, mx)}")
        if outcome:
            break
    if not outcome:
        return choice(self._best)
    return outcome

respond

Responds to the given partner (source) using the side negotiator (negotiator_id).

Remarks
  • source: is the ID of the partner.
  • the mapping from negotiator_id to source is stable within a negotiation.
Source code in anl2025/negotiator.py 
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
def respond(
    self, negotiator_id: str, state: SAOState, source: str | None = None
) -> ResponseType:
    """Responds to the given partner (source) using the side negotiator (negotiator_id).

    Remarks:
        - source: is the ID of the partner.
        - the mapping from negotiator_id to source is stable within a negotiation.

    """
    assert self.ufun
    _, cntxt = self.negotiators[negotiator_id]
    ufun: SideUFun = cntxt["ufun"]
    nmi = self.negotiators[negotiator_id][0].nmi
    self.ensure_inverter(negotiator_id)
    # end the negotiation if there are no rational outcomes
    level = self.calc_level(nmi, state, normalized=False)

    if self._mx < float(ufun(None)):
        return ResponseType.END_NEGOTIATION

    # print(f"{self.id} got {ufun(state.current_offer)} at level {level}")
    if (self.reject_exactly_as_reserved and level >= ufun(state.current_offer)) or (
        not self.reject_exactly_as_reserved and level > ufun(state.current_offer)
    ):
        return ResponseType.REJECT_OFFER
    return ResponseType.ACCEPT_OFFER

anl2025.negotiator.Boulware2025

Bases: TimeBased2025

Source code in anl2025/negotiator.py 
401
402
403
404
405
class Boulware2025(TimeBased2025):
    def __init__(
        self, *args, deltas: tuple[float, ...] = (0.1, 0.2, 0.4, 0.8, 1), **kwargs
    ):
        super().__init__(*args, aspiration_type="boulware", deltas=deltas, **kwargs)

anl2025.negotiator.Linear2025

Bases: TimeBased2025

Source code in anl2025/negotiator.py 
408
409
410
411
412
class Linear2025(TimeBased2025):
    def __init__(
        self, *args, deltas: tuple[float, ...] = (0.1, 0.2, 0.4, 0.8, 1), **kwargs
    ):
        super().__init__(*args, aspiration_type="linear", deltas=deltas, **kwargs)

anl2025.negotiator.Conceder2025

Bases: TimeBased2025

Source code in anl2025/negotiator.py 
415
416
417
418
419
class Conceder2025(TimeBased2025):
    def __init__(
        self, *args, deltas: tuple[float, ...] = (0.1, 0.2, 0.4, 0.8, 1), **kwargs
    ):
        super().__init__(*args, aspiration_type="conceder", deltas=deltas, **kwargs)

anl2025.negotiator.IndependentBoulware2025

Bases: ANL2025Negotiator

You can participate by an agent that runs any SAO negotiator independently for each thread.

Source code in anl2025/negotiator.py 
182
183
184
185
186
187
188
189
class IndependentBoulware2025(ANL2025Negotiator):
    """
    You can participate by an agent that runs any SAO negotiator independently for each thread.
    """

    def __init__(self, **kwargs):
        kwargs["default_negotiator_type"] = AspirationNegotiator
        super().__init__(**kwargs)

anl2025.negotiator.IndependentLinear2025

Bases: ANL2025Negotiator

You can participate by an agent that runs any SAO negotiator independently for each thread.

Source code in anl2025/negotiator.py 
192
193
194
195
196
197
198
199
class IndependentLinear2025(ANL2025Negotiator):
    """
    You can participate by an agent that runs any SAO negotiator independently for each thread.
    """

    def __init__(self, **kwargs):
        kwargs["default_negotiator_type"] = LinearTBNegotiator
        super().__init__(**kwargs)

anl2025.negotiator.IndependentConceder2025

Bases: ANL2025Negotiator

You can participate by an agent that runs any SAO negotiator independently for each thread.

Source code in anl2025/negotiator.py 
202
203
204
205
206
207
208
209
class IndependentConceder2025(ANL2025Negotiator):
    """
    You can participate by an agent that runs any SAO negotiator independently for each thread.
    """

    def __init__(self, **kwargs):
        kwargs["default_negotiator_type"] = ConcederTBNegotiator
        super().__init__(**kwargs)

Utility Functions

anl2025.ufun.CenterUFun

Bases: UtilityFunction, ABC

Base class of center utility functions.

Remarks
  • Can be constructed by either passing a single outcome_space and n_edges or a tuple of outcome_spaces
  • It's eval() method receives a tuple of negotiation results and returns a float

Methods:

Name Description
__call__

Entry point to calculate the utility of a set of offers (called by the mechanism).
eval

Evaluates the utility of a given set of offers.
eval_with_expected

Calculates the utility of a set of offers with control over whether or not to use stored expected outcomes.
side_ufuns

Should return an independent ufun for each side negotiator of the center.
ufun_type

Returns the center ufun category.
Source code in anl2025/ufun.py 
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
class CenterUFun(UtilityFunction, ABC):
    """
    Base class of center utility functions.

    Remarks:
        - Can be constructed by either passing a single `outcome_space` and `n_edges` or a tuple of `outcome_spaces`
        - It's eval() method  receives a tuple of negotiation results and returns a float
    """

    def __init__(
        self,
        *args,
        outcome_spaces: tuple[CartesianOutcomeSpace, ...] = (),
        n_edges: int = 0,
        combiner_type: type[OSCombiner] = DefaultCombiner,
        expected_outcomes: tuple[Outcome | None, ...]
        | list[Outcome | None]
        | None = None,
        stationary: bool = True,
        stationary_sides: bool | None = None,
        side_ufuns: tuple[BaseUtilityFunction | None, ...] | None = None,
        **kwargs,
    ):
        super().__init__(*args, **kwargs)
        if not outcome_spaces and self.outcome_space:
            outcome_spaces = tuple([self.outcome_space] * n_edges)
        self._combiner = combiner_type(outcome_spaces)
        self._outcome_spaces = self._combiner.separated_spaces()
        self.n_edges = len(outcome_spaces)
        self.stationary = stationary
        self.stationary_sides = (
            stationary_sides if stationary_sides is not None else stationary
        )
        self.__kwargs = dict(
            reserved_value=self.reserved_value,
            owner=self.owner,
            invalid_value=self._invalid_value,
            name=self.name,
            id=self.id,
        )
        self.__nissues = _calc_n_issues(self._outcome_spaces)
        try:
            self.outcome_space = self._combiner.combined_space()
        except Exception:
            warn("Failed to find the Cartesian product of input outcome spaces")
            self.outcome_space, self.__nissues = None, tuple()
        self._expected: list[Outcome | None] = (
            list(expected_outcomes)
            if expected_outcomes
            else ([None for _ in range(self.n_edges)])
        )

        if side_ufuns is None:
            ufuns = tuple(None for _ in range(self.n_edges))
        else:
            ufuns = tuple(deepcopy(_) for _ in side_ufuns)
        self._effective_side_ufuns = tuple(
            make_side_ufun(self, i, side) for i, side in enumerate(ufuns)
        )

    def is_stationary(self) -> bool:
        return self.stationary

    def set_expected_outcome(self, index: int, outcome: Outcome | None) -> None:
        # print(f"Setting expected outcome for {index} to {outcome}")
        self._expected[index] = outcome
        # print(f"{self._expected}")

    @property
    def outcome_spaces(self) -> tuple[OutcomeSpace, ...]:
        return self._outcome_spaces

    def eval_with_expected(self, offer: Outcome | None, use_expected=True) -> float:
        """Calculates the utility of a set of offers with control over whether or not to use stored expected outcomes."""
        outcomes = self._combiner.separated_outcomes(offer)
        if outcomes:
            if use_expected:
                outcomes = tuple(
                    outcome if outcome else expected
                    for outcome, expected in zip(outcomes, self._expected, strict=True)
                )
            if all(_ is None for _ in outcomes):
                outcomes = None
        if outcomes is None:
            return self.reserved_value
        return self.eval(outcomes)

    def __call__(self, offer: Outcome | None, use_expected=True) -> float:
        """Entry point to calculate the utility of a set of offers (called by the mechanism).

        Override to avoid using expected outcomes."""

        return self.eval_with_expected(offer, use_expected=use_expected)

    @abstractmethod
    def eval(self, offer: tuple[Outcome | None, ...] | Outcome | None) -> float:
        """
        Evaluates the utility of a given set of offers.

        Remarks:
            - Order matters: The order of outcomes in the offer is stable over all calls.
            - A missing offer is represented by `None`
        """

    @abstractmethod
    def ufun_type(self) -> CenterUFunCategory:
        """Returns the center ufun category.

        Currently, we have two categories (Global and Local). See `CenterUFunCategory` for
        their definitions.
        """
        ...

    def side_ufuns(self) -> tuple[BaseUtilityFunction | None, ...]:
        """Should return an independent ufun for each side negotiator of the center."""
        ufuns = self._effective_side_ufuns
        # Make sure that these side ufuns are connected to self
        for i, u in enumerate(ufuns):
            if id(u._center_ufun) == id(self):
                continue
            u._center_ufun = self
            u._index = i
            u._n_edges = self.n_edges
        return ufuns

    def to_dict(self, python_class_identifier=TYPE_IDENTIFIER) -> dict[str, Any]:
        return {
            python_class_identifier: get_full_type_name(type(self)),
            "outcome_spaces": serialize(
                self._outcome_spaces, python_class_identifier=python_class_identifier
            ),
            "name": self.name,
            "reserved_value": self.reserved_value,
        }

    @classmethod
    def from_dict(cls, d, python_class_identifier=TYPE_IDENTIFIER):
        d.pop(python_class_identifier, None)
        for f in ("outcome_spaces", "ufuns"):
            if f in d:
                d[f] = deserialize(
                    d[f], python_class_identifier=python_class_identifier
                )
        return cls(**d)

__call__

Entry point to calculate the utility of a set of offers (called by the mechanism).

Override to avoid using expected outcomes.

Source code in anl2025/ufun.py 
405
406
407
408
409
410
def __call__(self, offer: Outcome | None, use_expected=True) -> float:
    """Entry point to calculate the utility of a set of offers (called by the mechanism).

    Override to avoid using expected outcomes."""

    return self.eval_with_expected(offer, use_expected=use_expected)

eval abstractmethod

Evaluates the utility of a given set of offers.

Remarks
  • Order matters: The order of outcomes in the offer is stable over all calls.
  • A missing offer is represented by None
Source code in anl2025/ufun.py 
412
413
414
415
416
417
418
419
420
@abstractmethod
def eval(self, offer: tuple[Outcome | None, ...] | Outcome | None) -> float:
    """
    Evaluates the utility of a given set of offers.

    Remarks:
        - Order matters: The order of outcomes in the offer is stable over all calls.
        - A missing offer is represented by `None`
    """

eval_with_expected

Calculates the utility of a set of offers with control over whether or not to use stored expected outcomes.

Source code in anl2025/ufun.py 
390
391
392
393
394
395
396
397
398
399
400
401
402
403
def eval_with_expected(self, offer: Outcome | None, use_expected=True) -> float:
    """Calculates the utility of a set of offers with control over whether or not to use stored expected outcomes."""
    outcomes = self._combiner.separated_outcomes(offer)
    if outcomes:
        if use_expected:
            outcomes = tuple(
                outcome if outcome else expected
                for outcome, expected in zip(outcomes, self._expected, strict=True)
            )
        if all(_ is None for _ in outcomes):
            outcomes = None
    if outcomes is None:
        return self.reserved_value
    return self.eval(outcomes)

side_ufuns

Should return an independent ufun for each side negotiator of the center.

Source code in anl2025/ufun.py 
431
432
433
434
435
436
437
438
439
440
441
def side_ufuns(self) -> tuple[BaseUtilityFunction | None, ...]:
    """Should return an independent ufun for each side negotiator of the center."""
    ufuns = self._effective_side_ufuns
    # Make sure that these side ufuns are connected to self
    for i, u in enumerate(ufuns):
        if id(u._center_ufun) == id(self):
            continue
        u._center_ufun = self
        u._index = i
        u._n_edges = self.n_edges
    return ufuns

ufun_type abstractmethod

Returns the center ufun category.

Currently, we have two categories (Global and Local). See CenterUFunCategory for their definitions.

Source code in anl2025/ufun.py 
422
423
424
425
426
427
428
429
@abstractmethod
def ufun_type(self) -> CenterUFunCategory:
    """Returns the center ufun category.

    Currently, we have two categories (Global and Local). See `CenterUFunCategory` for
    their definitions.
    """
    ...

anl2025.ufun.FlatCenterUFun

Bases: UtilityFunction

A flattened version of a center ufun.

A normal CenterUFun takes outcomes as a tuple of outcomes (one for each edge). A flattened version of the same ufun takes input as just a single outcome containing a concatenation of the outcomes in all edges.

Example:

```python
x = CenterUFun(...)
y = x.flatten()

x(((1, 0.5), (3, true), (7,))) == y((1, 0.5, 3, true, 7))
```
Source code in anl2025/ufun.py 
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
class FlatCenterUFun(UtilityFunction):
    """
    A flattened version of a center ufun.

    A normal CenterUFun takes outcomes as a tuple of outcomes (one for each edge).
    A flattened version of the same ufun takes input as just a single outcome containing
    a concatenation of the outcomes in all edges.

    Example:

        ```python
        x = CenterUFun(...)
        y = x.flatten()

        x(((1, 0.5), (3, true), (7,))) == y((1, 0.5, 3, true, 7))
        ```
    """

    def __init__(
        self, *args, base_ufun: CenterUFun, nissues: tuple[int, ...], **kwargs
    ):
        super().__init__(*args, **kwargs)
        self.__base = base_ufun
        self.__nissues = list(nissues)

    def _unflatten(
        self, outcome: Outcome | tuple[Outcome | None, ...] | None
    ) -> tuple[Outcome | None, ...] | None:
        if outcome is None:
            return None

        if isinstance(outcome, tuple) and isinstance(outcome[0], Outcome):
            return outcome

        beg = [0] + self.__nissues[:-1]
        end = self.__nissues
        outcomes = []
        for i, j in zip(beg, end, strict=True):
            x = outcome[i:j]
            if all(_ is None for _ in x):
                outcomes.append(None)
            else:
                outcomes.append(tuple(x))
        return tuple(outcomes)

    def eval(self, offer: Outcome | tuple[Outcome | None] | None) -> float:
        return self.__base.eval(self._unflatten(offer))

anl2025.ufun.LambdaCenterUFun

Bases: CenterUFun

A center utility function that implements an arbitrary evaluator

Source code in anl2025/ufun.py 
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
class LambdaCenterUFun(CenterUFun):
    """
    A center utility function that implements an arbitrary evaluator
    """

    def __init__(self, *args, evaluator: CenterEvaluator, **kwargs):
        super().__init__(*args, **kwargs)
        self._evaluator = evaluator

    def eval(self, offer: tuple[Outcome | None, ...] | Outcome | None) -> float:
        return self._evaluator(self._combiner.separated_outcomes(offer))

    def ufun_type(self) -> CenterUFunCategory:
        return CenterUFunCategory.Global

    def to_dict(self, python_class_identifier=TYPE_IDENTIFIER) -> dict[str, Any]:
        return super().to_dict(python_class_identifier) | dict(
            evaluator=serialize(
                self._evaluator, python_class_identifier=python_class_identifier
            )
        )

anl2025.ufun.LambdaUtilityFunction

Bases: UtilityFunction

A utility function that implements an arbitrary mapping

Source code in anl2025/ufun.py 
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
class LambdaUtilityFunction(UtilityFunction):
    """A utility function that implements an arbitrary mapping"""

    def __init__(self, *args, evaluator: EdgeEvaluator, **kwargs):
        super().__init__(*args, **kwargs)
        self._evaluator = evaluator

    def __call__(self, offer: Outcome | None) -> float:
        return self._evaluator(offer)

    def eval(self, offer: Outcome) -> float:
        return self._evaluator(offer)

    def to_dict(self, python_class_identifier=TYPE_IDENTIFIER) -> dict[str, Any]:
        return super().to_dict(python_class_identifier) | dict(
            evaluator=serialize(
                self._evaluator, python_class_identifier=python_class_identifier
            )
        )

anl2025.ufun.MaxCenterUFun

Bases: UtilityCombiningCenterUFun

The max center ufun.

The utility of the center is the maximum of the utilities it got in each negotiation (called side utilities)

Source code in anl2025/ufun.py 
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
class MaxCenterUFun(UtilityCombiningCenterUFun):
    """
    The max center ufun.

    The utility of the center is the maximum of the utilities it got in each negotiation (called side utilities)
    """

    def set_expected_outcome(self, index: int, outcome: Outcome | None) -> None:
        # sets the reserved value of all sides
        super().set_expected_outcome(index, outcome)
        r = None
        try:
            set_ufun = self._effective_side_ufuns[index]
        except Exception:
            return

        if isinstance(set_ufun, SideUFunAdapter):
            r = float(set_ufun._base_ufun(outcome))
        elif isinstance(set_ufun, SideUFun):
            return
        if r is None:
            return
        for i, side in enumerate(self._effective_side_ufuns):
            if side is None or i == index:
                continue
            side.reserved_value = max(side.reserved_value, r)
            if isinstance(side, SideUFunAdapter):
                side._base_ufun.reserved_value = max(side._base_ufun.reserved_value, r)

    def combine(self, values: Sequence[float]) -> float:
        return max(values)

anl2025.ufun.MeanSMCenterUFun

Bases: SingleAgreementSideUFunMixin, CenterUFun

A ufun that just returns the average mean+std dev. in each issue of the agreements as the utility value

Source code in anl2025/ufun.py 
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
class MeanSMCenterUFun(SingleAgreementSideUFunMixin, CenterUFun):
    """A ufun that just  returns the average mean+std dev. in each issue of the agreements as the utility value"""

    def eval(self, offer: tuple[Outcome | None, ...] | Outcome | None) -> float:
        offer = self._combiner.separated_outcomes(offer)
        if not offer:
            return 0.0
        n_edges = len(offer)
        if n_edges < 2:
            return 0.1
        vals = defaultdict(lambda: [0.0] * n_edges)
        for e, outcome in enumerate(offer):
            if not outcome:
                continue
            for i, v in enumerate(outcome):
                try:
                    vals[e][i] = float(v[1:])
                except Exception:
                    pass

        return float(sum(np.mean(x) + np.std(x) for x in vals.values())) / len(vals)

anl2025.ufun.SideUFun

Bases: BaseUtilityFunction

Side ufun corresponding to the i's component of a center ufun.

Source code in anl2025/ufun.py 
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
class SideUFun(BaseUtilityFunction):
    """
    Side ufun corresponding to the i's component of a center ufun.
    """

    def __init__(
        self,
        *args,
        center_ufun: CenterUFun,
        index: int,
        n_edges: int,
        **kwargs,
    ):
        super().__init__(*args, **kwargs)
        self._center_ufun = center_ufun
        self._index = index
        self._n_edges = n_edges

    def is_stationary(self) -> bool:
        return self._center_ufun.stationary_sides

    def set_expected_outcome(
        self, outcome: Outcome | None, index: int | None = None
    ) -> None:
        if index is None:
            index = self._index
        self._center_ufun.set_expected_outcome(index, outcome)

    def eval(self, offer: Outcome | None) -> float:
        exp_outcome = [_ for _ in self._center_ufun._expected]
        offers = [_ for _ in self._center_ufun._expected]
        offers[self._index] = offer

        # For all offers that are after the current agent, set to None:
        for i in range(self._index + 1, self._n_edges):
            # if offers[i] is not None:
            # print(
            #     f"{self._index=}: {i} has a none None outcome ({offers[i]=})\n{offers=}"
            # )
            # raise AssertionError(
            #     f"{self._index=}: {i} has a none None outcome ({offers[i]=})\n{offers=}"
            # )
            self.set_expected_outcome(None, i)
            offers[i] = None
        u = self._center_ufun(tuple(offers))

        # restore expected_outcome:
        for i in range(self._index + 1, self._n_edges):
            self.set_expected_outcome(exp_outcome[i], i)

        return u

anl2025.ufun.SingleAgreementSideUFunMixin

Can be mixed with any CenterUFun that is not a combining ufun to create side_ufuns that assume failure on all other negotiations.

See Also

MeanSMCenterUFun

Source code in anl2025/ufun.py 
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
class SingleAgreementSideUFunMixin:
    """Can be mixed with any CenterUFun that is not a combining ufun to create side_ufuns that assume failure on all other negotiations.

    See Also:
        `MeanSMCenterUFun`
    """

    # def side_ufuns(self) -> tuple[BaseUtilityFunction, ...]:
    #     """Should return an independent ufun for each side negotiator of the center"""
    #     return tuple(
    #         SideUFun(center_ufun=self, n_edges=self.n_edges, index=i)  # type: ignore
    #         for i in range(self.n_edges)  # type: ignore
    #     )

    def ufun_type(self) -> CenterUFunCategory:
        return CenterUFunCategory.Local

anl2025.ufun.UtilityCombiningCenterUFun

Bases: CenterUFun

A center ufun with a side-ufun defined for each thread.

The utility of the center is a function of the ufuns of the edges.

Methods:

Name Description
combine

Combines the utilities of all negotiation threads into a single value
Source code in anl2025/ufun.py 
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
class UtilityCombiningCenterUFun(CenterUFun):
    """
    A center ufun with a side-ufun defined for each thread.

    The utility of the center is a function of the ufuns of the edges.
    """

    def __init__(self, *args, side_ufuns: tuple[BaseUtilityFunction, ...], **kwargs):
        super().__init__(*args, **kwargs)
        self.ufuns = tuple(deepcopy(_) for _ in side_ufuns)
        # This is already done in CenterUFun now
        # self._effective_side_ufuns = tuple(
        #     make_side_ufun(self, i, side) for i, side in enumerate(self.ufuns)
        # )

    @abstractmethod
    def combine(self, values: Sequence[float]) -> float:
        """Combines the utilities of all negotiation  threads into a single value"""

    def eval(self, offer: tuple[Outcome | None, ...] | Outcome | None) -> float:
        offer = self._combiner.separated_outcomes(offer)
        if not offer:
            return self.reserved_value
        return self.combine(tuple(float(u(_)) for u, _ in zip(self.ufuns, offer)))

    def ufun_type(self) -> CenterUFunCategory:
        return CenterUFunCategory.Local

    def to_dict(self, python_class_identifier=TYPE_IDENTIFIER) -> dict[str, Any]:
        return super().to_dict(python_class_identifier) | {
            "side_ufuns": serialize(
                self.ufuns, python_class_identifier=python_class_identifier
            ),
            python_class_identifier: get_full_type_name(type(self)),
        }

combine abstractmethod

Combines the utilities of all negotiation threads into a single value

Source code in anl2025/ufun.py 
700
701
702
@abstractmethod
def combine(self, values: Sequence[float]) -> float:
    """Combines the utilities of all negotiation  threads into a single value"""

Utility Function Helpers

anl2025.ufun.convert_to_center_ufun

Creates a center ufun from any standard ufun with ufuns side ufuns

Source code in anl2025/ufun.py 
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
def convert_to_center_ufun(
    ufun: UtilityFunction,
    nissues: tuple[int],
    combiner_type: type[OSCombiner] = DefaultCombiner,
    side_evaluators: list[EdgeEvaluator] | None = None,
) -> "CenterUFun":
    """Creates a center ufun from any standard ufun with ufuns side ufuns"""
    assert ufun.outcome_space and isinstance(ufun.outcome_space, CartesianOutcomeSpace)
    evaluator = ufun
    if side_evaluators is not None:
        return LocalEvaluationCenterUFun(
            outcome_spaces=unflatten_outcome_space(ufun.outcome_space, nissues),
            evaluator=evaluator,
            side_evaluators=tuple(side_evaluators),
            combiner_type=combiner_type,
        )
    return LambdaCenterUFun(
        outcome_spaces=unflatten_outcome_space(ufun.outcome_space, nissues),
        evaluator=evaluator,
        combiner_type=combiner_type,
    )

anl2025.ufun.unflatten_outcome_space

Distributes the issues of an outcome-space into a tuple of outcome-spaces.

Source code in anl2025/ufun.py 
109
110
111
112
113
114
115
116
117
118
119
def unflatten_outcome_space(
    outcome_space: CartesianOutcomeSpace, nissues: tuple[int, ...] | list[int]
) -> tuple[CartesianOutcomeSpace, ...]:
    """Distributes the issues of an outcome-space into a tuple of outcome-spaces."""
    nissues = list(nissues)
    beg = [0] + nissues[:-1]
    end = nissues
    return tuple(
        make_os(outcome_space.issues[i:j], name=f"OS{i}")
        for i, j in zip(beg, end, strict=True)
    )

Scenarios

anl2025.scenario.MultidealScenario

Defines the multi-deal scenario by setting utility functions (and implicitly outcome-spaces)

Methods:

Name Description
from_folder

Loads a multi-deal scenario from the given folder.
to_dict

Converts the scenario to a dictionary
Source code in anl2025/scenario.py 
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
@define
class MultidealScenario:
    """Defines the multi-deal scenario by setting utility functions (and implicitly outcome-spaces)"""

    center_ufun: CenterUFun
    edge_ufuns: tuple[UtilityFunction, ...]
    side_ufuns: tuple[UtilityFunction, ...] | None = None
    name: str = ""
    public_graph: bool = True
    code_files: dict[str, str] = field(factory=dict)

    def __attrs_post_init__(self):
        if self.public_graph:
            for e in self.edge_ufuns:
                e.n_edges = self.center_ufun.n_edges  # type: ignore
                e.oucome_spaces = [  # type: ignore
                    copy.deepcopy(_) for _ in self.center_ufun.outcome_spaces
                ]

    def save(
        self,
        base: Path | str,
        as_folder: bool = True,
        python_class_identifier: str = TYPE_IDENTIFIER,
    ):
        base = Path(base)
        name = self.name if self.name else "scenario"
        if as_folder:
            return self.to_folder(
                folder=base / name,
                mkdir=False,
                python_class_identifier=python_class_identifier,
            )
        return self.to_file(
            path=base / f"{name}.yml", python_class_identifier=python_class_identifier
        )

    @classmethod
    def from_folder(
        cls,
        folder: Path | str,
        name: str | None = None,
        public_graph: bool = True,
        python_class_identifier: str = TYPE_IDENTIFIER,
        type_marker=f"{TYPE_IDENTIFIER}:",
    ) -> Optional["MultidealScenario"]:
        """
        Loads a multi-deal scenario from the given folder.

        Args:
            folder: The path to load the scenario from
            name: The name to give to the scenario. If not given, the folder name
            edges_know_details: If given, edge ufuns will have `n_edges`, `outcome_spaces` members
                                that reveal the number of edges in total and the outcome space for each
                                negotiation thread.
            python_class_identifier: the key in the yaml to define a type.
            type_marker: A marker at the beginning of a string to define a type (for future proofing).
        """
        folder = Path(folder)
        folder = folder.resolve()
        center_file = folder / CENTER_FILE_NAME
        if not center_file.is_file():
            return None
        code_files = dict()
        for f in folder.glob("**/*.py"):
            path = str(f.relative_to(folder))
            with open(f, "r") as file:
                code_files[path] = file.read()
        added_paths = []
        if code_files:
            for code_file in code_files.keys():
                module = str(Path(code_file).parent)
                try:
                    del module
                except Exception:
                    pass
            _added_path = str(folder.resolve())
            added_paths.append(_added_path)
            sys.path.insert(0, _added_path)
        dparams = dict(
            python_class_identifier=python_class_identifier,
            type_marker=type_marker,
            type_name_adapter=type_name_adapter,
        )
        center_ufun = deserialize(load(center_file), **dparams)  # type: ignore
        assert isinstance(
            center_ufun, CenterUFun
        ), f"{type(center_ufun)} but expected a CenterUFun \n{center_ufun}"

        def load_ufuns(f: Path | str) -> tuple[UtilityFunction, ...] | None:
            f = Path(f)
            if not f.is_dir():
                return None
            return tuple(
                deserialize(load(_), **dparams)  # type: ignore
                for _ in f.glob("*.yml")
            )

        edge_ufuns = load_ufuns(folder / EDGES_FOLDER_NAME)
        assert edge_ufuns
        for u, os in zip(edge_ufuns, center_ufun.outcome_spaces):
            if u.outcome_space is None:
                u.outcome_space = os
                if isinstance(u, SideUFunAdapter):
                    u._base_ufun.outcome_space = os
        if public_graph:
            for u in edge_ufuns:
                u.n_edges = center_ufun.n_edges  # type: ignore
                u.outcome_spaces = tuple(  # type: ignore
                    copy.deepcopy(_) for _ in center_ufun.outcome_spaces
                )
        side_ufuns = load_ufuns(folder / SIDES_FOLDER_NAME)
        for p in added_paths:
            sys.path.remove(p)

        return cls(
            center_ufun=center_ufun,
            edge_ufuns=tuple(edge_ufuns),
            side_ufuns=side_ufuns,
            name=folder.name if name is None else name,
            code_files=code_files,
        )

    def to_folder(
        self,
        folder: Path | str,
        python_class_identifier: str = TYPE_IDENTIFIER,
        mkdir: bool = False,
    ):
        folder = Path(folder)
        if mkdir:
            name = self.name if self.name else "scenario"
            folder = folder / name
        folder.mkdir(parents=True, exist_ok=True)
        for path, code in self.code_files.items():
            full_path = folder / path
            full_path.parent.mkdir(exist_ok=True, parents=True)
            with open(full_path, "w") as f:
                f.write(code)

        dump(
            serialize(
                self.center_ufun, python_class_identifier=python_class_identifier
            ),
            folder / CENTER_FILE_NAME,
        )

        def save(fname, ufuns):
            if ufuns is None:
                return
            base = folder / fname
            base.mkdir(parents=True, exist_ok=True)
            for u in ufuns:
                dump(
                    serialize(u, python_class_identifier=python_class_identifier),
                    base / f"{u.name}.yml",
                )

        save(EDGES_FOLDER_NAME, self.edge_ufuns)
        save(SIDES_FOLDER_NAME, self.side_ufuns)

    def to_dict(self, python_class_identifier=TYPE_IDENTIFIER) -> dict[str, Any]:
        """Converts the scenario to a dictionary"""
        return dict(
            name=self.name,
            center_ufun=serialize(
                self.center_ufun, python_class_identifier=python_class_identifier
            ),
            edge_ufuns=serialize(
                self.edge_ufuns, python_class_identifier=python_class_identifier
            ),
            side_ufuns=serialize(
                self.side_ufuns, python_class_identifier=python_class_identifier
            ),
        )

    @classmethod
    def from_dict(
        cls, d: dict[str, Any], python_class_identifier=TYPE_IDENTIFIER
    ) -> Optional["MultidealScenario"]:
        return deserialize(d, python_class_identifier=python_class_identifier)  # type: ignore

    def to_file(self, path: Path | str, python_class_identifier=TYPE_IDENTIFIER):
        dump(self.to_dict(python_class_identifier=python_class_identifier), path)

    @classmethod
    def from_file(
        cls, path: Path | str, python_class_identifier=TYPE_IDENTIFIER
    ) -> Optional["MultidealScenario"]:
        return cls.to_dict(load(path), python_class_identifier=python_class_identifier)  # type: ignore

from_folder classmethod

Loads a multi-deal scenario from the given folder.

Parameters:

Name Type Description Default
folder Path | str

The path to load the scenario from

 required
name str | None

The name to give to the scenario. If not given, the folder name

 None
edges_know_details

If given, edge ufuns will have n_edges, outcome_spaces members that reveal the number of edges in total and the outcome space for each negotiation thread.

 required
python_class_identifier str

the key in the yaml to define a type.

 TYPE_IDENTIFIER
type_marker

A marker at the beginning of a string to define a type (for future proofing).

 f'{TYPE_IDENTIFIER}:'
Source code in anl2025/scenario.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
@classmethod
def from_folder(
    cls,
    folder: Path | str,
    name: str | None = None,
    public_graph: bool = True,
    python_class_identifier: str = TYPE_IDENTIFIER,
    type_marker=f"{TYPE_IDENTIFIER}:",
) -> Optional["MultidealScenario"]:
    """
    Loads a multi-deal scenario from the given folder.

    Args:
        folder: The path to load the scenario from
        name: The name to give to the scenario. If not given, the folder name
        edges_know_details: If given, edge ufuns will have `n_edges`, `outcome_spaces` members
                            that reveal the number of edges in total and the outcome space for each
                            negotiation thread.
        python_class_identifier: the key in the yaml to define a type.
        type_marker: A marker at the beginning of a string to define a type (for future proofing).
    """
    folder = Path(folder)
    folder = folder.resolve()
    center_file = folder / CENTER_FILE_NAME
    if not center_file.is_file():
        return None
    code_files = dict()
    for f in folder.glob("**/*.py"):
        path = str(f.relative_to(folder))
        with open(f, "r") as file:
            code_files[path] = file.read()
    added_paths = []
    if code_files:
        for code_file in code_files.keys():
            module = str(Path(code_file).parent)
            try:
                del module
            except Exception:
                pass
        _added_path = str(folder.resolve())
        added_paths.append(_added_path)
        sys.path.insert(0, _added_path)
    dparams = dict(
        python_class_identifier=python_class_identifier,
        type_marker=type_marker,
        type_name_adapter=type_name_adapter,
    )
    center_ufun = deserialize(load(center_file), **dparams)  # type: ignore
    assert isinstance(
        center_ufun, CenterUFun
    ), f"{type(center_ufun)} but expected a CenterUFun \n{center_ufun}"

    def load_ufuns(f: Path | str) -> tuple[UtilityFunction, ...] | None:
        f = Path(f)
        if not f.is_dir():
            return None
        return tuple(
            deserialize(load(_), **dparams)  # type: ignore
            for _ in f.glob("*.yml")
        )

    edge_ufuns = load_ufuns(folder / EDGES_FOLDER_NAME)
    assert edge_ufuns
    for u, os in zip(edge_ufuns, center_ufun.outcome_spaces):
        if u.outcome_space is None:
            u.outcome_space = os
            if isinstance(u, SideUFunAdapter):
                u._base_ufun.outcome_space = os
    if public_graph:
        for u in edge_ufuns:
            u.n_edges = center_ufun.n_edges  # type: ignore
            u.outcome_spaces = tuple(  # type: ignore
                copy.deepcopy(_) for _ in center_ufun.outcome_spaces
            )
    side_ufuns = load_ufuns(folder / SIDES_FOLDER_NAME)
    for p in added_paths:
        sys.path.remove(p)

    return cls(
        center_ufun=center_ufun,
        edge_ufuns=tuple(edge_ufuns),
        side_ufuns=side_ufuns,
        name=folder.name if name is None else name,
        code_files=code_files,
    )

to_dict

Converts the scenario to a dictionary

Source code in anl2025/scenario.py 
210
211
212
213
214
215
216
217
218
219
220
221
222
223
def to_dict(self, python_class_identifier=TYPE_IDENTIFIER) -> dict[str, Any]:
    """Converts the scenario to a dictionary"""
    return dict(
        name=self.name,
        center_ufun=serialize(
            self.center_ufun, python_class_identifier=python_class_identifier
        ),
        edge_ufuns=serialize(
            self.edge_ufuns, python_class_identifier=python_class_identifier
        ),
        side_ufuns=serialize(
            self.side_ufuns, python_class_identifier=python_class_identifier
        ),
    )

anl2025.scenario.make_multideal_scenario

Source code in anl2025/scenario.py 
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
def make_multideal_scenario(
    nedges: int = 5,
    nissues: int = 3,
    nvalues: int = 7,
    # edge ufuns
    center_reserved_value_min: float = 0.0,
    center_reserved_value_max: float = 0.0,
    center_ufun_type: str | type[CenterUFun] = "LinearCombinationCenterUFun",
    center_ufun_params: dict[str, Any] | None = None,
    # edge ufuns
    edge_reserved_value_min: float = 0.1,
    edge_reserved_value_max: float = 0.4,
) -> MultidealScenario:
    ufuns = [generate_multi_issue_ufuns(nissues, nvalues) for _ in range(nedges)]
    edge_ufuns = [_[0] for _ in ufuns]
    for u in edge_ufuns:
        u.reserved_value = sample_between(
            edge_reserved_value_min, edge_reserved_value_max
        )
    # side ufuns are utilities of the center on individual threads (may or may not be used, see next comment)
    side_ufuns = tuple(_[1] for _ in ufuns)
    # create center ufun using side-ufuns if possible and without them otherwise.
    center_r = sample_between(center_reserved_value_min, center_reserved_value_max)
    utype = get_ufun_class(center_ufun_type)
    center_ufun_params = center_ufun_params if center_ufun_params else dict()
    try:
        center_ufun = utype(
            side_ufuns=side_ufuns,
            reserved_value=center_r,
            outcome_spaces=tuple(u.outcome_space for u in side_ufuns),  # type: ignore
            **center_ufun_params,
        )
    except TypeError:
        try:
            center_ufun = utype(
                ufuns=side_ufuns,
                reserved_value=center_r,
                outcome_spaces=tuple(u.outcome_space for u in side_ufuns),  # type: ignore
                **center_ufun_params,
            )
        except TypeError:
            # if the center ufun does not take `ufuns` as an input, do not pass it
            center_ufun = utype(
                reserved_value=center_r,
                outcome_spaces=tuple(u.outcome_space for u in side_ufuns),  # type: ignore
                **center_ufun_params,
            )

    return MultidealScenario(
        center_ufun=center_ufun,
        side_ufuns=side_ufuns,
        edge_ufuns=tuple(edge_ufuns),
    )

anl2025.scenarios.make_dinners_scenario

Creates a variation of the Dinners multideal scenario

Parameters:

Name Type Description Default
n_friends int

Number of friends.

 3
n_days int | None

Number of days. If None, it will be the same as n_friends

 None
friend_names tuple[str, ...] | None

Optionally, list of friend names (otherwise we will use Friend{i})

 None
center_reserved_value tuple[float, float] | float

The reserved value of the center. Either a number of a min-max range

 0.0
edge_reserved_valus

The reserved value of the friends (edges). Always, a min-max range

 required
values dict[tuple[int, ...], float] | None

A mapping from the number of dinners per day (a tuple of n_days integers) to utility value of the center

 None
public_graph bool

Should edges know n_edges and outcome_spaces?

 True

Returns:

Type Description
MultidealScenario

An initialized MultidealScenario.
Source code in anl2025/scenarios/dinners.py 
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
def make_dinners_scenario(
    n_friends: int = 3,
    n_days: int | None = None,
    friend_names: tuple[str, ...] | None = None,
    center_reserved_value: tuple[float, float] | float = 0.0,
    edge_reserved_values: tuple[float, float] = (0.0, 0.5),
    values: dict[tuple[int, ...], float] | None = None,
    public_graph: bool = True,
) -> MultidealScenario:
    """Creates a variation of the Dinners multideal scenario

    Args:
        n_friends: Number of friends.
        n_days: Number of days. If `None`, it will be the same as `n_friends`
        friend_names: Optionally, list of friend names (otherwise we will use Friend{i})
        center_reserved_value: The reserved value of the center. Either a number of a min-max range
        edge_reserved_valus: The reserved value of the friends (edges). Always, a min-max range
        values: A mapping from the number of dinners per day (a tuple of n_days integers) to utility value of the center
        public_graph: Should edges know n_edges and outcome_spaces?

    Returns:
        An initialized `MultidealScenario`.
    """
    if n_days is None:
        n_days = n_friends
    if not friend_names:
        friend_names = tuple(f"Friend{i+1}" for i in range(n_friends))
    assert (
        len(friend_names) == n_friends
    ), f"You passed {len(friend_names)} friend names but {n_friends=}"
    outcome_spaces = [
        make_os([make_issue(n_days, name="Day")], name=f"{name}Day")
        for name in friend_names
    ]
    if not isinstance(center_reserved_value, Iterable):
        r = float(center_reserved_value)
    else:
        r = center_reserved_value[0] + random.random() * (
            center_reserved_value[-1] - center_reserved_value[0]
        )
    return MultidealScenario(
        name="dinners",
        edge_ufuns=tuple(
            LinearUtilityAggregationFunction.random(
                os, reserved_value=edge_reserved_values, normalized=True
            )
            for os in outcome_spaces
        ),
        center_ufun=LambdaCenterUFun(
            outcome_spaces=outcome_spaces,
            evaluator=DinnersEvaluator(
                reserved_value=r,
                n_days=n_days,
                n_friends=n_friends,
                values=values,
            ),
            reserved_value=r,
        ),
        public_graph=public_graph,
    )

anl2025.scenarios.make_target_quantity_scenario

Creates a target-quantity type scenario

Parameters:

Name Type Description Default
n_suppliers IntRange

Number of suppliers (sources/sellers)

 4
quantity IntRange

The range of values for each supplier (if an integer, then the range is (0, quanityt-1))

 (1, 5)
target_quantity IntRange

The target quantity of the collector (buyer)

 (2, 20)
shortfall_penalty FloatRange

The collector's penalty for buying an item less than target. Can be a range to sample form it

 (0.1, 0.3)
excess_penalty FloatRange

The penalty for buying an item more than target. Can be a range to sample form it

 (0.1, 0.4)
public_graph bool

Whether edges (suppliers) know n_edges and outcome-spaces of the center (collector)

 True
supplier_names list[str] | None

Names of suppliers

 None
collector_name str

Name of the collector

 'Collector'
collector_reserved_value FloatRange

Range or a single value for collector's reserved value

 0.0
supplier_reserved_values FloatRange

Range or a single value for supplier's reserved value

 0.0
supplier_shortfall_penalty FloatRange | None

suppliers' penalty for buying an item less than their target. Can be a range to sample form it

 (0.3, 0.9)
supplier_excess_penalty FloatRange | None

suppliers' penalty for buying an item more than their target. Can be a range to sample form it

 (0.3, 0.9)
Remarks
  • Supplier's target value is sampled uniformly from the range of values
Source code in anl2025/scenarios/target_quantity.py 
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
def make_target_quantity_scenario(
    n_suppliers: IntRange = 4,
    quantity: IntRange = (1, 5),
    target_quantity: IntRange = (2, 20),
    shortfall_penalty: FloatRange = (0.1, 0.3),
    excess_penalty: FloatRange = (0.1, 0.4),
    public_graph: bool = True,
    supplier_names: list[str] | None = None,
    collector_name: str = "Collector",
    collector_reserved_value: FloatRange = 0.0,
    supplier_reserved_values: FloatRange = 0.0,
    supplier_shortfall_penalty: FloatRange | None = (0.3, 0.9),
    supplier_excess_penalty: FloatRange | None = (0.3, 0.9),
) -> MultidealScenario:
    """Creates a target-quantity type scenario

    Args:
        n_suppliers: Number of suppliers (sources/sellers)
        quantity: The range of values for each supplier (if an integer, then the range is (0, quanityt-1))
        target_quantity: The target quantity of the collector (buyer)
        shortfall_penalty: The collector's penalty for buying an item less than target. Can be a range to sample form it
        excess_penalty: The penalty for buying an item more than target. Can be a range to sample form it
        public_graph: Whether edges (suppliers) know n_edges and outcome-spaces of the center (collector)
        supplier_names: Names of suppliers
        collector_name: Name of the collector
        collector_reserved_value: Range or a single value for collector's reserved value
        supplier_reserved_values: Range or a single value for supplier's reserved value
        supplier_shortfall_penalty: suppliers' penalty for buying an item less than their target. Can be a range to sample form it
        supplier_excess_penalty: suppliers' penalty for buying an item more than their target. Can be a range to sample form it

    Remarks:
        - Supplier's target value is sampled uniformly from the range of values
    """
    n_suppliers = int_in(n_suppliers)
    if supplier_shortfall_penalty is None:
        supplier_shortfall_penalty = shortfall_penalty
    if supplier_excess_penalty is None:
        supplier_excess_penalty = excess_penalty
    if supplier_names is None:
        supplier_names = [f"supplier{_+1:02}" for _ in range(n_suppliers)]
    os = make_os([make_issue(quantity, name="quantity")], name="TargetQantity")
    assert isinstance(os, DiscreteCartesianOutcomeSpace)
    quantities = list(os.issues[0].all)
    if isinstance(os.issues[0], ContiguousIssue):
        totals = list(range(n_suppliers * os.issues[0].max_value + 1))
    else:
        totals = sorted(
            list(set([sum(_) for _ in product(*([[0] + quantities] * n_suppliers))]))
        )
    center_ufun = LambdaCenterUFun(
        n_edges=n_suppliers,
        outcome_space=os,
        evaluator=TargetEvaluator(
            values=make_values(
                totals, int_in(target_quantity), shortfall_penalty, excess_penalty
            )
        ),
        name=collector_name,
        reserved_value=float_in(collector_reserved_value),
    )
    edge_ufuns = tuple(
        LinearAdditiveUtilityFunction(
            outcome_space=os,
            reserved_value=float_in(supplier_reserved_values),
            values=(
                TableFun(
                    make_values(
                        quantities,
                        int_in((os.issues[0].min_value, os.issues[0].max_value)),
                        float_in(supplier_shortfall_penalty),
                        float_in(supplier_excess_penalty),
                    )
                ),
            ),
            weights=(1.0,),
            name=name,
        )
        for name in supplier_names
    )

    return MultidealScenario(center_ufun, edge_ufuns, public_graph=public_graph)

anl2025.scenarios.make_job_hunt_scenario

Source code in anl2025/scenarios/job_hunt.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
def make_job_hunt_scenario(
    n_employers: int = 4,
    work_days: tuple[int, int] | int = 6,
    salary: tuple[int, int] | list[int] = [100, 150, 200, 250, 300],
    public_graph: bool = True,
    employer_names: list[str] | None = None,
    employee_name: str = "Employee",
):
    if employer_names is None:
        employer_names = [f"employer{_+1:02}" for _ in range(n_employers)]
    os = make_os(
        [make_issue(work_days, name="days"), make_issue(salary, name="salary")],
        name="JobHunt",
    )
    assert isinstance(os, DiscreteCartesianOutcomeSpace)
    ufun_pairs = [
        generate_ufuns_for(os, ufun_names=(f"with_{ename}", f"{ename}"))
        for ename in employer_names
    ]
    side_ufuns = tuple(_[0] for _ in ufun_pairs)
    edge_ufuns = tuple(_[1] for _ in ufun_pairs)
    center_ufun = MaxCenterUFun(
        side_ufuns=side_ufuns,
        n_edges=n_employers,
        outcome_space=os,
        reserved_value=0.0,
        name=employee_name,
    )

    return MultidealScenario(
        center_ufun, edge_ufuns, side_ufuns=side_ufuns, public_graph=public_graph
    )

anl2025.scenarios.get_example_scenario_names

Returns a list of example scenario names.

Source code in anl2025/scenarios/__init__.py 
41
42
43
def get_example_scenario_names() -> list[str]:
    """Returns a list of example scenario names."""
    return [_.name for _ in _base_path().glob("*") if _.is_dir()]

anl2025.scenarios.load_example_scenario

Loads an example scenario.

Remarks
  • Currently the following scenarios are available: Dinners and TargetQuantity.
  • If you do not pass any name a randomly chosen example scenario will be returned.
Source code in anl2025/scenarios/__init__.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
def load_example_scenario(name: str | None = None) -> MultidealScenario:
    """Loads an example scenario.

    Remarks:
        - Currently the following scenarios are available: Dinners and TargetQuantity.
        - If you do not pass any name a randomly chosen example scenario will be returned.
    """
    if not name:
        paths = [_.resolve() for _ in _base_path().glob("*") if _.is_dir()]
    else:
        paths = [_base_path() / name]
    s = MultidealScenario.from_folder(choice(paths))
    assert s is not None
    return s

Sessions

anl2025.runner.run_session

Runs a multideal negotiation session and runs it.

Parameters:

Name Type Description Default
scenario MultidealScenario

The negotiation scenario (must be MultidealScenario).

 required
method

the method to use for running all the sessions. Acceptable options are: sequential, ordered, threads, processes. See negmas.mechanisms.Mechanism.run_all() for full details.

 DEFAULT_METHOD
center_type str | type[ANL2025Negotiator]

Type of the center agent

 'Boulware2025'
center_params dict[str, Any] | None

Optional parameters to pass to the center agent.

 None
center_ufun_type

Type of the center agent ufun.

 required
center_ufun_params

Parameters to pass to the center agent ufun.

 required
edge_types list[str | type[ANL2025Negotiator]]

Types of edge agents

 [Boulware2025, Linear2025, Conceder2025]
nsteps int

Number of negotiation steps.

 100
keep_order bool

Keep the order of edges when advancing the negotiation.

 True
share_ufuns bool

If given, agents will have access to partner ufuns through self.opponent_ufun.

 True
atomic bool

If given, one step corresponds to one offer instead of a full round.

 False
output Path | None

Folder to store the logs and results within.

 home() / 'negmas' / 'anl2025' / 'session'
name str

Name of the session

 ''
dry bool

IF true, nothing will be run.

 False
verbose bool

Print progress

 False
sample_edges bool

If true, the edge_types will be used as a pool to sample from otherwise edges will be of the types defined by edge_types in order

 False

Returns:

Type Description
SessionResults

SessionResults giving the results of the multideal negotiation session.
Source code in anl2025/runner.py 
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
def run_session(
    scenario: MultidealScenario,
    # center
    center_type: str | type[ANL2025Negotiator] = "Boulware2025",
    center_params: dict[str, Any] | None = None,
    # edges
    edge_types: list[str | type[ANL2025Negotiator]] = [
        Boulware2025,
        Linear2025,
        Conceder2025,
    ],
    # mechanism params
    nsteps: int = 100,
    keep_order: bool = True,
    share_ufuns: bool = True,
    atomic: bool = False,
    # output and logging
    output: Path | None = Path.home() / "negmas" / "anl2025" / "session",
    name: str = "",
    dry: bool = False,
    method=DEFAULT_METHOD,
    verbose: bool = False,
    sample_edges: bool = False,
) -> SessionResults:
    """Runs a multideal negotiation session and runs it.

    Args:
        scenario: The negotiation scenario (must be `MultidealScenario`).
        method: the method to use for running all the sessions.
                Acceptable options are: sequential, ordered, threads, processes.
                See `negmas.mechanisms.Mechanism.run_all()` for full details.
        center_type: Type of the center agent
        center_params: Optional parameters to pass to the center agent.
        center_ufun_type: Type of the center agent ufun.
        center_ufun_params: Parameters to pass to the center agent ufun.
        edge_types: Types of edge agents
        nsteps: Number of negotiation steps.
        keep_order: Keep the order of edges when advancing the negotiation.
        share_ufuns: If given, agents will have access to partner ufuns through `self.opponent_ufun`.
        atomic: If given, one step corresponds to one offer instead of a full round.
        output: Folder to store the logs and results within.
        name: Name of the session
        dry: IF true, nothing will be run.
        verbose: Print progress
        sample_edges: If true, the `edge_types` will be used as a pool to sample from otherwise edges will
                      be of the types defined by edge_types in order

    Returns:
        `SessionResults` giving the results of the multideal negotiation session.
    """
    if output and isinstance(output, str):
        output = Path(output)
    run_params = RunParams(
        nsteps=nsteps,
        keep_order=keep_order,
        share_ufuns=share_ufuns,
        atomic=atomic,
        method=method,
    )
    # if not sample_edges:
    #     assert (
    #         len(edge_types) == len(scenario.edge_ufuns)
    #     ), f"You are trying to run a session without sampling, but the number of provided edge types ({len(edge_types)}) is not equal to the number of edge ufuns in the scenario ({len(scenario.edge_ufuns)})."

    assigned = assign_scenario(
        scenario=scenario,
        run_params=run_params,
        center_type=center_type,
        center_params=center_params,
        edge_types=edge_types,
        verbose=verbose,
        sample_edges=sample_edges,
    )
    return assigned.run(output=output, name=name, dry=dry, verbose=verbose)

anl2025.runner.run_generated_session

Generates a multideal negotiation session and runs it.

Parameters:

Name Type Description Default
method

the method to use for running all the sessions. Acceptable options are: sequential, ordered, threads, processes. See negmas.mechanisms.Mechanism.run_all() for full details.

 DEFAULT_METHOD
center_type str

Type of the center agent

 'Boulware2025'
center_params dict[str, Any] | None

Optional parameters to pass to the center agent.

 None
center_reserved_value_min float

Minimum reserved value for the center agent.

 0.0
center_reserved_value_max float

Maximum reserved value for the center agent.

 0.0
center_ufun_type str | type[CenterUFun]

Type of the center agent ufun.

 'MaxCenterUFun'
center_ufun_params dict[str, Any] | None

Parameters to pass to the center agent ufun.

 None
nedges int

Number of edges.

 10
edge_reserved_value_min float

Minimum reserved value for edges.

 0.1
edge_reserved_value_max float

Maximum reserved value for edges.

 0.4
edge_types list[str | type[ANL2025Negotiator]]

Types of edge agents

 [Boulware2025, Random2025, Linear2025, Conceder2025]
nissues int

Number of issues to use for each thread.

 3
nvalues int

Number of values per issue for each thread.

 7
nsteps int

Number of negotiation steps.

 100
keep_order bool

Keep the order of edges when advancing the negotiation.

 True
share_ufuns bool

If given, agents will have access to partner ufuns through self.opponent_ufun.

 True
atomic bool

If given, one step corresponds to one offer instead of a full round.

 False
output Path | str | None

Folder to store the logs and results within.

 home() / 'negmas' / 'anl2025' / 'session'
name str

Name of the session

 ''
dry bool

IF true, nothing will be run.

 False
verbose bool

Print progress

 False

Returns:

Type Description
SessionResults

SessionResults giving the results of the multideal negotiation session.
Source code in anl2025/runner.py 
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
def run_generated_session(
    # center
    center_type: str = "Boulware2025",
    center_params: dict[str, Any] | None = None,
    center_reserved_value_min: float = 0.0,
    center_reserved_value_max: float = 0.0,
    center_ufun_type: str | type[CenterUFun] = "MaxCenterUFun",
    center_ufun_params: dict[str, Any] | None = None,
    # edges
    nedges: int = 10,
    edge_reserved_value_min: float = 0.1,
    edge_reserved_value_max: float = 0.4,
    edge_types: list[str | type[ANL2025Negotiator]] = [
        Boulware2025,
        Random2025,
        Linear2025,
        Conceder2025,
    ],
    # outcome space
    nissues: int = 3,
    nvalues: int = 7,
    # mechanism params
    nsteps: int = 100,
    keep_order: bool = True,
    share_ufuns: bool = True,
    atomic: bool = False,
    # output and logging
    output: Path | str | None = Path.home() / "negmas" / "anl2025" / "session",
    name: str = "",
    dry: bool = False,
    method=DEFAULT_METHOD,
    verbose: bool = False,
) -> SessionResults:
    """Generates a multideal negotiation session and runs it.

    Args:
        method: the method to use for running all the sessions.
                Acceptable options are: sequential, ordered, threads, processes.
                See `negmas.mechanisms.Mechanism.run_all()` for full details.
        center_type: Type of the center agent
        center_params: Optional parameters to pass to the center agent.
        center_reserved_value_min: Minimum reserved value for the center agent.
        center_reserved_value_max: Maximum reserved value for the center agent.
        center_ufun_type: Type of the center agent ufun.
        center_ufun_params: Parameters to pass to the center agent ufun.
        nedges: Number of edges.
        edge_reserved_value_min: Minimum reserved value for edges.
        edge_reserved_value_max: Maximum reserved value for edges.
        edge_types: Types of edge agents
        nissues: Number of issues to use for each thread.
        nvalues: Number of values per issue for each thread.
        nsteps: Number of negotiation steps.
        keep_order: Keep the order of edges when advancing the negotiation.
        share_ufuns: If given, agents will have access to partner ufuns through `self.opponent_ufun`.
        atomic: If given, one step corresponds to one offer instead of a full round.
        output: Folder to store the logs and results within.
        name: Name of the session
        dry: IF true, nothing will be run.
        verbose: Print progress

    Returns:
        `SessionResults` giving the results of the multideal negotiation session.
    """
    if output and isinstance(output, str):
        output = Path(output)
    sample_edges = nedges > 0
    if not sample_edges:
        nedges = len(edge_types)
    scenario = make_multideal_scenario(
        nedges=nedges,
        nissues=nissues,
        nvalues=nvalues,
        center_reserved_value_min=center_reserved_value_min,
        center_reserved_value_max=center_reserved_value_max,
        center_ufun_type=center_ufun_type,
        center_ufun_params=center_ufun_params,
        edge_reserved_value_min=edge_reserved_value_min,
        edge_reserved_value_max=edge_reserved_value_max,
    )
    run_params = RunParams(
        nsteps=nsteps,
        keep_order=keep_order,
        share_ufuns=share_ufuns,
        atomic=atomic,
        method=method,
    )
    assigned = assign_scenario(
        scenario=scenario,
        run_params=run_params,
        center_type=center_type,
        center_params=center_params,
        edge_types=edge_types,
        verbose=verbose,
        sample_edges=sample_edges,
    )
    return assigned.run(output=output, name=name, dry=dry, verbose=verbose)

anl2025.runner.RunParams

Defines the running parameters of the multi-deal negotiation like time-limits.

Attributes:

Name Type Description
nsteps int

Number of negotiation steps
keep_order bool

Keep the order of negotiation threads when scheduling next action
share_ufuns bool

If given, agents can access partner ufuns through self.opponeng_ufun
atomic bool

Every step is a single offer (if not given, on the other hand, every step is a complete round)
method str

the method to use for running all the sessions. Acceptable options are: sequential, ordered, threads, processes. See negmas.mechanisms.Mechanism.run_all() for full details. ANL2025 uses the sequential option which means that a complete negotiation is finished before the next starts.
time_limit float | None

Number of seconds allowed per negotiation
Source code in anl2025/common.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
@define
class RunParams:
    """Defines the running parameters of the multi-deal negotiation like time-limits.

    Attributes:
        nsteps: Number of negotiation steps
        keep_order: Keep the order of  negotiation threads when scheduling next action
        share_ufuns: If given, agents can access partner ufuns through `self.opponeng_ufun`
        atomic: Every step is a single offer (if not given, on the other hand, every step is a complete round)
        method: the method to use for running all the sessions.  Acceptable options are: sequential,
                ordered, threads, processes. See `negmas.mechanisms.Mechanism.run_all()` for full details.
                ANL2025 uses the sequential option which means that a complete negotiation is finished before
                the next starts.
        time_limit: Number of seconds allowed per negotiation
    """

    # mechanism params
    nsteps: int = 100
    keep_order: bool = True
    share_ufuns: bool = False
    atomic: bool = False
    method: str = DEFAULT_METHOD
    time_limit: float | None = None

anl2025.runner.SessionResults

Results of a single multideal negotiation

Attributes:

Name Type Description
mechanisms list[SAOMechanism]

The mechanisms representing negotiation threads.
center ANL2025Negotiator

The center agent
edges list[ANL2025Negotiator]

Edge agents
agreements list[Outcome | None]

Negotiation outcomes for all threads.
center_utility float

The utility received by the center.
edge_utilities list[float]

The utilities of all edges.
Source code in anl2025/runner.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
@define
class SessionResults:
    """Results of a single multideal negotiation

    Attributes:
        mechanisms: The mechanisms representing negotiation threads.
        center: The center agent
        edges: Edge agents
        agreements:  Negotiation outcomes for all threads.
        center_utility: The utility received by the center.
        edge_utilities: The utilities of all edges.
    """

    mechanisms: list[SAOMechanism]
    center: ANL2025Negotiator
    edges: list[ANL2025Negotiator]
    agreements: list[Outcome | None]
    center_utility: float
    edge_utilities: list[float]

Tournaments

anl2025.tournament.Tournament

Represents a tournament

Attributes:

Name Type Description
competitors tuple[str | type[ANL2025Negotiator], ...]

the competing agents of type ANL2025Negotiator each
scenarios tuple[MultidealScenario, ...]

the scenarios in which the competitors are tested
run_params RunParams

parameters controlling the tournament run (See RunParams)
competitor_params tuple[dict[str, Any] | None, ...] | None

Parameters to pass to the competitors

Methods:

Name Description
from_scenarios

Loads a tournament from the given scenarios (optionally generating new ones)
load

Loads the tournament information.
run

Run the tournament
save

Saves the tournament information.
Source code in anl2025/tournament.py 
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
@define
class Tournament:
    """Represents a tournament

    Attributes:
        competitors: the competing agents of type `ANL2025Negotiator` each
        scenarios: the scenarios in which the competitors are tested
        run_params: parameters controlling the tournament run (See `RunParams`)
        competitor_params: Parameters to pass to the competitors
    """

    competitors: tuple[str | type[ANL2025Negotiator], ...]
    scenarios: tuple[MultidealScenario, ...]
    run_params: RunParams
    competitor_params: tuple[dict[str, Any] | None, ...] | None = None

    @classmethod
    def from_scenarios(
        cls,
        competitors: Sequence[str | type[ANL2025Negotiator]],
        run_params: RunParams,
        scenarios: tuple[MultidealScenario, ...] = tuple(),
        n_generated: int = 0,
        nedges: int = 3,
        nissues: int = 3,
        nvalues: int = 7,
        # edge ufuns
        center_reserved_value_min: float = 0.0,
        center_reserved_value_max: float = 0.0,
        center_ufun_type: str | type[CenterUFun] = "MaxCenterUFun",
        center_ufun_params: dict[str, Any] | None = None,
        # edge ufuns
        edge_reserved_value_min: float = 0.1,
        edge_reserved_value_max: float = 0.4,
        competitor_params: tuple[dict[str, Any] | None, ...] | None = None,
    ) -> Self:
        """Loads a tournament from the given scenarios (optionally generating new ones)

        Args:
            competitors: Competing agents
            run_params: `RunParams` controlling the timing of each multideal negotiation
            scenarios: An optional tuple of predefined scenarios (`MultidealScenario`)
            n_generated: Number of new scenarios to generate
            nedges: Number of negotiation threads (only used if `n_generated` > 0)
            nissues:Number of negotiation issues per thread (only used if `n_generated` > 0)
            nvalues: Number of values per issue (only used if `n_generated` > 0)
            center_reserved_value_min: Minimum reserved value of the center for generated scenarios.
            center_reserved_value_max: Maximum reserved value of the center for generated scenarios.
            center_ufun_type: center agent ufun for generated scenarios.
            center_ufun_params: center agent ufun params for generated scenarios.
            edge_reserved_value_min: Minimum reserved value of  edges for generated scenarios.
            edge_reserved_value_max: Maximum reserved value of  edges for generated scenarios.
            competitor_params: Optional competitor paramters

        Returns:
            A `Tournament` ready to run
        """
        # if nedges > len(competitors):
        #     raise ValueError(
        #         f"We have {len(competitors)} competitors which is not enough for {nedges} edges"
        #     )
        return cls(
            competitors=tuple(competitors),
            competitor_params=competitor_params,
            run_params=run_params,
            scenarios=tuple(
                list(scenarios)
                + [
                    make_multideal_scenario(
                        nedges=nedges,
                        nissues=nissues,
                        nvalues=nvalues,
                        center_reserved_value_min=center_reserved_value_min,
                        center_reserved_value_max=center_reserved_value_max,
                        center_ufun_type=center_ufun_type,
                        center_ufun_params=center_ufun_params,
                        edge_reserved_value_min=edge_reserved_value_min,
                        edge_reserved_value_max=edge_reserved_value_max,
                    )
                    for _ in range(n_generated)
                ]
            ),
        )

    def __attrs_post_init__(self):
        if not self.competitor_params:
            self.competitor_params = tuple(dict() for _ in range(len(self.competitors)))
        self.competitor_params = tuple(
            dict() if not _ else _ for _ in self.competitor_params
        )

    def save(
        self,
        path: Path | str,
        separate_scenarios: bool = False,
        python_class_identifier=TYPE_IDENTIFIER,
    ):
        """
        Saves the tournament information.

        Args:
            path: A file to save information about the tournament to
            separate_scenarios: If `True`, scenarios will be saved inside a `scenarios` folder beside the path given otherwise they will be included in the file
        """
        path = path if isinstance(path, Path) else Path(path)
        data = dict(
            competitors=[get_full_type_name(_) for _ in self.competitors],
            run_params=asdict(self.run_params),
            competitor_params=None
            if not self.competitor_params
            else [
                serialize(_, python_class_identifier=python_class_identifier)
                for _ in self.competitor_params
            ],
        )
        if separate_scenarios:
            base = path.resolve().parent / "scenarios"
            for i, s in enumerate(self.scenarios):
                name = s.name if s.name else f"s{i:03}"
                dst = base
                dst.mkdir(parents=True, exist_ok=True)
                dump(
                    serialize(s, python_class_identifier=python_class_identifier),
                    dst / f"{name}.yaml",
                )
        else:
            data["scenarios"] = [
                serialize(_, python_class_identifier=python_class_identifier)
                for _ in self.scenarios
            ]
        dump(data, path)

    @classmethod
    def load(cls, path: Path | str, python_class_identifier=TYPE_IDENTIFIER):
        """Loads the tournament information."""

        path = path if isinstance(path, Path) else Path(path)
        info = load(path)
        base = path.resolve().parent / "scenarios"
        if "scenarios" not in info:
            info["scenarios"] = []
        else:
            info["scenarios"] = list(info["scenarios"])

        if base.exists():
            info["scenarios"] += [
                deserialize(f, python_class_identifier=python_class_identifier)
                for f in base.glob("*.yaml")
            ]

        return cls(
            competitors=info["competitors"],
            scenarios=[
                deserialize(_, python_class_identifier=python_class_identifier)
                for _ in info["scenarios"]
            ],  # type: ignore
            run_params=RunParams(**info["run_params"]),
            competitor_params=None  # type: ignore
            if not info.get("competitor_params", None)
            else deserialize(
                info["competitor_params"],
                python_class_identifier=python_class_identifier,
            ),
        )

    def run(
        self,
        n_repetitions: int,
        path: Path | str | None = None,
        verbose: bool = False,
        dry: bool = False,
        no_double_scores: bool = True,
        non_comptitor_types: tuple[str | type[ANL2025Negotiator], ...] | None = None,
        non_comptitor_params: tuple[dict[str, Any], ...] | None = None,
        n_jobs: int | float | None = 0,
        center_multiplier: float | None = None,
        edge_multiplier: float = 1,
    ) -> TournamentResults:
        """Run the tournament

        Args:
            n_repetitions: Number of repetitions of rotations over scenarios
            path: Path to save the results to
            verbose: Print progress
            dry: Do not really run the negotiations.
            no_double_scores: Avoid having the same agent in multiple positions in the same negotiation
            non_comptitor_types: Types to use to fill missing edge locations if not enough competitors are available
            non_comptitor_params: Paramters of non-competitor-types
            n_jobs: Number of parallel jobs to use.
                    None (and negative numbers) mean serially, 0 means use all cores, fractions mean fraction of available
                    cores, integers mean exact number of cores
            center_multiplier: A number to multiply center utilities with before calculating the score. Can be used
                               to give more or less value to being a center. If None, it will be equal to the number of edges.
            edge_multiplier: A number to multiply edge utilities with before calculating the score. Can be used
                               to give more or less value to being an edge

        Returns:
            `TournamentResults` with all scores and final-scores
        """
        if path is not None:
            path = path if isinstance(path, Path) else Path(path)
        if n_jobs is not None:
            if isinstance(n_jobs, float) and n_jobs < 1.0:
                n_jobs = int(0.5 + cpu_count() * n_jobs)
            elif isinstance(n_jobs, float):
                n_jobs = int(0.5 + n_jobs)
            if n_jobs < 0:
                n_jobs = None
            elif n_jobs == 0:
                n_jobs = cpu_count()

        results = []
        assert isinstance(self.competitor_params, tuple)
        final_scores = defaultdict(float)
        final_scoresC = defaultdict(float)
        final_scoresE = defaultdict(float)
        count_edge = defaultdict(float)
        count_center = defaultdict(float)

        scores = []
        center_multiplier_val = center_multiplier

        def type_name(x):
            return get_full_type_name(x).replace("anl2025.negotiator.", "")

        if non_comptitor_types:
            non_comptitor_types = tuple(get_class(_) for _ in non_comptitor_types)
            non_comptitor_params = (
                non_comptitor_params
                if non_comptitor_params
                else tuple(dict() for _ in range(len(non_comptitor_types)))
            )
            non_competitors = [
                (n, p)
                for n, p in zip(non_comptitor_types, non_comptitor_params, strict=True)
            ]
        else:
            non_competitors = None

        jobs = []
        for i in track(range(n_repetitions), "Preparing Negotiation Sessions"):
            competitors = [
                (get_class(c), p)
                for c, p in zip(self.competitors, self.competitor_params, strict=True)
            ]
            for k, scenario in enumerate(self.scenarios):
                nedges = len(scenario.edge_ufuns)
                sname = scenario.name if scenario.name else f"s{k:03}"
                random.shuffle(competitors)
                for j in range(len(competitors)):
                    if len(competitors) >= nedges + 1:
                        players = competitors[: nedges + 1]
                    else:
                        # add extra players at the end if not enough competitors are available
                        players = competitors + list(
                            random.choices(
                                non_competitors if non_competitors else competitors,
                                k=nedges + 1 - len(competitors),
                            )
                        )
                    # ignore the randomly added edges if no-double-scores is set
                    nedges_counted = (
                        nedges
                        if not no_double_scores
                        else min(len(competitors) - 1, nedges)
                    )
                    if path:
                        output = path / "results" / sname / f"r{j:03}t{i:03}"
                    else:
                        output = None
                    center, center_params = players[j]
                    edge_info = [_ for _ in players[:j] + players[j + 1 :]]
                    # not sure if the following shuffle is useful!
                    # It tries to randomize the order of the edges to avoid
                    # having a systematic bias but we randomize competitors anyway.
                    random.shuffle(edge_info)
                    edges = [_[0] for _ in edge_info]
                    edge_params = [_[1] if _[1] else dict() for _ in edge_info]
                    assigned = assign_scenario(
                        scenario=scenario,
                        run_params=self.run_params,
                        center_type=center,
                        center_params=center_params,
                        edge_types=edges,  # type: ignore
                        edge_params=edge_params,
                        verbose=verbose,
                        sample_edges=False,
                    )
                    jobs.append(
                        JobInfo(
                            assigned,
                            output,
                            sname,
                            i,
                            j,
                            k,
                            center,
                            center_params,
                            edges,
                            edge_params,
                            edge_info,
                            nedges_counted,
                        )
                    )
            # This rotation guarantees that every competitor is
            # the center once per scenario per repetition
            competitors = [competitors[-1]] + competitors[:-1]
        if verbose:
            print(f"Will run {len(jobs)} negotiations")

        def process_info(job: JobInfo, info: SessionInfo):
            center_multiplier = (
                center_multiplier_val
                if center_multiplier_val is not None
                else len(job.edge_info)
            )
            r = info.results
            results.append(info)
            center, center_params = job.center, job.center_params
            cname = (
                type_name(center)
                if not center_params
                else f"{type_name(center)}_{hash(str(center_params))}"
            )
            mean_edge_utility = sum(r.edge_utilities) / len(r.edge_utilities)
            scores.append(
                dict(
                    agent=cname,
                    utility=r.center_utility * center_multiplier,
                    partner_average_utility=mean_edge_utility,
                    scenario=job.sname,
                    repetition=job.rep_index,
                    rotation=job.competitor_index,
                    scenario_index=job.scenario_index,
                    index=0,
                )
            )
            final_scores[cname] += r.center_utility * center_multiplier
            final_scoresC[cname] += r.center_utility * center_multiplier
            count_center[cname] += 1
            for e, (c, p) in enumerate(job.edge_info[: job.nedges_counted]):
                cname = type_name(c) if not p else f"{type_name(c)}_{hash(str(p))}"
                scores.append(
                    dict(
                        agent=cname,
                        utility=r.edge_utilities[e] * edge_multiplier,
                        partner_average_utility=r.center_utility,
                        scenario=job.sname,
                        repetition=job.rep_index,
                        rotation=job.competitor_index,
                        scenario_index=job.scenario_index,
                        index=e + 1,
                    )
                )
                final_scores[cname] += r.edge_utilities[e] * edge_multiplier
                final_scoresE[cname] += r.edge_utilities[e] * edge_multiplier
                count_edge[cname] += 1

            if verbose:
                print(f"Center Utility: {r.center_utility}")
                print(f"Edge Utilities: {r.edge_utilities}")
                print(f"Agreement: {r.agreements}")

        if n_jobs is None:
            for job in track(jobs, "Running Negotiations"):
                job, info = run_session(job, dry, verbose)
                process_info(job, info)
        else:
            assert n_jobs > 0
            with ProcessPoolExecutor(max_workers=n_jobs) as executor:
                # Submit all jobs and store the futures
                futures = [
                    executor.submit(run_session, job, dry, verbose) for job in jobs
                ]

                # Process results as they become available
                for future in as_completed(futures):
                    try:
                        job, info = future.result()
                        process_info(job, info)
                    except Exception as e:
                        print(f"Job failed with exception: {e}")

        weighted_average = {}
        for agent in final_scores.keys():
            average_score_E = (
                final_scoresE[agent] / count_edge[agent] if count_edge[agent] > 0 else 0
            )
            average_score_C = (
                final_scoresC[agent] / count_center[agent]
                if count_center[agent] > 0
                else 0
            )
            weighted_average[agent] = 0.5 * (average_score_C + average_score_E)

        return TournamentResults(
            final_scores={k: v for k, v in final_scores.items()},
            edge_count={k: v for k, v in count_edge.items()},
            center_count={k: v for k, v in count_center.items()},
            final_scoresC={k: v for k, v in final_scoresC.items()},
            final_scoresE={k: v for k, v in final_scoresE.items()},
            weighted_average={k: v for k, v in weighted_average.items()},
            scores=scores,
            session_results=results,
        )

from_scenarios classmethod

Loads a tournament from the given scenarios (optionally generating new ones)

Parameters:

Name Type Description Default
competitors Sequence[str | type[ANL2025Negotiator]]

Competing agents

 required
run_params RunParams

RunParams controlling the timing of each multideal negotiation

 required
scenarios tuple[MultidealScenario, ...]

An optional tuple of predefined scenarios (MultidealScenario)

 tuple()
n_generated int

Number of new scenarios to generate

 0
nedges int

Number of negotiation threads (only used if n_generated > 0)

 3
nissues int

Number of negotiation issues per thread (only used if n_generated > 0)

 3
nvalues int

Number of values per issue (only used if n_generated > 0)

 7
center_reserved_value_min float

Minimum reserved value of the center for generated scenarios.

 0.0
center_reserved_value_max float

Maximum reserved value of the center for generated scenarios.

 0.0
center_ufun_type str | type[CenterUFun]

center agent ufun for generated scenarios.

 'MaxCenterUFun'
center_ufun_params dict[str, Any] | None

center agent ufun params for generated scenarios.

 None
edge_reserved_value_min float

Minimum reserved value of edges for generated scenarios.

 0.1
edge_reserved_value_max float

Maximum reserved value of edges for generated scenarios.

 0.4
competitor_params tuple[dict[str, Any] | None, ...] | None

Optional competitor paramters

 None

Returns:

Type Description
Self

A Tournament ready to run
Source code in anl2025/tournament.py 
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
@classmethod
def from_scenarios(
    cls,
    competitors: Sequence[str | type[ANL2025Negotiator]],
    run_params: RunParams,
    scenarios: tuple[MultidealScenario, ...] = tuple(),
    n_generated: int = 0,
    nedges: int = 3,
    nissues: int = 3,
    nvalues: int = 7,
    # edge ufuns
    center_reserved_value_min: float = 0.0,
    center_reserved_value_max: float = 0.0,
    center_ufun_type: str | type[CenterUFun] = "MaxCenterUFun",
    center_ufun_params: dict[str, Any] | None = None,
    # edge ufuns
    edge_reserved_value_min: float = 0.1,
    edge_reserved_value_max: float = 0.4,
    competitor_params: tuple[dict[str, Any] | None, ...] | None = None,
) -> Self:
    """Loads a tournament from the given scenarios (optionally generating new ones)

    Args:
        competitors: Competing agents
        run_params: `RunParams` controlling the timing of each multideal negotiation
        scenarios: An optional tuple of predefined scenarios (`MultidealScenario`)
        n_generated: Number of new scenarios to generate
        nedges: Number of negotiation threads (only used if `n_generated` > 0)
        nissues:Number of negotiation issues per thread (only used if `n_generated` > 0)
        nvalues: Number of values per issue (only used if `n_generated` > 0)
        center_reserved_value_min: Minimum reserved value of the center for generated scenarios.
        center_reserved_value_max: Maximum reserved value of the center for generated scenarios.
        center_ufun_type: center agent ufun for generated scenarios.
        center_ufun_params: center agent ufun params for generated scenarios.
        edge_reserved_value_min: Minimum reserved value of  edges for generated scenarios.
        edge_reserved_value_max: Maximum reserved value of  edges for generated scenarios.
        competitor_params: Optional competitor paramters

    Returns:
        A `Tournament` ready to run
    """
    # if nedges > len(competitors):
    #     raise ValueError(
    #         f"We have {len(competitors)} competitors which is not enough for {nedges} edges"
    #     )
    return cls(
        competitors=tuple(competitors),
        competitor_params=competitor_params,
        run_params=run_params,
        scenarios=tuple(
            list(scenarios)
            + [
                make_multideal_scenario(
                    nedges=nedges,
                    nissues=nissues,
                    nvalues=nvalues,
                    center_reserved_value_min=center_reserved_value_min,
                    center_reserved_value_max=center_reserved_value_max,
                    center_ufun_type=center_ufun_type,
                    center_ufun_params=center_ufun_params,
                    edge_reserved_value_min=edge_reserved_value_min,
                    edge_reserved_value_max=edge_reserved_value_max,
                )
                for _ in range(n_generated)
            ]
        ),
    )

load classmethod

Loads the tournament information.

Source code in anl2025/tournament.py 
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
@classmethod
def load(cls, path: Path | str, python_class_identifier=TYPE_IDENTIFIER):
    """Loads the tournament information."""

    path = path if isinstance(path, Path) else Path(path)
    info = load(path)
    base = path.resolve().parent / "scenarios"
    if "scenarios" not in info:
        info["scenarios"] = []
    else:
        info["scenarios"] = list(info["scenarios"])

    if base.exists():
        info["scenarios"] += [
            deserialize(f, python_class_identifier=python_class_identifier)
            for f in base.glob("*.yaml")
        ]

    return cls(
        competitors=info["competitors"],
        scenarios=[
            deserialize(_, python_class_identifier=python_class_identifier)
            for _ in info["scenarios"]
        ],  # type: ignore
        run_params=RunParams(**info["run_params"]),
        competitor_params=None  # type: ignore
        if not info.get("competitor_params", None)
        else deserialize(
            info["competitor_params"],
            python_class_identifier=python_class_identifier,
        ),
    )

run

Run the tournament

Parameters:

Name Type Description Default
n_repetitions int

Number of repetitions of rotations over scenarios

 required
path Path | str | None

Path to save the results to

 None
verbose bool

Print progress

 False
dry bool

Do not really run the negotiations.

 False
no_double_scores bool

Avoid having the same agent in multiple positions in the same negotiation

 True
non_comptitor_types tuple[str | type[ANL2025Negotiator], ...] | None

Types to use to fill missing edge locations if not enough competitors are available

 None
non_comptitor_params tuple[dict[str, Any], ...] | None

Paramters of non-competitor-types

 None
n_jobs int | float | None

Number of parallel jobs to use. None (and negative numbers) mean serially, 0 means use all cores, fractions mean fraction of available cores, integers mean exact number of cores

 0
center_multiplier float | None

A number to multiply center utilities with before calculating the score. Can be used to give more or less value to being a center. If None, it will be equal to the number of edges.

 None
edge_multiplier float

A number to multiply edge utilities with before calculating the score. Can be used to give more or less value to being an edge

 1

Returns:

Type Description
TournamentResults

TournamentResults with all scores and final-scores
Source code in anl2025/tournament.py 
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
def run(
    self,
    n_repetitions: int,
    path: Path | str | None = None,
    verbose: bool = False,
    dry: bool = False,
    no_double_scores: bool = True,
    non_comptitor_types: tuple[str | type[ANL2025Negotiator], ...] | None = None,
    non_comptitor_params: tuple[dict[str, Any], ...] | None = None,
    n_jobs: int | float | None = 0,
    center_multiplier: float | None = None,
    edge_multiplier: float = 1,
) -> TournamentResults:
    """Run the tournament

    Args:
        n_repetitions: Number of repetitions of rotations over scenarios
        path: Path to save the results to
        verbose: Print progress
        dry: Do not really run the negotiations.
        no_double_scores: Avoid having the same agent in multiple positions in the same negotiation
        non_comptitor_types: Types to use to fill missing edge locations if not enough competitors are available
        non_comptitor_params: Paramters of non-competitor-types
        n_jobs: Number of parallel jobs to use.
                None (and negative numbers) mean serially, 0 means use all cores, fractions mean fraction of available
                cores, integers mean exact number of cores
        center_multiplier: A number to multiply center utilities with before calculating the score. Can be used
                           to give more or less value to being a center. If None, it will be equal to the number of edges.
        edge_multiplier: A number to multiply edge utilities with before calculating the score. Can be used
                           to give more or less value to being an edge

    Returns:
        `TournamentResults` with all scores and final-scores
    """
    if path is not None:
        path = path if isinstance(path, Path) else Path(path)
    if n_jobs is not None:
        if isinstance(n_jobs, float) and n_jobs < 1.0:
            n_jobs = int(0.5 + cpu_count() * n_jobs)
        elif isinstance(n_jobs, float):
            n_jobs = int(0.5 + n_jobs)
        if n_jobs < 0:
            n_jobs = None
        elif n_jobs == 0:
            n_jobs = cpu_count()

    results = []
    assert isinstance(self.competitor_params, tuple)
    final_scores = defaultdict(float)
    final_scoresC = defaultdict(float)
    final_scoresE = defaultdict(float)
    count_edge = defaultdict(float)
    count_center = defaultdict(float)

    scores = []
    center_multiplier_val = center_multiplier

    def type_name(x):
        return get_full_type_name(x).replace("anl2025.negotiator.", "")

    if non_comptitor_types:
        non_comptitor_types = tuple(get_class(_) for _ in non_comptitor_types)
        non_comptitor_params = (
            non_comptitor_params
            if non_comptitor_params
            else tuple(dict() for _ in range(len(non_comptitor_types)))
        )
        non_competitors = [
            (n, p)
            for n, p in zip(non_comptitor_types, non_comptitor_params, strict=True)
        ]
    else:
        non_competitors = None

    jobs = []
    for i in track(range(n_repetitions), "Preparing Negotiation Sessions"):
        competitors = [
            (get_class(c), p)
            for c, p in zip(self.competitors, self.competitor_params, strict=True)
        ]
        for k, scenario in enumerate(self.scenarios):
            nedges = len(scenario.edge_ufuns)
            sname = scenario.name if scenario.name else f"s{k:03}"
            random.shuffle(competitors)
            for j in range(len(competitors)):
                if len(competitors) >= nedges + 1:
                    players = competitors[: nedges + 1]
                else:
                    # add extra players at the end if not enough competitors are available
                    players = competitors + list(
                        random.choices(
                            non_competitors if non_competitors else competitors,
                            k=nedges + 1 - len(competitors),
                        )
                    )
                # ignore the randomly added edges if no-double-scores is set
                nedges_counted = (
                    nedges
                    if not no_double_scores
                    else min(len(competitors) - 1, nedges)
                )
                if path:
                    output = path / "results" / sname / f"r{j:03}t{i:03}"
                else:
                    output = None
                center, center_params = players[j]
                edge_info = [_ for _ in players[:j] + players[j + 1 :]]
                # not sure if the following shuffle is useful!
                # It tries to randomize the order of the edges to avoid
                # having a systematic bias but we randomize competitors anyway.
                random.shuffle(edge_info)
                edges = [_[0] for _ in edge_info]
                edge_params = [_[1] if _[1] else dict() for _ in edge_info]
                assigned = assign_scenario(
                    scenario=scenario,
                    run_params=self.run_params,
                    center_type=center,
                    center_params=center_params,
                    edge_types=edges,  # type: ignore
                    edge_params=edge_params,
                    verbose=verbose,
                    sample_edges=False,
                )
                jobs.append(
                    JobInfo(
                        assigned,
                        output,
                        sname,
                        i,
                        j,
                        k,
                        center,
                        center_params,
                        edges,
                        edge_params,
                        edge_info,
                        nedges_counted,
                    )
                )
        # This rotation guarantees that every competitor is
        # the center once per scenario per repetition
        competitors = [competitors[-1]] + competitors[:-1]
    if verbose:
        print(f"Will run {len(jobs)} negotiations")

    def process_info(job: JobInfo, info: SessionInfo):
        center_multiplier = (
            center_multiplier_val
            if center_multiplier_val is not None
            else len(job.edge_info)
        )
        r = info.results
        results.append(info)
        center, center_params = job.center, job.center_params
        cname = (
            type_name(center)
            if not center_params
            else f"{type_name(center)}_{hash(str(center_params))}"
        )
        mean_edge_utility = sum(r.edge_utilities) / len(r.edge_utilities)
        scores.append(
            dict(
                agent=cname,
                utility=r.center_utility * center_multiplier,
                partner_average_utility=mean_edge_utility,
                scenario=job.sname,
                repetition=job.rep_index,
                rotation=job.competitor_index,
                scenario_index=job.scenario_index,
                index=0,
            )
        )
        final_scores[cname] += r.center_utility * center_multiplier
        final_scoresC[cname] += r.center_utility * center_multiplier
        count_center[cname] += 1
        for e, (c, p) in enumerate(job.edge_info[: job.nedges_counted]):
            cname = type_name(c) if not p else f"{type_name(c)}_{hash(str(p))}"
            scores.append(
                dict(
                    agent=cname,
                    utility=r.edge_utilities[e] * edge_multiplier,
                    partner_average_utility=r.center_utility,
                    scenario=job.sname,
                    repetition=job.rep_index,
                    rotation=job.competitor_index,
                    scenario_index=job.scenario_index,
                    index=e + 1,
                )
            )
            final_scores[cname] += r.edge_utilities[e] * edge_multiplier
            final_scoresE[cname] += r.edge_utilities[e] * edge_multiplier
            count_edge[cname] += 1

        if verbose:
            print(f"Center Utility: {r.center_utility}")
            print(f"Edge Utilities: {r.edge_utilities}")
            print(f"Agreement: {r.agreements}")

    if n_jobs is None:
        for job in track(jobs, "Running Negotiations"):
            job, info = run_session(job, dry, verbose)
            process_info(job, info)
    else:
        assert n_jobs > 0
        with ProcessPoolExecutor(max_workers=n_jobs) as executor:
            # Submit all jobs and store the futures
            futures = [
                executor.submit(run_session, job, dry, verbose) for job in jobs
            ]

            # Process results as they become available
            for future in as_completed(futures):
                try:
                    job, info = future.result()
                    process_info(job, info)
                except Exception as e:
                    print(f"Job failed with exception: {e}")

    weighted_average = {}
    for agent in final_scores.keys():
        average_score_E = (
            final_scoresE[agent] / count_edge[agent] if count_edge[agent] > 0 else 0
        )
        average_score_C = (
            final_scoresC[agent] / count_center[agent]
            if count_center[agent] > 0
            else 0
        )
        weighted_average[agent] = 0.5 * (average_score_C + average_score_E)

    return TournamentResults(
        final_scores={k: v for k, v in final_scores.items()},
        edge_count={k: v for k, v in count_edge.items()},
        center_count={k: v for k, v in count_center.items()},
        final_scoresC={k: v for k, v in final_scoresC.items()},
        final_scoresE={k: v for k, v in final_scoresE.items()},
        weighted_average={k: v for k, v in weighted_average.items()},
        scores=scores,
        session_results=results,
    )

save

Saves the tournament information.

Parameters:

Name Type Description Default
path Path | str

A file to save information about the tournament to

 required
separate_scenarios bool

If True, scenarios will be saved inside a scenarios folder beside the path given otherwise they will be included in the file

 False
Source code in anl2025/tournament.py 
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
def save(
    self,
    path: Path | str,
    separate_scenarios: bool = False,
    python_class_identifier=TYPE_IDENTIFIER,
):
    """
    Saves the tournament information.

    Args:
        path: A file to save information about the tournament to
        separate_scenarios: If `True`, scenarios will be saved inside a `scenarios` folder beside the path given otherwise they will be included in the file
    """
    path = path if isinstance(path, Path) else Path(path)
    data = dict(
        competitors=[get_full_type_name(_) for _ in self.competitors],
        run_params=asdict(self.run_params),
        competitor_params=None
        if not self.competitor_params
        else [
            serialize(_, python_class_identifier=python_class_identifier)
            for _ in self.competitor_params
        ],
    )
    if separate_scenarios:
        base = path.resolve().parent / "scenarios"
        for i, s in enumerate(self.scenarios):
            name = s.name if s.name else f"s{i:03}"
            dst = base
            dst.mkdir(parents=True, exist_ok=True)
            dump(
                serialize(s, python_class_identifier=python_class_identifier),
                dst / f"{name}.yaml",
            )
    else:
        data["scenarios"] = [
            serialize(_, python_class_identifier=python_class_identifier)
            for _ in self.scenarios
        ]
    dump(data, path)