Labour market
Hamill and Gilbert's labour market of firms posting wage offers and unemployed workers searching.
- Name:
labour_market - Field:
economics/markets - Formalism: agent-based model
- References: Hamill and Gilbert 2016 ch. 7
Factory
@register_model(
field='economics/markets',
tags=('labour', 'market', 'matching', 'agents'),
references=('Hamill and Gilbert 2016 ch. 7',),
defaults={'until': 200},
observables=lambda system: [
ObservableSpec('unemployment', group='workers', variable='employer', reduce=_unemployment),
ObservableSpec('searching', group='workers', variable='supply', reduce='sum'),
],
)
def labour_market(
n_workers: int = 200,
n_firms: int = 40,
*,
separation: float = 0.05,
vacancy_rate: float = 0.3,
reservation: float = 1.0,
decline: float = 0.1,
low_wage: float = 0.5,
high_wage: float = 1.0,
unemployment: float = 0.5,
random_state: int | np.random.Generator | None = 0,
) -> System:
"""Return Hamill and Gilbert's labour market of firms posting wage offers and unemployed workers searching.
In the phase of offers, jobs are destroyed at the separation rate `s`, every firm posts a vacancy with a uniform
wage offer at the vacancy rate, and unemployed workers set reservation wages that fall with the duration of their
search. In the phase of matching, a deferred-acceptance market pairs vacancies with workers whose reservation wage
they cover, every worker getting the best offer of a stable matching. The unemployment rate settles at the
flow-balance value `s / (s + f)`, with `f` the share of searching workers hired in a step.
Args:
n_workers:
The number of workers.
n_firms:
The number of firms, each with at most one vacancy per step.
separation:
The probability `s` that a job is destroyed in a step.
vacancy_rate:
The probability that a firm posts a vacancy in a step.
reservation:
The reservation wage at the start of a search.
decline:
The relative fall of the reservation wage per step of search.
low_wage:
The lowest wage offer.
high_wage:
The highest wage offer.
unemployment:
The initial share of unemployed workers.
random_state:
The seed or generator of the initial employment.
Returns:
system:
The model, with the groups `firms` and `workers`, the market `labour` with the reporters `labour.value`
(the mean wage of the hires) and `labour.volume` (the hires), and the catalog observables of the
unemployment rate `unemployment` and of the searching workers `searching`.
Examples:
>>> from skcomplex.models.economics.markets import labour_market
>>> [group.size for group in labour_market(20, 5).elements]
[5, 20]
"""
rng = np.random.default_rng(random_state)
employed = rng.random(n_workers) >= unemployment
employers = np.where(employed, rng.integers(n_firms, size=n_workers), -1).astype(float)
firms = GroupElement(
'firms',
element=EmployerElement,
size=n_firms,
values={'vacancy_rate': vacancy_rate, 'low_wage': low_wage, 'high_wage': high_wage, 'bid': low_wage},
)
workers = GroupElement(
'workers',
element=JobseekerElement,
size=n_workers,
values={
'employer': employers,
'wage': np.where(employed, (low_wage + high_wage) / 2, 0.0),
'supply': 0.0,
'ask': reservation,
'separation': separation,
'reservation': reservation,
'decline': decline,
},
)
offers = [
MapInteraction(f'{group}_{name}', target=group, state=name, function=function, form='agent', phase='offers')
for group, steps in (('firms', EmployerElement.get_posting()), ('workers', JobseekerElement.get_search()))
for name, function in steps
]
losses = TransitionInteraction(
'workers_separation',
target='workers',
state='employer',
source=lambda value: value >= 0,
destination=-1.0,
probability=lambda agent, model: agent.separation,
form='agent',
phase='offers',
)
spells = [
MapInteraction(f'workers_{name}', target='workers', state=name, function=function, form='agent', phase='offers')
for name, function in JobseekerElement.get_spell()
]
market = MatchingInteraction(
'labour',
requesters='firms',
providers='workers',
rule=StableMatchingSpec(clearing='requester'),
limit=('bid', 'ask'),
quantity=('demand', 'supply'),
phase='matching',
on_match=hire,
)
return System(
elements=[firms, workers],
interactions=[
*offers,
losses,
*spells,
market,
],
time=DiscreteTime(schedule=PhasesScheduleSpec([PhaseSpec('offers'), PhaseSpec('matching')])),
)
Components
class EmployerElement(Element):
"""Firm that posts a vacancy with a random wage offer in some steps, filled by at most one worker.
Args:
label:
The label of the agent or group.
visual:
How the element is drawn, or `None` for `default_visual_`.
bid:
The wage offered for the vacancy.
demand:
The number of open vacancies, one or zero.
posted:
The step in which the vacancy was posted.
vacancy_rate:
The probability of posting a vacancy in a step.
low_wage:
The lowest wage offer.
high_wage:
The highest wage offer.
Examples:
>>> from skcomplex.core.elements import GroupElement
>>> from skcomplex.models.economics.markets import EmployerElement
>>> GroupElement('firms', element=EmployerElement, size=5).size
5
"""
def __init__(
self,
label: str,
visual: VisualSpec | None = None,
*,
bid: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
demand: Annotated[float, StateSpec(domain=NN)] = 0.0,
posted: Annotated[float, StateSpec()] = -1.0,
vacancy_rate: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.5,
low_wage: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.5,
high_wage: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
) -> None:
super().__init__(label, visual=visual)
self.bid = bid
self.demand = demand
self.posted = posted
self.vacancy_rate = vacancy_rate
self.low_wage = low_wage
self.high_wage = high_wage
@classmethod
def _get_view(cls) -> type[EmployerView]:
"""Return the view of an employer, with its vacancy and the parameters of its offers."""
return EmployerView
@staticmethod
def get_posting() -> tuple[tuple[str, Callable[[EmployerView, ModelView], float]], ...]:
"""Return the states a firm sets in the phase of offers, in their order.
The firm withdraws its last vacancy and posts a new one at the vacancy rate, with a uniform wage offer and the
step of posting.
Returns:
posting:
The name of every state and the function of its value, of the firm and the model view.
"""
return (
('demand', lambda agent, model: float(model.rng.random() < agent.vacancy_rate)),
(
'bid',
lambda agent, model: (
float(model.rng.uniform(agent.low_wage, agent.high_wage)) if agent.demand else agent.bid
),
),
('posted', lambda agent, model: model.time if agent.demand else agent.posted),
)
class JobseekerElement(Element):
"""Worker who loses its job at the separation rate and, while unemployed, lowers its reservation wage.
A worker who loses its job searches from the next step. An unemployed worker asks the reservation wage
`w_0 (1 - decline)^d`, where `d` is the number of steps of search, and is hired by the matching market.
Args:
label:
The label of the agent or group.
visual:
How the element is drawn, or `None` for `default_visual_`.
ask:
The reservation wage.
supply:
One while the worker searches in the current step, zero otherwise.
employer:
The id of the employer, or -1 for an unemployed worker.
wage:
The wage of the current job.
duration:
The number of steps of the current spell of search.
hired:
The step of the last hire.
offer:
The step in which the offer of the last hire was posted.
separation:
The probability that a job is destroyed in a step.
reservation:
The reservation wage at the start of a search.
decline:
The relative fall of the reservation wage per step of search.
Examples:
>>> from skcomplex.core.elements import GroupElement
>>> from skcomplex.models.economics.markets import JobseekerElement
>>> GroupElement('workers', element=JobseekerElement, size=5).size
5
"""
def __init__(
self,
label: str,
visual: VisualSpec | None = None,
*,
ask: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
supply: Annotated[float, StateSpec(domain=NN)] = 1.0,
employer: Annotated[float, StateSpec(reference='firms')] = -1.0,
wage: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
duration: Annotated[float, StateSpec(domain=NN)] = 0.0,
hired: Annotated[float, StateSpec()] = -1.0,
offer: Annotated[float, StateSpec()] = -1.0,
separation: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.05,
reservation: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
decline: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.1,
) -> None:
super().__init__(label, visual=visual)
self.ask = ask
self.supply = supply
self.employer = employer
self.wage = wage
self.duration = duration
self.hired = hired
self.offer = offer
self.separation = separation
self.reservation = reservation
self.decline = decline
@classmethod
def _get_view(cls) -> type[JobseekerView]:
"""Return the view of a worker, with its search and job states and their parameters."""
return JobseekerView
@staticmethod
def get_search() -> tuple[tuple[str, Callable[[JobseekerView, ModelView], float]], ...]:
"""Return the states a worker sets in the phase of offers before the separations, in their order.
A worker unemployed at the start of the step searches, with a reservation wage that falls with the duration of
its search.
Returns:
search:
The name of every state and the function of its value, of the worker and the model view.
"""
return (
('supply', lambda agent, model: float(agent.employer < 0)),
(
'ask',
lambda agent, model: (
agent.reservation * (1 - agent.decline) ** agent.duration if agent.supply else agent.ask
),
),
)
@staticmethod
def get_spell() -> tuple[tuple[str, Callable[[JobseekerView, ModelView], float]], ...]:
"""Return the states a worker sets in the phase of offers after the separations, in their order.
A worker who lost its job earns no wage, and the duration of a search grows while the worker searches.
Returns:
spell:
The name of every state and the function of its value, of the worker and the model view.
"""
return (
('wage', lambda agent, model: 0.0 if agent.employer < 0 else agent.wage),
(
'duration',
lambda agent, model: (
agent.duration + 1 if agent.supply else 0.0 if agent.employer < 0 else agent.duration
),
),
)
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 TransitionInteraction(Interaction):
"""Move of a discrete state of the target from one value to another, or between all its values by a matrix.
A member whose state holds `source` moves to `destination` at a `rate`, as a jump in continuous time, with a
`probability` in every step of discrete time, or in the step in which a `condition` holds. With `over='neighbors'`
the rate and the probability are those of one neighbour, written with `namespace.target.other` for the neighbour:
the rates of the neighbours add up, and their probabilities combine as `1 - prod(1 - p)`, the probability that at
least one of them moves the member. With a matrix instead, the state takes the values `0` to `n - 1` and steps by
the rows of a transition matrix in discrete time, the row of its current value. An `emission` matrix then draws the
`observed` state of the target from the row that the next value of the state selects, as a hidden Markov source.
With a residence time `after`, a member moves by a probability or a condition only once it has held the source
value for `after` steps, at the earliest at the end of the step that completes them.
In the `'symbolic'` form, every function reads `namespace.target`, the states and parameters of the target, and the
time `namespace.t`. In the `'agent'` form, the target is an agent group and every function takes the agent and
the model view, `f(agent, model)`, and returns a number, a truth value, or the rows of a matrix, and a rate `r`
moves an agent with probability `1 - exp(-r)` in a step. With `over='cell'`, every member acts on the other members
of its cell: it moves each one that holds the source value with the probability, the rate, or the condition
`f(agent, other, model)` of that member `agent` and the acting member `other`, each by its own draw.
The `schedule` of the agent form orders the members. Under `'synchronous'` every member decides from the states at
the start of the step, under `'sequential'` the members act one after the other in their order, and under
`'random-sequential'` in a new random order every step, each reading the values the members before it set. Under
`'member'` every member acts in its own activation, together with the rules of its element and the other member
interactions, in the order the system lists them, so one rule of a member runs its movement, transfers,
transitions, and life cycle.
Args:
label:
The label of the interaction, unique among the parts of its system.
target:
The path of the element or group whose state moves.
state:
The name of the discrete state of the target.
source:
The value the state moves from, a collection of such values, or a function of the value of the state that
returns whether the state moves from it, such as `lambda value: value >= 0`, a condition in the symbolic
form. A state that moves from several values moves to the destination from each.
destination:
The value the state moves to.
rate:
The function of the rate of the move, or `None`.
probability:
The function of the probability of the move in a step, or `None`.
condition:
The function of the condition under which the state moves in a step, or `None`.
matrix:
The function of the transition matrix, one row and one column per value of the state, or `None`.
emission:
With a matrix, the function of the emission matrix, one row per value of the state and one column per value
of the observed state, or `None`.
observed:
The name of the state of the target that the emission matrix draws, or `None`.
over:
`'member'` for a rate or a probability of the member, `'neighbors'` for that of one neighbour, or `'cell'`
for the agent form whose members act on the members of their cell.
neighborhood:
The neighbourhood of `over='neighbors'`, or `None` for the neighbours of the network or the lattice.
after:
The number of steps a member holds the source value before it can move, or `None`.
schedule:
The order of the members of the agent form, `'synchronous'`, `'sequential'`, `'random-sequential'`, or
`'member'`.
form:
`'symbolic'` for 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 transition has not exactly one kind, the source and the destination are equal, an emission has no
matrix or no observed state, `over` is unknown or goes with a form or a kind it cannot take, the
residence time is negative or goes with a rate, a matrix, or members acting on their cell, the form or the
schedule is unknown, the symbolic form has a schedule, or a function is not callable.
Examples:
>>> from skcomplex.core.interactions import TransitionInteraction
>>> recovery = TransitionInteraction(
... 'recovery', target='hosts', state='state', source=1, destination=0, rate=lambda s: 0.5
... )
>>> recovery.check().over
'member'
>>> agents = TransitionInteraction(
... 'recovery', target='hosts', state='state', source=1, destination=2, probability=lambda a, m: 0.1,
... after=3, form='agent'
... )
>>> type(agents.check().prepare().run(SimpleNamespace()).laws_[0]).__name__
'RuleLaw'
>>> separation = TransitionInteraction(
... 'separation', target='workers', state='employer', source=lambda value: value >= 0, destination=-1,
... probability=lambda a, m: 0.05, form='agent'
... )
>>> separation.check().destination
-1
"""
def __init__(
self,
label: str,
*,
target: Annotated[str, RoleSpec()],
state: str = 'state',
source: SourceValues = 0.0,
destination: float = 1.0,
rate: Callable[..., SupportsFloat] | None = None,
probability: Callable[..., SupportsFloat] | None = None,
condition: Callable[..., object] | None = None,
matrix: Callable[..., Sequence[Sequence[object]]] | None = None,
emission: Callable[..., Sequence[Sequence[object]]] | None = None,
observed: str | None = None,
over: str = 'member',
neighborhood: BaseNeighborhood | None = None,
after: float | None = None,
schedule: str = 'synchronous',
form: str = 'symbolic',
phase: str | None = None,
) -> None:
super().__init__(label)
self.target = target
self.state = state
self.source = source
self.destination = destination
self.rate = rate
self.probability = probability
self.condition = condition
self.matrix = matrix
self.emission = emission
self.observed = observed
self.over = over
self.neighborhood = neighborhood
self.after = after
self.schedule = schedule
self.form = form
self.phase = phase
def _check_kind(self) -> str:
"""Raise an error for a transition without exactly one callable kind, or with an emission it cannot take."""
kinds = [kind for kind in KINDS if getattr(self, kind) is not None]
if len(kinds) != 1 or not callable(getattr(self, kinds[0])):
error_msg = (
f'The transition {self.label!r} has the forms {kinds}. Give exactly one of {", ".join(KINDS)}, as a '
'function of the namespace.'
)
raise ValidationError(error_msg)
unpaired = (self.emission is None) != (self.observed is None)
if unpaired or (self.emission is not None and self.matrix is None):
error_msg = (
f'The transition {self.label!r} has the emission {self.emission!r} of the state {self.observed!r}. '
'Give both with a matrix, or neither.'
)
raise ValidationError(error_msg)
return kinds[0]
def _check_form(self) -> None:
"""Raise an error for a transition without exactly one callable kind, or with arguments its kind ignores."""
kind = self._check_kind()
if self.over not in OVERS:
error_msg = f'The transition {self.label!r} runs over {self.over!r}. Use one of {", ".join(OVERS)}.'
raise ValidationError(error_msg)
if self.form not in FORMS:
error_msg = f'The form of transition {self.label!r} is {self.form!r}. Use one of {", ".join(FORMS)}.'
raise ValidationError(error_msg)
if self.schedule not in SCHEDULES or (self.form == 'symbolic' and self.schedule != 'synchronous'):
error_msg = (
f'The schedule of transition {self.label!r} is {self.schedule!r}. Use one of {", ".join(SCHEDULES)}, '
'and the agent form for another than `synchronous`.'
)
raise ValidationError(error_msg)
if self.over == 'cell' and (self.form != 'agent' or kind == 'matrix' or self.after is not None):
error_msg = (
f'The members of transition {self.label!r} act on their cell, which only the agent form does, by a '
'rate, a probability, or a condition and without a residence time.'
)
raise ValidationError(error_msg)
if self.over == 'neighbors' and (kind not in {'rate', 'probability'} or self.form == 'agent'):
error_msg = (
f'The transition {self.label!r} runs over neighbours, which only a rate or a probability of the '
'symbolic form does. In the agent form, read the neighbours of the agent in the function.'
)
raise ValidationError(error_msg)
single = not (callable(self.source) or isinstance(self.source, Collection))
if self.matrix is None and single and self.source == self.destination:
error_msg = f'The transition {self.label!r} moves from {self.source} to itself. Give distinct values.'
raise ValidationError(error_msg)
if self.after is not None and (kind not in {'probability', 'condition'} or not self.after >= 0):
error_msg = (
f'The transition {self.label!r} has the residence time {self.after!r}. Give a number of steps at least '
'0 with a probability or a condition.'
)
raise ValidationError(error_msg)
def _check_transition(self) -> None:
"""Raise an error for arguments the transition cannot take."""
self._check_form()
def _check(self) -> None:
"""Raise an error for a transition without exactly one kind, or with arguments its kind ignores.
Raises:
ValidationError:
The transition has not exactly one kind, the source and the destination are equal, an emission has
no matrix or no observed state, `over` is unknown or goes with a condition, a matrix, or the agent
form, the residence time is negative or goes with a rate or a matrix, the form is unknown, or a
function is not callable.
"""
super()._check()
self._check_transition()
def _get_moving(self, value: Expression, source: SourceValues) -> Expression:
"""Return 1 for a member whose state holds a source value, and 0 otherwise."""
if callable(source):
return where(SR(source(value)), 1, 0)
if isinstance(source, Collection):
return sum((where(value == item, 1, 0) for item in set(source)), SR(0))
return where(value == source, 1, 0)
def _get_jump(self, value: Expression, source: SourceValues, destination: float, rate: Expression) -> JumpLaw:
"""Return the jump of the state from a source value to the destination at a rate."""
if callable(source) or isinstance(source, Collection):
return JumpLaw(rate * self._get_moving(value, source), {value: destination - value})
return JumpLaw(rate * self._get_moving(value, source), {value: destination - source})
def _get_step(self, value: Expression, source: SourceValues, destination: float, moves: Expression) -> UpdateLaw:
"""Return the update of the state that moves a member from a source value where a condition holds."""
if callable(source) or isinstance(source, Collection):
return UpdateLaw(value, where(self._get_moving(value, source) > 0, where(moves, destination, value), value))
return UpdateLaw(value, where(value == source, where(moves, destination, source), value), source=source)
def _holds_source(self, value: float) -> bool:
"""Return whether the value of the state of an agent is a source value."""
if callable(self.source):
return bool(self.source(value))
if isinstance(self.source, Collection):
return value in set(self.source)
return value == self.source
def _combine_rates(self, rate: Expression) -> Expression:
"""Return the rate of the member, or the sum of the rates of its neighbours."""
return neighbors(rate, over=self.neighborhood) if self.over == 'neighbors' else rate
def _combine_probabilities(self, probability: Expression) -> Expression:
"""Return the probability of the member, or the probability that at least one neighbour moves it."""
if self.over == 'member':
return probability
return 1 - exp(neighbors(log(1 - probability), over=self.neighborhood))
def _get_draw(self) -> Expression:
"""Return the uniform draw of the transition in a step."""
return uniform(re.sub(r'\W', '_', f'transition_{self.label}'))
def _get_chain(self, namespace: SimpleNamespace, value: Expression) -> list[BasePart]:
"""Return the transition law of the state by its matrix, with the emission law of the observed state."""
laws: list[BasePart] = []
if self.matrix is not None:
laws.append(TransitionLaw(value, self.matrix(namespace), label=self.label))
if self.emission is not None and self.observed is not None:
observed = getattr(self._get_member(namespace, 'target'), self.observed)
laws.append(EmissionLaw(observed, value, self.emission(namespace), label=self.label))
return laws
def _get_held(self, agent: AgentViewProtocol, model: ModelViewProtocol) -> int:
"""Return the number of whole steps an agent has held its value, counted in the storage of the model."""
value = float(getattr(agent, self.state))
record = model.storage.get(f'{self.label}.{agent.id}')
held = record[1] + 1 if isinstance(record, tuple) and record[0] == value else 0
model.storage[f'{self.label}.{agent.id}'] = (value, held)
return held
def _draw_row(self, row: Sequence[object], model: ModelViewProtocol) -> float:
"""Return a value drawn by the probabilities of a row of a matrix."""
weights = np.asarray(row, dtype=float)
return float(model.rng.choice(weights.size, p=weights / weights.sum()))
def _evaluate[T](
self,
function: Callable[..., T],
agent: AgentViewProtocol,
other: AgentViewProtocol | None,
model: ModelViewProtocol,
) -> T:
"""Return a function of an agent, or of an agent and the member of its cell that acts on it."""
return function(agent, model) if other is None else function(agent, other, model)
def _moves_agent(
self,
agent: AgentViewProtocol,
model: ModelViewProtocol,
other: AgentViewProtocol | None = None,
) -> bool:
"""Return whether an agent that holds the source value moves in this step, moved by another one or alone."""
if self.condition is not None:
return bool(self._evaluate(self.condition, agent, other, model))
if self.rate is not None:
probability = 1.0 - math.exp(-float(self._evaluate(self.rate, agent, other, model)))
elif self.probability is not None:
probability = float(self._evaluate(self.probability, agent, other, model))
else:
probability = 0.0
return bool(model.rng.random() < probability)
def _get_next(self, agent: AgentViewProtocol, model: ModelViewProtocol) -> float | None:
"""Return the next value of the state of an agent, or `None` when it keeps its value."""
value = float(getattr(agent, self.state))
if self.matrix is not None:
return self._draw_row(self.matrix(agent, model)[int(value)], model)
held = self._get_held(agent, model) if self.after is not None else 0
if not self._holds_source(value) or (self.after is not None and held + 1 < self.after):
return None
return float(self.destination) if self._moves_agent(agent, model) else None
def _get_moves(self, agent: AgentViewProtocol, model: ModelViewProtocol) -> list[tuple[AgentViewProtocol, float]]:
"""Return the members an agent moves in this step, itself or the members of its cell, with their values."""
if self.over != 'cell':
value = self._get_next(agent, model)
return [] if value is None else [(agent, value)]
return [
(other, float(self.destination))
for other in agent.agents_here(agent.group)
if other is not agent
and self._holds_source(float(getattr(other, self.state)))
and self._moves_agent(other, model, agent)
]
def _apply(self, moves: list[tuple[AgentViewProtocol, float]], model: ModelViewProtocol) -> None:
"""Set the moved states, and draw the observed state of an emission."""
for agent, value in moves:
setattr(agent, self.state, value)
if self.emission is not None and self.observed is not None:
setattr(agent, self.observed, self._draw_row(self.emission(agent, model)[int(value)], model))
def _move_member(self, agent: AgentViewProtocol, model: ModelViewProtocol) -> None:
"""Move the state of one member, or of the members of its cell, from the states the members before it set."""
if agent.alive:
self._apply(self._get_moves(agent, model), model)
def move(self, model: ModelViewProtocol) -> None:
"""Move the state of every agent of the target that moves in this step, by the schedule.
Args:
model:
The model view of the interaction.
"""
agents = list(model.agents(self._get_role('target')))
if self.schedule == 'synchronous':
moves = [move for agent in agents for move in self._get_moves(agent, model)]
self._apply(moves, model)
return
order = model.rng.permutation(len(agents)) if self.schedule == 'random-sequential' else range(len(agents))
for index in order:
self._move_member(agents[int(index)], model)
def _get_laws(self, namespace: SimpleNamespace) -> list[BasePart]:
"""Return the jump, the update, the transition or emission law of the state, or the rule of the agent form.
Args:
namespace:
The namespace of the symbols the laws can use, with `namespace.target`.
Returns:
laws:
The law of the move, or the rule that moves the agents.
"""
if self.form == 'agent' and self.schedule == 'member':
return [RuleLaw(self._move_member, phase=self.phase, members=self._get_role('target'))]
if self.form == 'agent':
return [RuleLaw(self.move, phase=self.phase)]
value = getattr(self._get_member(namespace, 'target'), self.state)
if self.matrix is not None:
return self._get_chain(namespace, value)
if self.rate is not None:
rate = self._combine_rates(SR(self.rate(namespace)))
return [self._get_jump(value, self.source, self.destination, rate)]
if self.probability is not None:
probability = self._combine_probabilities(SR(self.probability(namespace)))
moves = self._get_draw() < probability
else:
moves = SR(self.condition(namespace)) if self.condition is not None else SR(0)
if self.after is not None:
moves = both(residence(value) + 1 >= self.after, moves)
return [self._get_step(value, self.source, self.destination, moves)]
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)