Shops and digital markets
Shops on a grid that compete for the shoppers on every cell, in person or in a digital market.
- Name:
shops - Field:
economics/markets - Formalism: agent-based model
- References: Hamill and Gilbert 2016 sec. 6.3-6.4
Factory
@register_model(
field='economics/markets',
tags=('market', 'agents', 'retail', 'digital market', 'grid'),
references=('Hamill and Gilbert 2016 sec. 6.3-6.4',),
defaults={'until': 100},
observables=_shops_observables,
)
def shops(
n_shops: int = 16,
*,
width: int = 20,
height: int = 20,
online: float = 0.0,
distance_cost: float = 1.0,
capacity: float = 1000.0,
fixed_cost: float = 20.0,
value: float = 20.0,
random_state: int | np.random.Generator | None = 0,
) -> System:
"""Return shops on a grid that compete for the shoppers on every cell, in person or in a digital market.
Every step every shopper wants one unit and shops online with probability `online`, and in person otherwise, when it
pays a travel cost for the distance to the shop. The shoppers choose in random order the shops they prefer by price
plus travel cost, within the shops' capacities, a greedy matching that pays the shop's ask for a unit. Shops search
for the price of the largest profit by trial and error, pay a fixed cost every step, and close when their money runs
out. Distance gives every shop a local market, so many shops survive with high prices. In the digital market the
cheapest shop takes every shopper, the others undercut it, prices fall toward the unit cost, and most shops close
(Hamill and Gilbert 2016 sec. 6.3-6.4).
Args:
n_shops:
The initial number of shops.
width:
The number of columns of the grid, with one shopper per cell.
height:
The number of rows of the grid.
online:
The probability that a shopper shops online in a step.
distance_cost:
The travel cost per cell of shopping in person.
capacity:
The units every shop can sell in a step.
fixed_cost:
The cost of running a shop for a step.
value:
The highest price a shopper pays.
random_state:
The seed or generator of the cells and initial prices of the shops.
Returns:
system:
The model, with the groups `shoppers` and `shops` and the market `shopping`, whose reporters are the
`shopping.value`, `shopping.volume`, and `shopping.unmatched` of the purchases.
Examples:
>>> from skcomplex.models.economics.markets import shops
>>> [group.size for group in shops(4, width=5, height=5).elements]
[25, 4]
"""
rng = np.random.default_rng(random_state)
cells = [(x, y) for x in range(width) for y in range(height)]
places = rng.choice(len(cells), size=min(n_shops, len(cells)), replace=False)
shoppers = GroupElement(
'shoppers',
element=ResidentElement,
size=len(cells),
values={'bid': value, 'online': online, 'distance_cost': distance_cost},
location=cells,
)
stores = GroupElement(
'shops',
element=StoreElement,
size=len(places),
values={
'price': rng.uniform(0.25 * value, 0.75 * value, len(places)),
'capacity': capacity,
'fixed_cost': fixed_cost,
},
variable=True,
location=[cells[index] for index in places],
)
plan = [
MapInteraction(
'demand',
target='shoppers',
state='demand',
function=lambda agent, model: 1.0,
form='agent',
phase='market',
),
MapInteraction(
'travel',
target='shoppers',
state='travel',
function=ResidentElement.get_travel,
form='agent',
phase='market',
),
*(
MapInteraction(
f'opening_{name}',
target='shops',
state=name,
function=function,
form='agent',
phase='market',
)
for name, function in StoreElement.get_opening()
),
]
settlement = [
*(
MapInteraction(
f'settlement_{name}',
target='shops',
state=name,
function=function,
form='agent',
phase='accounts',
)
for name, function in StoreElement.get_settlement()
),
SignAdjustmentInteraction(
'settlement_price',
target='shops',
state='price',
signal=lambda agent, model: agent.direction,
step=lambda agent, model: agent.adjustment,
lower=lambda agent, model: agent.unit_cost,
form='agent',
phase='accounts',
),
LifeCycleInteraction('closure', target='shops', death=lambda agent, model: agent.money < 0, phase='accounts'),
]
shopping = MatchingInteraction[ModelView](
'shopping',
requesters='shoppers',
providers='shops',
rule=PartnerSpec(choice='score', score=ResidentElement.get_preference, clearing='provider'),
limit=('bid', 'ask'),
quantity=('demand', 'supply'),
transfers=(('money', -1),),
phase='market',
on_match=StoreElement.record_sale,
)
return System(
elements=[shoppers, stores],
interactions=[*plan, shopping, *settlement],
space=LatticeSpace((width, height), boundary='fixed', neighborhood='moore'),
time=DiscreteTime(schedule=PhasesScheduleSpec([PhaseSpec('market'), PhaseSpec('accounts')])),
)
Components
class ResidentElement(Element):
"""Shopper who wants one unit a step and shops in person, paying for travel, or online, where distance is free.
Args:
label:
The label of the agent or group.
visual:
How the element is drawn, or `None` for `default_visual_`.
bid:
The highest price the shopper pays.
demand:
The units still wanted in the step.
money:
The money of the shopper.
travel:
The travel cost per cell of the current step, zero online.
distance_cost:
The travel cost per cell of shopping in person.
online:
The probability of shopping online in a step.
Examples:
>>> from skcomplex.core.elements import GroupElement
>>> from skcomplex.models.economics.markets import ResidentElement
>>> GroupElement('shoppers', element=ResidentElement, size=3, values={'online': 1.0}).size
3
"""
def __init__(
self,
label: str,
visual: VisualSpec | None = VisualSpec('agents', {'color_by': 'travel', 'color_range': (0.0, 1.0)}),
*,
bid: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 20.0,
demand: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
money: Annotated[float, StateSpec()] = 1000.0,
travel: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
distance_cost: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
online: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
) -> None:
super().__init__(label, visual=visual)
self.bid = bid
self.demand = demand
self.money = money
self.travel = travel
self.distance_cost = distance_cost
self.online = online
@classmethod
def _get_view(cls) -> type[ResidentView]:
"""Return the view of a shopper, with its states and shopping parameters."""
return ResidentView
@staticmethod
def get_travel(agent: ResidentView, model: ModelView) -> float:
"""Return the travel cost per cell of the step, zero when the shopper shops online with its probability.
Args:
agent:
The shopper.
model:
The model view of the plan.
Returns:
travel:
The travel cost per cell.
"""
return 0.0 if model.rng.random() < agent.online else agent.distance_cost
@staticmethod
def get_preference(shopper: object, shop: object) -> float:
"""Return the preference of a shopper for a shop, the opposite of the delivered price.
The delivered price is the shop's ask plus the shopper's travel cost per cell times the distance to the shop.
Args:
shopper:
The shopper.
shop:
The shop.
Returns:
preference:
The opposite of the delivered price, higher for a preferred shop.
"""
if not isinstance(shopper, ResidentView) or not isinstance(shop, StoreView):
return -math.inf
return -(shop.ask + shopper.travel * shopper.distance_to(shop))
class StoreElement(Element):
"""Shop that offers its capacity at its price, searches for the price of the largest profit, and may close.
The shop searches for its price by trial and error: it moves the price by `adjustment` in the same direction as
long as its profit grows, and turns back when the profit falls, never below the unit cost. It pays the unit cost
of every unit sold and the fixed cost every step, and closes when its money runs out.
Args:
label:
The label of the agent or group.
visual:
How the element is drawn, or `None` for `default_visual_`.
price:
The price of a unit.
ask:
The price offered in the current step.
supply:
The units still offered in the step.
sales:
The units sold in the last step.
money:
The money of the shop.
profit:
The profit of the last step.
direction:
The direction of the last change of the price, 1 up or -1 down.
capacity:
The units the shop can sell in a step.
unit_cost:
The cost of a unit sold.
fixed_cost:
The cost of running the shop for a step.
adjustment:
The relative change of the price in a step.
Examples:
>>> from skcomplex.core.elements import GroupElement
>>> from skcomplex.models.economics.markets import StoreElement
>>> GroupElement('shops', element=StoreElement, size=4, values={'price': 10.0}).size
4
"""
def __init__(
self,
label: str,
visual: VisualSpec | None = VisualSpec('agents', {'color_by': 'price'}),
*,
price: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 10.0,
ask: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 10.0,
supply: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
sales: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
money: Annotated[float, StateSpec()] = 100.0,
profit: Annotated[float, StateSpec()] = 0.0,
direction: Annotated[float, StateSpec()] = 1.0,
capacity: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 1000.0,
unit_cost: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
fixed_cost: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 20.0,
adjustment: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.05,
) -> None:
super().__init__(label, visual=visual)
self.price = price
self.ask = ask
self.supply = supply
self.sales = sales
self.money = money
self.profit = profit
self.direction = direction
self.capacity = capacity
self.unit_cost = unit_cost
self.fixed_cost = fixed_cost
self.adjustment = adjustment
@classmethod
def _get_view(cls) -> type[StoreView]:
"""Return the view of a shop, with its states and cost parameters."""
return StoreView
@staticmethod
def get_profit(agent: StoreView) -> float:
"""Return the profit of the step, the margin on the units sold less the fixed cost.
Args:
agent:
The shop.
Returns:
profit:
The profit of the step.
"""
return agent.sales * (agent.ask - agent.unit_cost) - agent.fixed_cost
@staticmethod
def get_opening() -> tuple[tuple[str, Callable[[StoreView, ModelView], float]], ...]:
"""Return the states a shop sets when it opens for the market, its capacity offered at its price.
Returns:
opening:
The name of every state and the function of its value, of the shop and the model view.
"""
return (
('supply', lambda agent, model: agent.capacity),
('ask', lambda agent, model: agent.price),
('sales', lambda agent, model: 0.0),
)
@staticmethod
def get_settlement() -> tuple[tuple[str, Callable[[StoreView, ModelView], float]], ...]:
"""Return the states a shop sets when it settles its accounts, in their order.
The shop turns its price back when its profit falls, records the profit, and pays its costs. The pricing of
the model then steps the price by the sign of the direction.
Returns:
settlement:
The name of every state and the function of its value, of the shop and the model view.
"""
profit = StoreElement.get_profit
return (
('direction', lambda agent, model: -agent.direction if profit(agent) < agent.profit else agent.direction),
('profit', lambda agent, model: profit(agent)),
('money', lambda agent, model: agent.money - agent.sales * agent.unit_cost - agent.fixed_cost),
)
@staticmethod
def record_sale(match: MatchSpec, model: ModelView) -> None:
"""Count the units of a purchase in the sales of the shop.
Args:
match:
The purchase of a shopper, the requester, from a shop, the provider.
model:
The model view of the market.
"""
shop = match.provider
if isinstance(shop, StoreView):
shop.sales += match.quantity
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)
class SignAdjustmentInteraction(Interaction):
"""Step of a state of discrete time by the sign of a signal, kept within bounds.
In the mode `'multiplicative'` the next value of the state `x` is `x (1 + step sign(signal))`, and in the mode
`'additive'` it is `x + step sign(signal)`, so the state rises while the signal is positive, falls while it is
negative, and stays while it is zero, by trial and error rather than by the size of the signal. The next value is
then limited to `[lower, upper]`.
In the `'symbolic'` form, the signal, the step, and the bounds are numbers or functions of the namespace, where
`namespace.target` holds the states and parameters of the target. In the `'agent'` form, the target is an agent
group and they are numbers or functions `f(agent, model)` that return numbers. The agents step at once from the
states at the start of the rule under the schedule `'synchronous'`, one after the other in their order or in a new
random order every step under `'sequential'` or `'random-sequential'`, each reading the states the agents before
it set, or each in its own activation under `'member'`, together with the rules of its element and the other
member interactions, in the order the system lists them.
Args:
label:
The label of the interaction, unique among the parts of its system.
target:
The path of the element or group whose state steps.
state:
The name of the state of the target.
signal:
The function of the signal whose sign sets the direction of the step.
step:
The size of the step, relative in the mode `'multiplicative'`, a number or a function.
mode:
`'multiplicative'` or `'additive'`.
lower:
The lowest next value, a number or a function, or `None` for none.
upper:
The highest next value, a number or a function, or `None` for none.
form:
`'symbolic'` for an update law of the state 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.
schedule:
The order of the agents in the agent form, `'synchronous'`, `'sequential'`, `'random-sequential'`, or
`'member'`.
Raises:
ValidationError:
The mode, the form, or the schedule is unknown, the symbolic form is not synchronous, or the signal is not
callable, raised by `check`.
Examples:
>>> from typing import Annotated
>>> from skcomplex.core.elements import Element
>>> from skcomplex.core.interactions import SignAdjustmentInteraction
>>> from skcomplex.core.spaces import DiscreteTime
>>> from skcomplex.core.system import System
>>> from skcomplex.engine.runtime import Simulation
>>> from skcomplex.specs import StateSpec
>>> class QuoteElement(Element):
... def __init__(self, label, x: Annotated[float, StateSpec()] = 1.0):
... super().__init__(label)
... self.x = x
>>> search = SignAdjustmentInteraction(
... 'search', target='quote', state='x', signal=lambda s: 2 - s.target.x, step=0.5, upper=2.0
... )
>>> system = System(elements=[QuoteElement('quote')], interactions=[search], time=DiscreteTime())
>>> Simulation(system).check().prepare().run(until=3).trajectory_.values('quote.x').tolist()
[1.0, 1.5, 2.0, 2.0]
"""
def __init__(
self,
label: str,
*,
target: Annotated[str, RoleSpec()],
state: str,
signal: Callable[..., object],
step: Value = 0.1,
mode: str = 'multiplicative',
lower: Value | None = None,
upper: Value | None = None,
form: str = 'symbolic',
phase: str | None = None,
schedule: str = 'synchronous',
) -> None:
super().__init__(label)
self.target = target
self.state = state
self.signal = signal
self.step = step
self.mode = mode
self.lower = lower
self.upper = upper
self.form = form
self.phase = phase
self.schedule = schedule
def _check(self) -> None:
"""Raise an error for an unknown mode, form, or schedule, or a signal that cannot be called.
Raises:
ValidationError:
The mode, the form, or the schedule is unknown, the symbolic form is not synchronous, or the signal is
not callable.
"""
super()._check()
for name, value, variants in (('mode', self.mode, MODES), ('form', self.form, FORMS)):
if value not in variants:
error_msg = f'The {name} of {self.label!r} is {value!r}. Use one of {", ".join(variants)}.'
raise ValidationError(error_msg)
self._check_variant('schedule', self.schedule, SCHEDULES)
if self.form == 'symbolic' and self.schedule != 'synchronous':
error_msg = f'The schedule {self.schedule!r} of {self.label!r} runs agent rules. Use `form="agent"`.'
raise ValidationError(error_msg)
if not callable(self.signal):
error_msg = f'The signal of {self.label!r} must be callable, but it is {self.signal!r}.'
raise ValidationError(error_msg)
def _get_stepped(self, value: Expression, step: Expression, direction: Expression) -> Expression:
"""Return the value stepped in a direction by the mode."""
if self.mode == 'additive':
return value + step * direction
return value * (1 + step * direction)
def _read(self, item: Value | None, *arguments: object) -> object:
"""Return a number, the value of a function at the arguments, or `None`."""
return item(*arguments) if callable(item) else item
def _get_number(self, item: Value | None, agent: AgentViewProtocol, model: ModelViewProtocol) -> float | None:
"""Return a number or the value of a function of an agent as a number, or `None`."""
found = self._read(item, agent, model)
if found is None:
return None
if not isinstance(found, SupportsFloat):
error_msg = f'A function of {self.label!r} returns {found!r} for an agent. Return a number.'
raise ValidationError(error_msg)
return float(found)
def _get_value(self, agent: AgentViewProtocol, model: ModelViewProtocol) -> float:
"""Return the next value of the state of an agent."""
direction = float(np.sign(self._get_number(self.signal, agent, model) or 0.0))
step = self._get_number(self.step, agent, model) or 0.0
current = float(getattr(agent, self.state))
value = current + step * direction if self.mode == 'additive' else current * (1 + step * direction)
lower, upper = self._get_number(self.lower, agent, model), self._get_number(self.upper, agent, model)
if upper is not None:
value = min(upper, value)
if lower is not None:
value = max(lower, value)
return value
def _set_value(self, agent: AgentViewProtocol, value: float) -> None:
"""Set the state of an agent to its next value."""
setattr(agent, self.state, value)
def _adjust_member(self, agent: AgentViewProtocol, model: ModelViewProtocol) -> None:
"""Step the state of one agent, in its own activation."""
if agent.alive:
self._set_value(agent, self._get_value(agent, model))
def adjust(self, model: ModelViewProtocol) -> None:
"""Step the state of every agent of the target, 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_laws(self, namespace: SimpleNamespace) -> list[BasePart]:
"""Return the update of the state to its stepped value, or the rule of the agent form.
Args:
namespace:
The namespace of the symbols the laws can use, with `namespace.target`.
Returns:
laws:
The update law of the state, or the rule that steps the agents.
"""
if self.form == 'agent' and self.schedule == 'member':
return [RuleLaw(self._adjust_member, phase=self.phase, members=self._get_role('target'))]
if self.form == 'agent':
return [RuleLaw(self.adjust, phase=self.phase)]
value = SR(getattr(self._get_member(namespace, 'target'), self.state))
direction = sign(self._read(self.signal, namespace))
stepped = self._get_stepped(value, SR(self._read(self.step, namespace)), direction)
bounded = clip(stepped, self._read(self.lower, namespace), self._read(self.upper, namespace))
return [UpdateLaw(value, bounded)]
class LifeCycleInteraction(Interaction):
"""Life cycle of the agents of a group, which give birth to offspring and die, every step of discrete time.
Every step the agents act once each, in their order under the schedule `'sequential'`, in a new random order under
`'random-sequential'`, or each in its own activation under `'member'`, together with the rules of its element and
the other member interactions, in the order the system lists them, without the reports of the step and without
arrivals. An agent dies with the probability `death`, and an agent that survives gives birth with the probability
`birth`, or as many times as a Poisson number of the mean `rate`. A probability is a number or a function of the
agent and the model view, and a function that returns a condition gives probability one where it holds. Every birth
creates `offspring` new agents, a number or a function of the parent and the model view, such as a Poisson draw, of
the group `group`, or of the parent's group without one. A new agent stands in the cell of its parent, with the
placement `'here'`, in a neighbouring cell that no agent of the group holds, with the placement `'neighbor'`, or at
the location a function of the parent and the model view returns, where `None` means no place, and a birth without a
place creates fewer agents.
A new agent copies the states `inherit` of its parent and takes the values that `values(parent, model)` returns,
such as a generation one higher or drawn states. The parent splits each state of `split` with its offspring,
keeping the share `keep`, a number or a function of the parent and the model view such as a random cut, or else an
equal share, and the offspring share the rest equally. Then `mutate(values, model)` returns the values of a new
agent from those it would get, and the other states take their defaults. A parent that gave birth takes the values
`reset`, and with `replace` it is removed, so its offspring replace it, also when the birth created none. With
`renew` an agent that dies is renewed in place instead of removed: it keeps its identity, its place, and the states
the new values leave out, and takes the values of an offspring of its own. With the probability `cull` an agent
that does not die removes, after its births, one member of `group`, or of the target without a group, for which
`kin(agent, member)` holds, the first in the order of the group under the pick `'first'`, such as one it owns, or
a random one under `'random'`. Every
step a Poisson number of the mean `arrival`, a number or a function of the model view, of new agents arrive without
a parent, at the location of the placement function or else at a random place, with the values
`values(None, model)`. The life cycle reports the `births`, `deaths`, and `arrivals` of every step. It is an agent
rule, so the system runs in discrete time.
Args:
label:
The label of the interaction, unique among the parts of its system.
target:
The path of the variable group whose agents give birth and die.
birth:
The probability that an agent gives birth in a step, or `None` for no births by probability.
death:
The probability that an agent dies in a step, or `None` for no deaths.
rate:
The mean number of births of an agent in a step, a number or a function, or `None` for no births by rate.
offspring:
The number of new agents of a birth.
placement:
Where a new agent stands, `'here'`, `'neighbor'`, or a function of the parent, `None` for an arrival, and
the model view that returns its location.
inherit:
The names of the states a new agent copies from its parent.
split:
The names of the states the parent shares with its offspring.
mutate:
A function of the values of a new agent and the model view that returns its values, or `None`.
schedule:
The order of the agents, `'sequential'`, `'random-sequential'`, or `'member'`.
phase:
The phase of the step the life cycle runs in, or `None` for the default phase.
group:
The path of the variable group the new agents join, or `None` for the group of the target.
values:
A function of the parent, `None` for an arrival, and the model view that returns values of a new agent by
state name, or `None`.
keep:
The share of every split state the parent keeps, a number or a function, or `None` for an equal share.
reset:
The values a parent takes after it gives birth, by state name, or `None`.
replace:
Whether a parent that gives birth is removed.
renew:
Whether an agent that dies is renewed in place instead of removed.
cull:
The probability that an agent removes one member of `group`, or of the target without a group, in a
step, or `None` for no removals.
kin:
A function of an agent and a member of `group` that returns whether the agent may remove the member, or
`None` for every member.
arrival:
The mean number of new agents that arrive in a step without a parent, or `None` for no arrivals.
pick:
Which member of `group` a removal takes among those it may remove, `'first'` or `'random'`.
Raises:
ValidationError:
The placement, the schedule, or the pick is unknown, both a birth probability and a rate are given, a
function is not callable, or the schedule `'member'` has arrivals, raised by `check`.
Examples:
>>> from typing import Annotated
>>> from skcomplex.core.elements import Element, GroupElement
>>> from skcomplex.core.interactions import LifeCycleInteraction
>>> from skcomplex.core.spaces import LatticeSpace
>>> from skcomplex.core.system import System
>>> from skcomplex.engine.runtime import Simulation
>>> from skcomplex.specs import StateSpec
>>> class CellElement(Element):
... def __init__(self, label, visual=None, *, size: Annotated[float, StateSpec()] = 0.0):
... super().__init__(label, visual=visual)
... self.size = size
>>> cells = GroupElement('cells', element=CellElement, size=1, values={'size': 8.0}, variable=True)
>>> division = LifeCycleInteraction('division', target='cells', birth=1.0, split=('size',))
>>> system = System(elements=[cells], interactions=[division], space=LatticeSpace((3, 3)))
>>> trajectory = Simulation(system, random_state=0).check().prepare().run(until=2).trajectory_
>>> trajectory.agents('cells', 2).values['size'].tolist()
[2.0, 2.0, 2.0, 2.0]
"""
reports: ClassVar[tuple[str, ...]] = ('births', 'deaths', 'arrivals')
def __init__(
self,
label: str,
*,
target: Annotated[str, RoleSpec()],
birth: Chance | None = None,
death: Chance | None = None,
rate: Chance | None = None,
offspring: Count = 1,
placement: Placement = 'here',
inherit: Sequence[str] = (),
split: Sequence[str] = (),
mutate: Mutation | None = None,
schedule: str = 'random-sequential',
phase: str | None = None,
group: Annotated[str | None, RoleSpec(optional=True)] = None,
values: Values | None = None,
keep: Chance | None = None,
reset: Mapping[str, object] | None = None,
replace: bool = False,
renew: bool = False,
cull: Chance | None = None,
kin: Kinship | None = None,
arrival: Arrival | None = None,
pick: str = 'first',
) -> None:
super().__init__(label)
self.target = target
self.birth = birth
self.death = death
self.rate = rate
self.offspring = offspring
self.placement = placement
self.inherit = inherit
self.split = split
self.mutate = mutate
self.schedule = schedule
self.phase = phase
self.group = group
self.values = values
self.keep = keep
self.reset = reset
self.replace = replace
self.renew = renew
self.cull = cull
self.kin = kin
self.arrival = arrival
self.pick = pick
def _check_variants(self) -> None:
"""Raise an error for an unknown placement, schedule, or pick, or arrivals of a member schedule."""
self._check_variant('placement', self.placement, PLACEMENTS, functions=True)
self._check_variant('schedule', self.schedule, SCHEDULES)
self._check_variant('pick', self.pick, PICKS)
if self.schedule == 'member' and self.arrival is not None:
error_msg = f'{self.label!r} runs member by member, so no agent brings arrivals. Give `arrival=None`.'
raise ValidationError(error_msg)
def _check_space(self, space: BaseSpace | None) -> None:
"""Raise an error for a placement in a neighbouring cell or node in a continuous space."""
if self.placement == 'neighbor':
self._check_grid('placement', self.placement, space, "placement='here', or a placement function")
def _check(self) -> None:
"""Raise an error for an unknown placement, a birth by both probability and rate, or a bad function.
Raises:
ValidationError:
The placement, the schedule, or the pick is unknown, both a birth probability and a rate are given,
a value is neither a number nor a function, or the schedule `'member'` has arrivals.
"""
super()._check()
self._check_variants()
if self.birth is not None and self.rate is not None:
error_msg = f'{self.label!r} gives births by a probability and by a rate. Give one of them.'
raise ValidationError(error_msg)
for name in ('birth', 'death', 'rate', 'offspring', 'keep', 'cull', 'arrival'):
value = getattr(self, name)
if value is not None and not callable(value) and not isinstance(value, int | float):
error_msg = f'The {name} of {self.label!r} is {value!r}. Give a number or a function.'
raise ValidationError(error_msg)
for name in ('mutate', 'values', 'kin'):
value = getattr(self, name)
if value is not None and not callable(value):
error_msg = f'The {name} of {self.label!r} is {value!r}. Give a function, or None.'
raise ValidationError(error_msg)
@staticmethod
def _evaluate(value: Chance | Count, agent: AgentViewProtocol, model: ModelViewProtocol) -> float:
"""Return a number, or the value of a function of an agent and the model view, as a float."""
return float(value(agent, model) if callable(value) else value)
def _happens(self, chance: Chance | None, agent: AgentViewProtocol, model: ModelViewProtocol) -> bool:
"""Return whether an event of a probability happens to an agent in this step."""
if chance is None:
return False
probability = self._evaluate(chance, agent, model)
return probability >= 1.0 or (probability > 0.0 and model.rng.random() < probability)
def _count_births(self, agent: AgentViewProtocol, model: ModelViewProtocol) -> int:
"""Return the number of births of a parent in this step."""
if self.rate is not None:
return int(model.rng.poisson(self._evaluate(self.rate, agent, model)))
return int(self._happens(self.birth, agent, model))
def _get_places(
self,
agent: AgentViewProtocol,
count: int,
occupied: set[object],
model: ModelViewProtocol,
) -> list[object]:
"""Return the locations of the new agents of a parent, fewer when places run out."""
if callable(self.placement):
places = [self.placement(agent, model) for _ in range(count)]
return [place for place in places if place is not None]
if self.placement == 'here':
return [agent.location] * count
free = [cell for cell in agent.neighbor_cells() if cell not in occupied]
chosen = [free[int(index)] for index in model.rng.permutation(len(free))[:count]]
occupied.update(chosen)
return chosen
def _get_values(self, agent: AgentViewProtocol | None, model: ModelViewProtocol) -> dict[str, object]:
"""Return the inherited and given values of a new agent of a parent, or of an arrival."""
values: dict[str, object] = {} if agent is None else {name: getattr(agent, name) for name in self.inherit}
if self.values is not None:
values.update(self.values(agent, model))
return values
def _create(
self,
agent: AgentViewProtocol | None,
place: object,
values: dict[str, object],
model: ModelViewProtocol,
) -> None:
"""Create a new agent with mutated values at a place, `None` for a random one, in the group of the cycle."""
if self.mutate is not None:
values = self.mutate(values, model)
if self.group is not None:
model.create(self._get_role('group'), place, **values)
elif agent is not None:
agent.create(None, place, **values)
else:
model.create(self._get_role('target'), place, **values)
def _bear(self, agent: AgentViewProtocol, places: list[object], model: ModelViewProtocol) -> None:
"""Create the new agents of a parent at their places, with inherited, given, split, and mutated values."""
kept = 1 / (len(places) + 1) if self.keep is None else self._evaluate(self.keep, agent, model)
totals = {name: float(getattr(agent, name)) for name in self.split}
for name, amount in totals.items():
setattr(agent, name, amount * kept)
shares = {name: (amount - amount * kept) / max(len(places), 1) for name, amount in totals.items()}
for place in places:
values = self._get_values(agent, model)
values.update(shares)
self._create(agent, place, values, model)
def _renew(self, agent: AgentViewProtocol, model: ModelViewProtocol) -> None:
"""Give an agent that dies the values of an offspring of its own, in place."""
values = self._get_values(agent, model)
if self.mutate is not None:
values = self.mutate(values, model)
for name, value in values.items():
setattr(agent, name, value)
def _remove_kin(self, agent: AgentViewProtocol, model: ModelViewProtocol) -> None:
"""Remove the first or a random member of the group of the cycle that is kin of an agent, if there is one."""
kin = self.kin
group = self._get_role('target' if self.group is None else 'group')
members = (item for item in model.agents(group) if item.alive and (kin is None or kin(agent, item)))
if self.pick == 'random':
candidates = list(members)
if candidates:
candidates[int(model.rng.integers(len(candidates)))].remove()
return
member = next(members, None)
if member is not None:
member.remove()
def _die(self, agent: AgentViewProtocol, occupied: set[object], model: ModelViewProtocol) -> None:
"""Renew an agent in place, or remove it and free its cell."""
if self.renew:
self._renew(agent, model)
return
occupied.discard(agent.location_key)
agent.remove()
def _act(self, agent: AgentViewProtocol, occupied: set[object], model: ModelViewProtocol) -> tuple[int, int]:
"""Let an agent die, or give birth and remove kin, and return its numbers of new agents and deaths."""
if self._happens(self.death, agent, model):
self._die(agent, occupied, model)
return 0, 1
born, died = self._give_birth(agent, occupied, model)
if self._happens(self.cull, agent, model):
self._remove_kin(agent, model)
return born, died
def _give_birth(self, agent: AgentViewProtocol, occupied: set[object], model: ModelViewProtocol) -> tuple[int, int]:
"""Let an agent give birth, and return its number of new agents and whether it was replaced."""
born = 0
events = self._count_births(agent, model)
if events > 0:
births = events * int(self._evaluate(self.offspring, agent, model))
places = self._get_places(agent, births, occupied, model) if births else []
if places or self.split:
self._bear(agent, places, model)
born = len(places)
for name, value in (self.reset or {}).items():
setattr(agent, name, value)
if self.replace:
occupied.discard(agent.location_key)
agent.remove()
return born, 1
return born, 0
def _arrive(self, model: ModelViewProtocol) -> int:
"""Create the agents that arrive without a parent in this step, and return their number."""
if self.arrival is None:
return 0
mean = self.arrival(model) if callable(self.arrival) else self.arrival
count = int(model.rng.poisson(float(mean)))
for _ in range(count):
place = self.placement(None, model) if callable(self.placement) else None
self._create(None, place, self._get_values(None, model), model)
return count
def _live_member(self, agent: AgentViewProtocol, model: ModelViewProtocol) -> None:
"""Run the life cycle of one agent in its own activation."""
if not agent.alive:
return
neighbor = self.placement == 'neighbor'
occupied = {other.location_key for other in model.agents(self._get_role('target'))} if neighbor else set()
self._act(agent, occupied, model)
def _live(self, model: ModelViewProtocol) -> None:
"""Remove or renew the agents that die and create the offspring of those that give birth, by the schedule."""
agents = model.agents(self._get_role('target'))
occupied = {agent.location_key for agent in agents} if self.placement == 'neighbor' else set()
births = deaths = 0
order = model.rng.permutation(len(agents)) if self.schedule == 'random-sequential' else range(len(agents))
for index in order:
agent = agents[int(index)]
if not agent.alive:
continue
born, died = self._act(agent, occupied, model)
births, deaths = births + born, deaths + died
model.report('births', births)
model.report('deaths', deaths)
model.report('arrivals', self._arrive(model))
def _get_laws(self, namespace: SimpleNamespace) -> list[BaseLaw]:
"""Return the rule of the life cycle.
Args:
namespace:
The namespace of the symbols the laws can use.
Returns:
laws:
The rule that removes and creates agents every step.
"""
if self.schedule == 'member':
return [RuleLaw(self._live_member, phase=self.phase, members=self._get_role('target'))]
return [RuleLaw(self._live, phase=self.phase)]