Double auction
Gode and Sunder's double auction of zero-intelligence traders, a step per trading period.
- Name:
double_auction - Field:
economics/markets - Formalism: agent-based model
- References: Gooding 2019 ch. 8-12; Gode and Sunder 1993
Factory
@register_model(
field='economics/markets',
tags=('market', 'agents', 'double auction', 'zero intelligence'),
references=('Gooding 2019 ch. 8-12', 'Gode and Sunder 1993'),
defaults={'until': 20},
)
def double_auction(
n_buyers: int = 50,
n_sellers: int = 50,
*,
protocol: str = 'order_book',
rounds: int = 20,
constrained: bool = True,
ceiling: float = 200.0,
random_state: int | np.random.Generator | None = 0,
) -> System:
"""Return Gode and Sunder's double auction of zero-intelligence traders, a step per trading period.
Every buyer values one unit and every seller can produce one unit at a cost, both drawn uniformly below the price
ceiling. During a trading period of several rounds, every trader that has not traded yet quotes a random price and
the market matches the quotes, continuously in a limit order book or at one uniform price in a call auction. With
the budget constraint (ZI-C) no trader accepts a loss, and the market alone reaches nearly the whole competitive
surplus; without it (ZI-U), traders make losing trades and the efficiency drops.
Args:
n_buyers:
The number of buyers.
n_sellers:
The number of sellers.
protocol:
The protocol of the market, `'order_book'` or `'call_auction'`.
rounds:
The number of quoting rounds of a trading period.
constrained:
Whether traders never quote at a loss.
ceiling:
The highest price of the market.
random_state:
The seed or generator of the values and the costs.
Returns:
system:
The model, with the groups `buyers` and `sellers`, the market `auction`, and the realised `ledger.surplus`
and `ledger.efficiency` of every trading period.
Raises:
ValueError:
The protocol is unknown.
Examples:
>>> from skcomplex.models.economics.markets import double_auction
>>> [group.size for group in double_auction(10, 8).elements]
[10, 8]
"""
if protocol not in PROTOCOLS:
error_msg = f'Unknown protocol {protocol!r}. Use one of {PROTOCOLS}.'
raise ValueError(error_msg)
rng = np.random.default_rng(random_state)
values = rng.uniform(0.25 * ceiling, ceiling, n_buyers)
costs = rng.uniform(0.0, 0.75 * ceiling, n_sellers)
limit = float(constrained)
buyers = GroupElement(
'buyers',
element=ZeroIntelligenceBuyerElement,
size=n_buyers,
values={'value': values, 'ceiling': ceiling, 'constrained': limit},
)
sellers = GroupElement(
'sellers',
element=ZeroIntelligenceSellerElement,
size=n_sellers,
values={'value': costs, 'ceiling': ceiling, 'constrained': limit},
)
market = MatchingInteraction(
'auction',
requesters='buyers',
providers='sellers',
rule=ContinuousMatchingSpec() if protocol == 'order_book' else BatchMatchingSpec(),
limit=('bid', 'ask'),
quantity=('demand', 'supply'),
phase='trade',
on_match=_book_profit,
)
quotes = [
MapInteraction(label, target=group, state=state, function=function, form='agent', phase='trade')
for label, group, state, function in (
('bids', 'buyers', 'bid', ZeroIntelligenceBuyerElement.get_bid),
('asks', 'sellers', 'ask', ZeroIntelligenceSellerElement.get_ask),
)
]
return System(
elements=[buyers, sellers],
interactions=[*quotes, market, AuctionLedgerInteraction('ledger', phase='close')],
time=DiscreteTime(schedule=PhasesScheduleSpec([PhaseSpec('trade', repeat=rounds), PhaseSpec('close')])),
)
Components
class ZeroIntelligenceBuyerElement(Element):
"""Buyer of one unit per trading period who bids a random price, below its value under the budget constraint.
In every round of a trading period, a buyer that has not bought yet bids a price drawn uniformly from 0 to its
value, or to the price ceiling without the constraint, the zero-intelligence traders of Gode and Sunder (1993).
Args:
label:
The label of the agent or group.
visual:
How the element is drawn, or `None` for `default_visual_`.
value:
The value of the unit to the buyer, its redemption value.
bid:
The current bid.
demand:
The number of units still wanted in the trading period.
profit:
The accumulated surplus of the buyer's trades.
ceiling:
The highest price of the market.
constrained:
One when the buyer never bids above its value, zero otherwise.
Examples:
>>> from skcomplex.core.elements import GroupElement
>>> from skcomplex.models.economics.markets import ZeroIntelligenceBuyerElement
>>> GroupElement('buyers', element=ZeroIntelligenceBuyerElement, size=5, values={'value': 100.0}).size
5
"""
def __init__(
self,
label: str,
visual: VisualSpec | None = None,
*,
value: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 100.0,
bid: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
demand: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
profit: Annotated[float, StateSpec()] = 0.0,
ceiling: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 200.0,
constrained: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
) -> None:
super().__init__(label, visual=visual)
self.value = value
self.bid = bid
self.demand = demand
self.profit = profit
self.ceiling = ceiling
self.constrained = constrained
@classmethod
def _get_view(cls) -> type[ZeroIntelligenceBuyerView]:
"""Return the view of a buyer, with its value, bid, demand, and profit."""
return ZeroIntelligenceBuyerView
@staticmethod
def get_bid(agent: ZeroIntelligenceBuyerView, model: ModelView) -> float:
"""Return a bid drawn uniformly from 0 to the value, or to the price ceiling without the constraint.
Args:
agent:
The buyer.
model:
The model view of the quotes.
Returns:
bid:
The bid of the round.
"""
return float(model.rng.uniform(0.0, agent.value if agent.constrained else agent.ceiling))
class ZeroIntelligenceSellerElement(Element):
"""Seller of one unit per trading period who asks a random price, above its cost under the budget constraint.
In every round of a trading period, a seller that has not sold yet asks a price drawn uniformly from its cost, or
from 0 without the constraint, to the price ceiling.
Args:
label:
The label of the agent or group.
visual:
How the element is drawn, or `None` for `default_visual_`.
value:
The cost of the unit to the seller.
ask:
The current ask.
supply:
The number of units still offered in the trading period.
profit:
The accumulated surplus of the seller's trades.
ceiling:
The highest price of the market.
constrained:
One when the seller never asks below its cost, zero otherwise.
Examples:
>>> from skcomplex.core.elements import GroupElement
>>> from skcomplex.models.economics.markets import ZeroIntelligenceSellerElement
>>> GroupElement('sellers', element=ZeroIntelligenceSellerElement, size=5, values={'value': 50.0}).size
5
"""
def __init__(
self,
label: str,
visual: VisualSpec | None = None,
*,
value: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 100.0,
ask: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 200.0,
supply: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
profit: Annotated[float, StateSpec()] = 0.0,
ceiling: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 200.0,
constrained: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
) -> None:
super().__init__(label, visual=visual)
self.value = value
self.ask = ask
self.supply = supply
self.profit = profit
self.ceiling = ceiling
self.constrained = constrained
@classmethod
def _get_view(cls) -> type[ZeroIntelligenceSellerView]:
"""Return the view of a seller, with its cost, ask, supply, and profit."""
return ZeroIntelligenceSellerView
@staticmethod
def get_ask(agent: ZeroIntelligenceSellerView, model: ModelView) -> float:
"""Return an ask drawn uniformly from the cost, or from 0 without the constraint, to the price ceiling.
Args:
agent:
The seller.
model:
The model view of the quotes.
Returns:
ask:
The ask of the round.
"""
bottom = agent.value if agent.constrained else 0.0
return float(model.rng.uniform(bottom, max(bottom, agent.ceiling)))
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 AuctionLedgerInteraction(FeedbackInteraction[ModelView]):
"""Ledger that closes every trading period: it measures its surplus and gives every trader a new unit.
The realised surplus is the value of the units bought minus the cost of the units sold, and the allocative
efficiency is its ratio to the maximum surplus of the supply and demand schedules.
Args:
label:
The label of the ledger.
phase:
The phase the ledger acts in.
Examples:
>>> from skcomplex.models.economics.markets import AuctionLedgerInteraction
>>> AuctionLedgerInteraction('ledger', phase='close').reports
('surplus', 'efficiency')
"""
reports: ClassVar[tuple[str, ...]] = ('surplus', 'efficiency')
def act(self, model: ModelView) -> None:
"""Report the surplus and the efficiency of the trading period and open the next one.
Args:
model:
The model view scoped to the ledger.
"""
buyers = model.agents('buyers', ZeroIntelligenceBuyerView)
sellers = model.agents('sellers', ZeroIntelligenceSellerView)
values = np.array([buyer.value for buyer in buyers])
costs = np.array([seller.value for seller in sellers])
bought = np.array([buyer.demand <= 0 for buyer in buyers])
sold = np.array([seller.supply <= 0 for seller in sellers])
surplus = float(values[bought].sum() - costs[sold].sum())
model.report('surplus', surplus)
model.report('efficiency', surplus / maximum_surplus(values, costs))
for buyer in buyers:
buyer.demand = 1.0
for seller in sellers:
seller.supply = 1.0