Ridesharing
Riders who request trips on a city grid and cabs that a stable matching sends to them.
- Name:
ridesharing - Field:
engineering/traffic - Formalism: agent-based model
- References: Carmichael and Hadzikadic 2019 ch. 7
Factory
@register_model(
field='engineering/traffic',
tags=('ridesharing', 'agents', 'matching', 'queues', 'grid'),
references=('Carmichael and Hadzikadic 2019 ch. 7',),
defaults={'until': 200},
observables=_ridesharing_observables,
)
def ridesharing(
n_cabs: int = 30,
*,
demand: float = 1.0,
width: int = 30,
height: int = 30,
base_fare: float = 2.0,
block_fare: float = 0.5,
random_state: int | np.random.Generator | None = 0,
) -> System:
"""Return riders who request trips on a city grid and cabs that a stable matching sends to them.
Every step new riders appear at random cells with random destinations, and the dispatch matches the waiting
riders with the idle cabs by a stable matching in which riders prefer nearer cabs and cabs prefer riders who waited
longer. Every cab drives one block a step to its rider, collects the fare, and drives the rider to its destination.
While the demand stays below the trips the fleet can serve, riders wait about as long as a cab takes to reach them.
As the demand approaches that capacity, every cab is busy, the queue of waiting riders grows, and the waiting time
rises steeply, as in a queue near saturation.
Args:
n_cabs:
The number of cabs.
demand:
The mean number of new riders in a step.
width:
The number of columns of the grid.
height:
The number of rows of the grid.
base_fare:
The fare of every trip.
block_fare:
The fare per block of a trip.
random_state:
The seed or generator of the cells of the cabs.
Returns:
system:
The model, with the groups `riders` and `cabs`, the arrivals `arrival`, and the matching `dispatch`.
Examples:
>>> from skcomplex.models.engineering.traffic import ridesharing
>>> [group.size for group in ridesharing(5).elements]
[1, 5]
"""
rng = np.random.default_rng(random_state)
cells = rng.integers((width, height), size=(n_cabs + 1, 2))
riders = GroupElement(
'riders',
element=RiderElement,
size=1,
values={'destination_x': float(rng.integers(width)), 'destination_y': float(rng.integers(height))},
variable=True,
location=[(int(cells[0, 0]), int(cells[0, 1]))],
)
cabs = GroupElement(
'cabs',
element=CabElement,
size=n_cabs,
values={'base_fare': base_fare, 'block_fare': block_fare},
location=[(int(x), int(y)) for x, y in cells[1:]],
)
return System(
elements=[riders, cabs],
interactions=[
ArrivalInteraction('arrival', target='riders', demand=demand, width=width, height=height),
MapInteraction(
'waiting',
target='riders',
state='waited',
function=lambda agent, model: 1.0,
mode='increment',
form='agent',
phase='drive',
),
MatchingInteraction(
'dispatch',
requesters='riders',
providers='cabs',
rule=StableMatchingSpec(requester_preference=_nearness, provider_preference=_patience),
quantity='seats',
phase='dispatch',
on_match=_dispatch,
),
],
space=LatticeSpace((width, height), boundary='fixed', neighborhood='von_neumann'),
time=DiscreteTime(
schedule=PhasesScheduleSpec([PhaseSpec('arrival'), PhaseSpec('dispatch'), PhaseSpec('drive')]),
),
)
Components
class RiderElement(Element):
"""Rider who waits at a cell for a cab to a destination, and pays the fare when the cab picks it up.
Args:
label:
The label of the agent or group.
visual:
How the element is drawn, or `None` for `default_visual_`.
destination_x:
The column of the destination.
destination_y:
The row of the destination.
waited:
The number of steps the rider has waited.
seats:
The seats the rider still requests, one until a cab is on its way and zero after.
money:
The money of the rider.
Examples:
>>> from skcomplex.core.elements import GroupElement
>>> from skcomplex.models.engineering.traffic import RiderElement
>>> GroupElement('riders', element=RiderElement, size=2, values={'money': 50.0}).size
2
"""
def __init__(
self,
label: str,
visual: VisualSpec | None = VisualSpec('agents', {'color_by': 'waited'}),
*,
destination_x: Annotated[float, StateSpec(domain=NN)] = 0.0,
destination_y: Annotated[float, StateSpec(domain=NN)] = 0.0,
waited: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
seats: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
money: Annotated[float, StateSpec()] = 100.0,
) -> None:
super().__init__(label, visual=visual)
self.destination_x = destination_x
self.destination_y = destination_y
self.waited = waited
self.seats = seats
self.money = money
@classmethod
def _get_view(cls) -> type[RiderView]:
"""Return the view of the riders, which types their states."""
return RiderView
class CabElement(Element):
"""Cab that drives one block a step to its rider, collects the fare, and drives the rider to its destination.
The fare is a base fare plus a fare per block of the trip, paid as a transfer from rider to cab.
Args:
label:
The label of the agent or group.
visual:
How the element is drawn, or `None` for `default_visual_`.
task:
0 when idle, 1 on the way to a rider, and 2 on the way to a destination.
target_x:
The column of the cell the cab drives to.
target_y:
The row of the cell the cab drives to.
destination_x:
The column of the destination of the rider.
destination_y:
The row of the destination of the rider.
rider:
The id of the rider the cab is sent to.
seats:
The free seats of the cab, one when idle and zero on a trip.
money:
The fares the cab collected.
base_fare:
The fare of every trip.
block_fare:
The fare per block of a trip.
Examples:
>>> from skcomplex.core.elements import GroupElement
>>> from skcomplex.models.engineering.traffic import CabElement
>>> GroupElement('cabs', element=CabElement, size=3).size
3
"""
def __init__(
self,
label: str,
visual: VisualSpec | None = VisualSpec('agents', {'color_by': 'task', 'color_range': (0.0, 2.0)}),
*,
task: Annotated[float, StateSpec(domain=NN)] = IDLE,
target_x: Annotated[float, StateSpec(domain=NN)] = 0.0,
target_y: Annotated[float, StateSpec(domain=NN)] = 0.0,
destination_x: Annotated[float, StateSpec(domain=NN)] = 0.0,
destination_y: Annotated[float, StateSpec(domain=NN)] = 0.0,
rider: Annotated[float, StateSpec()] = -1.0,
seats: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
money: Annotated[float, StateSpec()] = 0.0,
base_fare: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 2.0,
block_fare: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.5,
) -> None:
super().__init__(label, visual=visual)
self.task = task
self.target_x = target_x
self.target_y = target_y
self.destination_x = destination_x
self.destination_y = destination_y
self.rider = rider
self.seats = seats
self.money = money
self.base_fare = base_fare
self.block_fare = block_fare
@classmethod
def _get_view(cls) -> type[CabView]:
"""Return the view of the cabs, which types their states and parameters."""
return CabView
def _get_laws(self, s: SimpleNamespace) -> list[BaseLaw]:
"""Return the driving rule of the phase of driving.
Args:
s:
The namespace of the symbols the laws can use.
Returns:
laws:
The laws of the component.
"""
return [RuleLaw(self.drive, phase='drive')]
def _pick_up(self, agent: CabView, model: ModelView) -> None:
"""Collect the fare of the rider, take it on board, and head for its destination."""
rider = next(item for item in model.agents('riders', RiderView) if item.id == int(agent.rider))
destination = (int(agent.destination_x), int(agent.destination_y))
blocks = CabElement._blocks(CabElement._cell(agent.location), destination)
fare = model.parameter('base_fare') + model.parameter('block_fare') * blocks
transfer(rider, agent, fare, 'money')
rider.remove()
new_target_x_value, new_target_y_value = destination
agent.target_x = new_target_x_value
agent.target_y = new_target_y_value
agent.task = DELIVERING
def drive(self, agent: CabView, model: ModelView) -> None:
"""Drive one block toward the target, and pick up or drop off the rider there.
Args:
agent:
The cab.
model:
The model.
"""
if agent.task == IDLE:
return
x, y = CabElement._cell(agent.location)
target = (int(agent.target_x), int(agent.target_y))
if (x, y) != target:
step = (int(np.sign(target[0] - x)), 0) if x != target[0] else (0, int(np.sign(target[1] - y)))
agent.move_to((x + step[0], y + step[1]))
if CabElement._cell(agent.location) != target:
return
if agent.task == PICKING_UP:
self._pick_up(agent, model)
else:
agent.task = IDLE
agent.seats = 1.0
@staticmethod
def _cell(location: object) -> tuple[int, int]:
"""Return the integer coordinates of a cell."""
x, y = np.atleast_1d(np.asarray(location, dtype=float))[:2]
return int(x), int(y)
@staticmethod
def _blocks(first: tuple[int, int], second: tuple[int, int]) -> int:
"""Return the number of blocks between two cells of the grid, the Manhattan distance."""
return abs(first[0] - second[0]) + abs(first[1] - second[1])
class ArrivalInteraction(LifeCycleInteraction):
"""Arrival of new riders at random cells with random destinations, a life cycle of arrivals only.
Every step a Poisson number of riders, with mean `demand`, appear at random cells with random destinations.
Args:
label:
The label of the arrivals.
target:
The path of the group of riders.
demand:
The mean number of new riders in a step.
width:
The number of columns of the grid.
height:
The number of rows of the grid.
phase:
The phase the riders arrive in.
Examples:
>>> from skcomplex.models.engineering.traffic import ArrivalInteraction
>>> ArrivalInteraction('arrival', target='riders').reports
('births', 'deaths', 'arrivals')
"""
def __init__(
self,
label: str,
*,
target: Annotated[str, RoleSpec()],
demand: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
width: int = 30,
height: int = 30,
phase: str | None = 'arrival',
) -> None:
super().__init__(
label,
target=target,
placement=self.place,
values=self.destine,
arrival=self.expect,
phase=phase,
)
self.demand = demand
self.width = width
self.height = height
def expect(self, model: ModelView) -> float:
"""Return the mean number of new riders in a step.
Args:
model:
The model view of the arrivals.
Returns:
demand:
The demand.
"""
return model.parameter('demand')
def place(self, agent: RiderView | None, model: ModelView) -> tuple[int, int]:
"""Return a random cell of the grid for a new rider.
Args:
agent:
The parent, `None` for an arrival.
model:
The model view of the arrivals.
Returns:
cell:
The coordinates of the cell.
"""
x, y = model.rng.integers((self.width, self.height))
return int(x), int(y)
def destine(self, agent: RiderView | None, model: ModelView) -> dict[str, object]:
"""Return a random destination on the grid for a new rider.
Args:
agent:
The parent, `None` for an arrival.
model:
The model view of the arrivals.
Returns:
values:
The coordinates of the destination by state name.
"""
to_x, to_y = model.rng.integers((self.width, self.height))
return {'destination_x': float(to_x), 'destination_y': float(to_y)}
class MapInteraction(Interaction):
"""Map of a state of discrete time, whose next value is a function, or the state plus it.
In the mode `'replace'` the next value of the state is `f`, and in the mode `'increment'` it is the state plus `f`.
Under the `'synchronous'` schedule every member of a bound group updates at once, from the states at the start of
the step. Under the `'sequential'` schedule the members update one after the other, in their order, and under the
`'random-sequential'` schedule in a new random order every step, each member reading the values the members before
it set. Under the `'member'` schedule of the agent form, every member updates in its own activation, together with
the rules of its element and the other member interactions, in the order the system lists them.
In the `'symbolic'` form, `function(namespace)` returns a symbolic expression, or a vector of them for a vector
state, where `namespace.target` holds the states and parameters of the target, `namespace.sources` the namespaces
of the sources, and `namespace.t` the time, and may hold draws such as `normal('noise')`. In the `'agent'` form,
the target is an agent group and `function(agent, model)` returns the value as a number, or as an array of the
shape of a vector state, so it can read the agent, its neighbours, and the history and storage of the model view.
Args:
label:
The label of the interaction, unique among the parts of its system.
target:
The path of the element or group whose state the map sets.
state:
The name of the state of the target.
function:
The function `f(namespace)` in the symbolic form, or `f(agent, model)` in the agent form.
sources:
The paths of the elements, or members of groups such as `cells[0]`, that the function reads.
mode:
`'replace'` or `'increment'`.
schedule:
`'synchronous'`, `'sequential'`, `'random-sequential'`, or `'member'` for the agent form.
form:
`'symbolic'` for update laws of the states of the target, or `'agent'` for a rule over an agent group.
phase:
The phase of the step the agent form runs in, or `None` for the default phase.
Raises:
ValidationError:
The mode, the schedule, or the form is unknown, the symbolic form has the schedule `'member'`, the
function is not callable, or the agent form returns a value that does not match the state, raised by
`run` of the simulation.
Examples:
>>> from skcomplex.core.elements import Element
>>> from skcomplex.core.interactions import MapInteraction
>>> from skcomplex.core.spaces import DiscreteTime
>>> from skcomplex.core.system import System
>>> from skcomplex.engine.runtime import Simulation
>>> from skcomplex.specs import StateSpec
>>> from typing import Annotated, SupportsFloat
>>> class ValueElement(Element):
... def __init__(self, label, x: Annotated[float, StateSpec()] = 0.5):
... super().__init__(label)
... self.x = x
>>> logistic = MapInteraction(
... 'map', target='value', state='x', function=lambda s: 3 * s.target.x * (1 - s.target.x)
... )
>>> system = System(elements=[ValueElement('value')], interactions=[logistic], time=DiscreteTime())
>>> Simulation(system).check().prepare().run(until=2).trajectory_.values('value.x').tolist()
[0.5, 0.75, 0.5625]
>>> agents = MapInteraction('map', target='walkers', state='x', function=lambda a, m: a.x + 1, form='agent')
>>> type(agents.check().prepare().run(SimpleNamespace()).laws_[0]).__name__
'RuleLaw'
>>> from skcomplex.core.elements import GroupElement
>>> class WeightsElement(Element):
... def __init__(self, label, visual=None, *, w: Annotated[object, StateSpec(vector=True)] = 0.0):
... super().__init__(label, visual=visual)
... self.w = w
>>> group = GroupElement('weights', element=WeightsElement, size=1, values={'w': [1.0, 2.0]})
>>> doubling = MapInteraction(
... 'double', target='weights', state='w', function=lambda a, m: [2 * x for x in a.w], form='agent'
... )
>>> trajectory = Simulation(System(elements=[group], interactions=[doubling])).check().prepare().run(until=2)
>>> trajectory.trajectory_.agents('weights', 2).values['w'].tolist()
[[4.0, 8.0]]
"""
def __init__(
self,
label: str,
*,
target: Annotated[str, RoleSpec()],
state: str,
function: Callable[..., object],
sources: Annotated[Sequence[str], RoleSpec(variadic=True)] = (),
mode: str = 'replace',
schedule: str = 'synchronous',
form: str = 'symbolic',
phase: str | None = None,
) -> None:
super().__init__(label)
self.target = target
self.state = state
self.function = function
self.sources = sources
self.mode = mode
self.schedule = schedule
self.form = form
self.phase = phase
def _check(self) -> None:
"""Raise an error for an unknown mode, schedule, or form, or a function that cannot be called.
Raises:
ValidationError:
The mode, the schedule, or the form is unknown, the symbolic form has the schedule `'member'`, or the
function is not callable.
"""
super()._check()
self._check_variant('mode', self.mode, MODES, (BaseSpec,))
for name, value, variants in (
('schedule', self.schedule, SCHEDULES),
('form', self.form, FORMS),
):
if value not in variants:
error_msg = f'The {name} of map {self.label!r} is {value!r}. Use one of {", ".join(variants)}.'
raise ValidationError(error_msg)
if self.schedule == 'member' and self.form != 'agent':
error_msg = (
f'The map {self.label!r} updates every member in its activation, which only the agent form does.'
)
raise ValidationError(error_msg)
if not callable(self.function):
error_msg = f'The function of map {self.label!r} must be callable, but it is {self.function!r}.'
raise ValidationError(error_msg)
def _apply_mode(self, current: Value, value: Value) -> Value:
"""Return the next value of a state from its current value and the value of the function, by the mode."""
if self.mode == 'increment':
return current + value
if self.mode == 'replace':
return value
raise self._get_variant_error(self.mode, 'mode', '_apply_mode')
def _get_vector(self, agent: AgentViewProtocol, model: ModelViewProtocol) -> Value:
"""Return the next value of a vector state of an agent."""
current = np.asarray(getattr(agent, self.state), dtype=np.float64)
value = np.asarray(self.function(agent, model), dtype=np.float64)
if value.shape != current.shape:
error_msg = (
f'The function of map {self.label!r} returns the shape {value.shape} for the state {self.state!r} of '
f'the shape {current.shape}. Return an array of the shape of the state.'
)
raise ValidationError(error_msg)
return self._apply_mode(current, value)
def _get_value(self, agent: AgentViewProtocol, model: ModelViewProtocol) -> Value:
"""Return the next value of the state of an agent, a number or the array of a vector state."""
if np.ndim(getattr(agent, self.state)) > 0:
return self._get_vector(agent, model)
found = self.function(agent, model)
if not isinstance(found, SupportsFloat):
error_msg = f'The function of map {self.label!r} returns {found!r} for an agent. Return a number.'
raise ValidationError(error_msg)
value = float(found)
return self._apply_mode(float(getattr(agent, self.state)), value)
def _set_value(self, agent: AgentViewProtocol, value: Value) -> None:
"""Set the state of an agent to its next value."""
setattr(agent, self.state, value)
def _update_member(self, agent: AgentViewProtocol, model: ModelViewProtocol) -> None:
"""Set the state of one member to its next value."""
if agent.alive:
self._set_value(agent, self._get_value(agent, model))
def update(self, model: ModelViewProtocol) -> None:
"""Set the state of every agent of the target to its next value, by the schedule.
Args:
model:
The model view of the interaction.
"""
self._update_agents(model, self.schedule, self._get_value, self._set_value)
def _get_order(self) -> object:
"""Return the order of the members of the symbolic form: none, their own order, or a uniform draw."""
if self.schedule == 'synchronous':
return None
if self.schedule == 'sequential':
return SR(0)
return uniform(re.sub(r'\W', '_', f'order_{self.label}'))
def _get_laws(self, namespace: SimpleNamespace) -> list[BasePart]:
"""Return the update of the state to its next value, or the rule of the agent form.
Args:
namespace:
The namespace of the symbols the laws can use, with `namespace.target` and `namespace.sources`.
Returns:
laws:
The update law of the state, or the rule that runs the map of the agents.
"""
if self.form == 'agent' and self.schedule == 'member':
return [RuleLaw(self._update_member, phase=self.phase, members=self._get_role('target'))]
if self.form == 'agent':
return [RuleLaw(self.update, phase=self.phase)]
value = getattr(self._get_member(namespace, 'target'), self.state)
mapped = cast(Value, self.function(namespace))
next_value = self._apply_mode(value, mapped)
return [UpdateLaw(value, next_value, increment=self.mode == 'increment', order=self._get_order())]
class MatchingInteraction[ModelT: ModelViewProtocol](Interaction):
"""Matching of the members of a group of requesters with those of a group of providers by a rule.
Requesters state the highest clearing value they accept in their `limit` state, the quantity they want in their
`quantity` state, and the most value they can give in their `allowance` state, and providers the lowest clearing
value they accept and the quantity they offer. A name given as a pair names the state of the requesters first and of
the providers second, and a missing state counts as no limit, one unit, and no allowance. The rule is a spec that
names the algorithm that pairs the offers and holds its settings: random search, partners, partner switching, a
queue, iterative, continuous, batch, or stable matching. The `value` of a match is its clearing value per unit.
Every match lowers the quantities of both sides and moves every state of `transfers`: a negative sign moves
`value * quantity` from the requester to the provider, and a positive sign moves `quantity` from the provider to the
requester, so the totals stay the same. `on_match(match, model)` adds any other effect. The matching runs
once per round of its phase and reports the quantity-weighted clearing value `value`, the matched `volume`, and the
`unmatched` quantity left, as the reporters `<label>.value`, `<label>.volume`, and `<label>.unmatched`.
Args:
label:
The label of the matching.
requesters:
The path of the group of requesters.
providers:
The path of the group of providers.
rule:
The spec of the algorithm that matches the members, such as `ContinuousMatchingSpec()`, or a function
`rule(requesters, providers, rng, capacity)` of the agents with a positive quantity that returns their
pairs, which match at the midpoint of both limits. A `PartnerSpec` picks among the providers that still
offer a quantity at a limit the requester accepts, in random order of the requesters.
limit:
The state with the limit of both sides, or a pair of the requesters' and the providers' states. An agent
without the default state `'limit'` has no limit as a requester and the limit 0 as a provider.
quantity:
The state with the quantity of both sides, or a pair of the requesters' and the providers' states. An
agent without the default state `'quantity'` offers the quantity 1.
allowance:
The state of a requester with the most value it can give. A requester without the default state
`'allowance'` has no limit on what it gives.
transfers:
The states every match moves, each with the sign `-1` for a move from the requester to the provider or
`1` for a move from the provider to the requester.
phase:
The phase the matching runs in, or `None` for the default phase.
on_match:
A function of every match and the model view, run after the transfers, or `None`.
Raises:
ValidationError:
A transfer has a sign other than `-1` or `1`, or the rule is neither a matching spec nor a function, or
it is invalid, raised by `check`, or an agent lacks a state other than a default one that names its
limit, quantity, or allowance, raised by `run`.
Examples:
>>> from skcomplex.core.interactions import MatchingInteraction
>>> from skcomplex.specs import ContinuousMatchingSpec
>>> matching = MatchingInteraction(
... 'exchange',
... requesters='traders',
... providers='traders',
... rule=ContinuousMatchingSpec(),
... limit=('bid', 'ask'),
... )
>>> matching.check().reports
('value', 'volume', 'unmatched')
"""
reports: ClassVar[tuple[str, ...]] = ('value', 'volume', 'unmatched')
def __init__(
self,
label: str,
*,
requesters: Annotated[str, RoleSpec()],
providers: Annotated[str, RoleSpec()],
rule: MatchingRule | Pairing = RandomSearchSpec(),
limit: str | tuple[str, str] = 'limit',
quantity: str | tuple[str, str] = 'quantity',
allowance: str = 'allowance',
transfers: Sequence[tuple[str, int]] = (),
phase: str | None = None,
on_match: Callable[[MatchSpec, ModelT], None] | None = None,
) -> None:
super().__init__(label)
self.requesters = requesters
self.providers = providers
self.rule = rule
self.limit = limit
self.quantity = quantity
self.allowance = allowance
self.transfers = transfers
self.phase = phase
self.on_match = on_match
@staticmethod
def _get_sides(name: str | tuple[str, str]) -> tuple[str, str]:
"""Return the state of the requesters and of the providers of a name given once or as a pair."""
return (name, name) if isinstance(name, str) else name
def _read(self, agent: AgentViewProtocol, name: str, default: float) -> float:
"""Return a state of an agent, or a default when it lacks a state of a default name."""
if hasattr(agent, name):
return float(getattr(agent, name))
if name in DEFAULT_STATES:
return default
error_msg = (
f'The agents of {agent.group!r} have no state {name!r}, which the matching {self.label!r} reads. Declare '
f'the state on their element, or name one of their states.'
)
raise ValidationError(error_msg)
@staticmethod
def _get_affordable(offer: SimpleNamespace, value: float) -> float:
"""Return the quantity a participant can still match at a value within its allowance."""
if value <= 0:
return float(offer.quantity)
return max(0.0, min(offer.quantity, offer.allowance / value))
@staticmethod
def _book(trades: list[MatchSpec], requester: SimpleNamespace, provider: SimpleNamespace, value: float) -> None:
"""Record a match of the largest quantity both sides can match at a value and lower both offers."""
quantity = min(MatchingInteraction._get_affordable(requester, value), provider.quantity)
if quantity <= 0:
return
requester.quantity -= quantity
requester.allowance -= quantity * value
provider.quantity -= quantity
trades.append(MatchSpec(requester.agent, provider.agent, quantity, value))
@staticmethod
def _get_value(requester: SimpleNamespace, provider: SimpleNamespace, clearing: str) -> float:
"""Return the value of a match under a clearing rule."""
if clearing == 'requester':
return requester.limit
if clearing == 'provider':
return provider.limit
return (requester.limit + provider.limit) / 2
@staticmethod
def _order(offers: list[SimpleNamespace], order: str, rng: np.random.Generator) -> list[SimpleNamespace]:
"""Return offers in random order, by decreasing limit, or in order of creation."""
if order == 'limit':
return sorted(offers, key=lambda item: (-item.limit, item.agent.id))
if order == 'sequential':
return sorted(offers, key=lambda item: item.agent.id)
return [offers[index] for index in rng.permutation(len(offers))]
def _get_offers(self, model: ModelT) -> tuple[list[SimpleNamespace], list[SimpleNamespace]]:
"""Return the offers of the requesters and of the providers with a positive quantity, which matches lower."""
limits, quantities = self._get_sides(self.limit), self._get_sides(self.quantity)
requesters = [
SimpleNamespace(
agent=agent,
limit=self._read(agent, limits[0], math.inf),
quantity=self._read(agent, quantities[0], 1.0),
allowance=self._read(agent, self.allowance, math.inf),
)
for agent in model.agents(self._get_role('requesters'))
]
providers = [
SimpleNamespace(
agent=agent,
limit=self._read(agent, limits[1], 0.0),
quantity=self._read(agent, quantities[1], 1.0),
allowance=math.inf,
)
for agent in model.agents(self._get_role('providers'))
]
return [offer for offer in requesters if offer.quantity > 0], [
offer for offer in providers if offer.quantity > 0
]
def _match_random_search(
self,
rule: RandomSearchSpec,
requesters: list[SimpleNamespace],
providers: list[SimpleNamespace],
rng: np.random.Generator,
) -> list[MatchSpec]:
"""Return the matches of requesters in random order that meet random providers and match on crossing limits."""
trades: list[MatchSpec] = []
for index in rng.permutation(len(requesters)):
requester = requesters[index]
for _ in range(rule.tries):
available = [provider for provider in providers if provider.quantity > 0]
if requester.quantity <= 0 or not available:
break
provider = available[int(rng.integers(len(available)))]
if provider.limit <= requester.limit:
self._book(trades, requester, provider, self._get_value(requester, provider, rule.clearing))
return trades
@staticmethod
def _choose_partner(
rule: PartnerSwitchSpec,
best: SimpleNamespace,
current: SimpleNamespace | None,
rng: np.random.Generator,
) -> SimpleNamespace:
"""Return the provider a requester matches with, its current partner or the best of its sample."""
if current is None or current.quantity <= 0:
return best
if best.limit >= current.limit:
return current
if best.limit <= 0:
return best
switching = 1 - math.exp(rule.intensity * (best.limit - current.limit) / best.limit)
return best if rng.random() < switching else current
def _match_partner_switch(
self,
rule: PartnerSwitchSpec,
requesters: list[SimpleNamespace],
providers: list[SimpleNamespace],
rng: np.random.Generator,
) -> list[MatchSpec]:
"""Return the matches of every requester with its kept or new partner, at the partner's limit."""
trades: list[MatchSpec] = []
by_id = {provider.agent.id: provider for provider in providers}
for index in rng.permutation(len(requesters)):
requester = requesters[index]
available = [provider for provider in providers if provider.quantity > 0]
if not available:
break
picks = rng.choice(len(available), size=min(rule.sample, len(available)), replace=False)
best = min((available[int(pick)] for pick in picks), key=lambda provider: provider.limit)
current = by_id.get(int(getattr(requester.agent, rule.partner, -1)))
chosen = self._choose_partner(rule, best, current, rng)
if hasattr(requester.agent, rule.partner):
setattr(requester.agent, rule.partner, chosen.agent.id)
if chosen.limit <= requester.limit:
self._book(trades, requester, chosen, chosen.limit)
return trades
def _match_queue(
self,
rule: QueueSpec,
requesters: list[SimpleNamespace],
providers: list[SimpleNamespace],
rng: np.random.Generator,
) -> list[MatchSpec]:
"""Return the matches of the queue of requesters served by the lowest providers at their limits."""
queue = self._order(requesters, rule.order, rng)
stock = sorted(providers, key=lambda provider: provider.limit)
trades: list[MatchSpec] = []
for requester in queue:
for provider in stock:
if requester.quantity <= 0 or provider.limit > requester.limit:
break
if provider.quantity > 0:
self._book(trades, requester, provider, provider.limit)
return trades
@staticmethod
def _adjust(
rule: IterativeMatchingSpec,
requesters: list[SimpleNamespace],
providers: list[SimpleNamespace],
start: float,
) -> float:
"""Return the value the tâtonnement reaches from a start."""
value = start
for _ in range(rule.iterations):
requested = sum(
MatchingInteraction._get_affordable(requester, value)
for requester in requesters
if requester.limit >= value
)
provided = sum(provider.quantity for provider in providers if provider.limit <= value)
total = requested + provided
if not total or abs(requested - provided) / total < rule.tolerance:
break
value *= 1 + rule.adjustment * (requested - provided) / total
return value
def _match_at(
self,
requesters: list[SimpleNamespace],
providers: list[SimpleNamespace],
value: float,
rng: np.random.Generator,
) -> list[MatchSpec]:
"""Return the matches at one value between the willing requesters and providers in random order."""
trades: list[MatchSpec] = []
willing_requesters = self._order([item for item in requesters if item.limit >= value], 'random', rng)
willing_providers = self._order([item for item in providers if item.limit <= value], 'random', rng)
pairs = iter(willing_providers)
provider = next(pairs, None)
for requester in willing_requesters:
while provider is not None and MatchingInteraction._get_affordable(requester, value) > 0:
self._book(trades, requester, provider, value)
if provider.quantity <= 0:
provider = next(pairs, None)
return trades
def _match_iterative(
self,
rule: IterativeMatchingSpec,
requesters: list[SimpleNamespace],
providers: list[SimpleNamespace],
rng: np.random.Generator,
storage: MutableMapping[str, object],
) -> list[MatchSpec]:
"""Return the matches at the value of the tâtonnement, which starts from the value of the last clearing."""
limits = [offer.limit for offer in requesters + providers if offer.limit > 0]
if not limits:
return []
last = storage.get('value')
start = last if isinstance(last, float) and math.isfinite(last) and last > 0 else float(np.median(limits))
value = self._adjust(rule, requesters, providers, start)
storage['value'] = value
return self._match_at(requesters, providers, value, rng)
@staticmethod
def _arrange(
rule: ContinuousMatchingSpec,
requesters: list[SimpleNamespace],
providers: list[SimpleNamespace],
rng: np.random.Generator,
) -> list[tuple[SimpleNamespace, bool]]:
"""Return the offers in their order of arrival, each with whether it is a request."""
offers = [(offer, True) for offer in requesters] + [(offer, False) for offer in providers]
if rule.arrival == 'sequential':
return sorted(offers, key=lambda item: item[0].agent.id)
return [offers[index] for index in rng.permutation(len(offers))]
def _match_continuous(
self,
rule: ContinuousMatchingSpec,
requesters: list[SimpleNamespace],
providers: list[SimpleNamespace],
rng: np.random.Generator,
) -> list[MatchSpec]:
"""Return the matches of every arriving offer with the best resting offers of the other side."""
requests: list[tuple[float, int, SimpleNamespace]] = []
provisions: list[tuple[float, int, SimpleNamespace]] = []
trades: list[MatchSpec] = []
for time, (offer, is_request) in enumerate(self._arrange(rule, requesters, providers, rng)):
if is_request:
while (
provisions
and provisions[0][0] <= offer.limit
and MatchingInteraction._get_affordable(offer, provisions[0][0]) > 0
):
resting = provisions[0][2]
self._book(trades, offer, resting, resting.limit)
if resting.quantity <= 0:
heapq.heappop(provisions)
if MatchingInteraction._get_affordable(offer, offer.limit) > 0:
heapq.heappush(requests, (-offer.limit, time, offer))
continue
while requests and -requests[0][0] >= offer.limit and offer.quantity > 0:
resting = requests[0][2]
self._book(trades, resting, offer, resting.limit)
if MatchingInteraction._get_affordable(resting, resting.limit) <= 0:
heapq.heappop(requests)
if offer.quantity > 0:
heapq.heappush(provisions, (offer.limit, time, offer))
return trades
@staticmethod
def _get_requested(offer: SimpleNamespace) -> float:
"""Return the quantity a requester can match at its own limit."""
return (
offer.quantity
if not math.isfinite(offer.allowance)
else MatchingInteraction._get_affordable(offer, offer.limit)
)
def _pair(
self,
requesters: list[SimpleNamespace],
providers: list[SimpleNamespace],
) -> tuple[list[tuple[SimpleNamespace, SimpleNamespace]], float]:
"""Return the pairs of crossing offers in order of their limits and the value of the last pair."""
ranked_requesters = sorted(requesters, key=lambda offer: (-offer.limit, offer.agent.id))
ranked_providers = sorted(providers, key=lambda offer: (offer.limit, offer.agent.id))
left = {id(offer): self._get_requested(offer) for offer in ranked_requesters}
offered = {id(offer): offer.quantity for offer in ranked_providers}
pairs: list[tuple[SimpleNamespace, SimpleNamespace]] = []
marginal = math.nan
requester_iter, provider_iter = iter(ranked_requesters), iter(ranked_providers)
requester, provider = next(requester_iter, None), next(provider_iter, None)
while requester is not None and provider is not None and requester.limit >= provider.limit:
pairs.append((requester, provider))
marginal = (requester.limit + provider.limit) / 2
quantity = min(left[id(requester)], offered[id(provider)])
left[id(requester)] -= quantity
offered[id(provider)] -= quantity
if left[id(requester)] <= 0:
requester = next(requester_iter, None)
if offered[id(provider)] <= 0:
provider = next(provider_iter, None)
return pairs, marginal
def _match_batch(self, requesters: list[SimpleNamespace], providers: list[SimpleNamespace]) -> list[MatchSpec]:
"""Return the matches of the crossing offers at the midpoint of the last crossing pair."""
pairs, marginal = self._pair(requesters, providers)
trades: list[MatchSpec] = []
for requester, provider in pairs:
self._book(trades, requester, provider, marginal)
return trades
@staticmethod
def _score_provider(rule: StableMatchingSpec, requester: SimpleNamespace, provider: SimpleNamespace) -> float:
"""Return the score of a provider for a requester, higher for a preferred provider."""
if rule.requester_preference is None:
return -provider.limit
return rule.requester_preference(requester.agent, provider.agent)
@staticmethod
def _score_requester(rule: StableMatchingSpec, provider: SimpleNamespace, requester: SimpleNamespace) -> float:
"""Return the score of a requester for a provider, higher for a preferred requester."""
if rule.provider_preference is None:
return requester.limit
return rule.provider_preference(provider.agent, requester.agent)
def _rank_providers(
self,
rule: StableMatchingSpec,
requesters: list[SimpleNamespace],
providers: list[SimpleNamespace],
) -> dict[int, list[int]]:
"""Return the ids of the acceptable providers of every requester, the preferred first."""
capacities = {provider.agent.id for provider in providers if int(provider.quantity) > 0}
ranks = {}
for requester in requesters:
acceptable = [
provider
for provider in providers
if provider.agent.id in capacities and provider.limit <= requester.limit
]
ranked = sorted(
(-self._score_provider(rule, requester, provider), provider.agent.id) for provider in acceptable
)
if ranked:
ranks[requester.agent.id] = [identity for _, identity in ranked]
return ranks
def _rank_requesters(
self,
rule: StableMatchingSpec,
ranks: dict[int, list[int]],
providers: list[SimpleNamespace],
by_id: dict[int, SimpleNamespace],
) -> dict[int, list[int]]:
"""Return the ids of the requesters that rank every provider, the preferred first."""
provider_ranks = {}
for provider in providers:
suitors = sorted(
(-self._score_requester(rule, provider, by_id[requester_id]), requester_id)
for requester_id, ranked in ranks.items()
if provider.agent.id in ranked
)
if suitors:
provider_ranks[provider.agent.id] = [identity for _, identity in suitors]
return provider_ranks
def _match_stable(
self,
rule: StableMatchingSpec,
requesters: list[SimpleNamespace],
providers: list[SimpleNamespace],
) -> list[MatchSpec]:
"""Return the matches of the requester-optimal stable matching, one unit per pair."""
by_id = {offer.agent.id: offer for offer in itertools.chain(requesters, providers)}
requester_ranks = self._rank_providers(rule, requesters, providers)
if not requester_ranks:
return []
provider_ranks = self._rank_requesters(rule, requester_ranks, providers, by_id)
capacities = {provider_id: int(by_id[provider_id].quantity) for provider_id in provider_ranks}
with warnings.catch_warnings():
warnings.simplefilter('ignore')
game = HospitalResident.create_from_dictionaries(requester_ranks, provider_ranks, capacities, clean=True)
solution = game.solve(optimal='resident')
trades: list[MatchSpec] = []
for hospital, residents in solution.items():
provider = by_id[hospital.name]
for resident in residents:
requester = by_id[resident.name]
value = self._get_value(requester, provider, rule.clearing)
quantity = min(1.0, MatchingInteraction._get_affordable(requester, value), provider.quantity)
if quantity > 0:
requester.quantity -= quantity
requester.allowance -= quantity * value
provider.quantity -= quantity
trades.append(MatchSpec(requester.agent, provider.agent, quantity, value))
return trades
def _match_partners(
self,
rule: PartnerSpec,
requesters: list[SimpleNamespace],
providers: list[SimpleNamespace],
rng: np.random.Generator,
) -> list[MatchSpec]:
"""Return the matches of every requester, in random order, with the feasible partners its rule picks."""
wanting = {id(offer.agent): offer for offer in requesters}
offering = {id(offer.agent): offer for offer in providers}
def is_feasible(requester: AgentViewProtocol, provider: AgentViewProtocol) -> bool:
"""Return whether a provider still offers a quantity at a limit the requester accepts."""
offer = offering[id(provider)]
return offer.quantity > 0 and offer.limit <= wanting[id(requester)].limit
trades: list[MatchSpec] = []
left, right = [offer.agent for offer in requesters], [offer.agent for offer in providers]
for requester, provider in MatchLaw(self.requesters, self.providers, rule).iterate_pairs(
left,
right,
rng,
is_feasible,
):
first, second = wanting[id(requester)], offering[id(provider)]
self._book(trades, first, second, self._get_value(first, second, rule.clearing))
return trades
def _match_function(
self,
rule: Pairing,
requesters: list[SimpleNamespace],
providers: list[SimpleNamespace],
rng: np.random.Generator,
) -> list[MatchSpec]:
"""Return the matches of the pairs of a function, each at the midpoint of the limits."""
wanting = {id(offer.agent): offer for offer in requesters}
offering = {id(offer.agent): offer for offer in providers}
trades: list[MatchSpec] = []
left = [offer.agent for offer in requesters]
right = left if self.requesters == self.providers else [offer.agent for offer in providers]
for requester, provider in MatchLaw(self.requesters, self.providers, rule).get_pairs(left, right, rng):
first, second = wanting[id(requester)], offering[id(provider)]
self._book(trades, first, second, self._get_value(first, second, 'midpoint'))
return trades
def _get_matches(
self,
requesters: list[SimpleNamespace],
providers: list[SimpleNamespace],
model: ModelT,
) -> list[MatchSpec]:
"""Return the matches of the offers by the algorithm the rule names, none when a side has no offers."""
if not requesters or not providers:
return []
rule, rng = self.rule, model.rng
match rule:
case RandomSearchSpec():
matches = self._match_random_search(rule, requesters, providers, rng)
case PartnerSwitchSpec():
matches = self._match_partner_switch(rule, requesters, providers, rng)
case QueueSpec():
matches = self._match_queue(rule, requesters, providers, rng)
case IterativeMatchingSpec():
matches = self._match_iterative(rule, requesters, providers, rng, model.storage)
case ContinuousMatchingSpec():
matches = self._match_continuous(rule, requesters, providers, rng)
case BatchMatchingSpec():
matches = self._match_batch(requesters, providers)
case StableMatchingSpec():
matches = self._match_stable(rule, requesters, providers)
case PartnerSpec():
matches = self._match_partners(rule, requesters, providers, rng)
case _:
matches = self._match_function(rule, requesters, providers, rng)
return matches
def _settle(self, match: MatchSpec, model: ModelT) -> None:
"""Move the transfers of a match, lower the quantities of both sides, and run `on_match`."""
for state, sign in self.transfers:
if sign < 0:
transfer(match.requester, match.provider, match.quantity * match.value, state)
else:
transfer(match.provider, match.requester, match.quantity, state)
for agent, name in zip((match.requester, match.provider), self._get_sides(self.quantity), strict=True):
if hasattr(agent, name):
setattr(agent, name, getattr(agent, name) - match.quantity)
if self.on_match is not None:
self.on_match(match, model)
def _report_matches(self, matches: list[MatchSpec], unmatched: float, model: ModelT) -> None:
"""Report the quantity-weighted value, the matched volume, and the unmatched quantity of one clearing.
Args:
matches:
The matches of the clearing.
unmatched:
The quantity of both sides left after the clearing.
model:
The model view of the matching.
"""
volume = sum(match.quantity for match in matches)
value = sum(match.quantity * match.value for match in matches) / volume if volume else math.nan
model.report('value', value)
model.report('volume', volume)
model.report('unmatched', unmatched)
def _check_space(self, space: BaseSpace | None) -> None:
"""Raise an error for candidates in neighbouring cells or nodes in a continuous space."""
self._check_partners(self.rule, space)
def _check(self) -> None:
"""Raise an error for a declaration the matching cannot take, a transfer without a sign, or a bad rule.
Raises:
ValidationError:
A transfer has a sign other than `-1` or `1`, the rule is neither a matching spec nor a function, or
it is invalid.
"""
super()._check()
for state, sign in self.transfers:
if sign not in SIGNS:
error_msg = f'The transfer of {state!r} has the sign {sign!r}. Use one of {SIGNS}.'
raise ValidationError(error_msg)
self._check_variant('rule', self.rule, (), RULES, functions=True)
if isinstance(self.rule, RULES):
self.rule.check()
def _get_laws(self, namespace: SimpleNamespace) -> Sequence[BasePart]:
"""Return the rule that matches the members in the matching's phase.
Args:
namespace:
The namespace of the symbols the laws can use.
Returns:
laws:
The laws of the interaction.
"""
return [RuleLaw(self.match, phase=self.phase)]
def match(self, model: ModelT) -> None:
"""Match the members, settle the matches, and report the value, the volume, and the unmatched quantity.
Args:
model:
The model view of the matching.
"""
requesters, providers = self._get_offers(model)
matches = self._get_matches(requesters, providers, model)
for match in matches:
self._settle(match, model)
unmatched = sum(offer.quantity for offer in requesters + providers)
self._report_matches(matches, unmatched, model)