Order book with imitation
A stock market of fundamentalists and chartists on a grid who trade in a limit order book and imitate.
- Name:
order_book - Field:
economics/finance - Formalism: agent-based model
- References: Caiani et al. 2016 ch. 3; Lux and Marchesi 1999
Factory
@register_model(
field='economics/finance',
tags=('stock market', 'agents', 'order book', 'imitation', 'fundamentalists', 'chartists'),
references=('Caiani et al. 2016 ch. 3', 'Lux and Marchesi 1999'),
defaults={'until': 200, 'invariants': {f'total_{name}': 1e-6 for name in HOLDINGS}},
observables=_order_book_observables,
)
def order_book(
width: int = 20,
height: int = 20,
*,
chartists: float = 0.3,
fundamental: float = 100.0,
reversion: float = 0.2,
extrapolation: float = 1.0,
noise: float = 0.02,
imitation: float = 0.1,
random_state: int | np.random.Generator | None = 0,
) -> System:
"""Return a stock market of fundamentalists and chartists on a grid who trade in a limit order book and imitate.
Every step every trader forecasts the price, as a fundamentalist who expects it to revert to the fundamental value
or as a chartist who expects the last change to go on, and quotes a limit order for one share at its forecast. The
order book matches the orders with price-time priority, and every trade moves money and a share between the two
traders. Then traders copy the strategy of a neighbour whose wealth grew more. Fundamentalists hold the price near
the fundamental value, chartists amplify its swings, and imitation lets the share of chartists wander, so the
returns come in bursts with fat tails (Caiani et al. 2016 ch. 3).
Args:
width:
The number of columns of the grid, with one trader per cell.
height:
The number of rows of the grid.
chartists:
The initial share of chartists.
fundamental:
The fundamental value of the share.
reversion:
The share of the gap to the fundamental value a fundamentalist expects to close.
extrapolation:
The share of the last change of the price a chartist expects to repeat.
noise:
The standard deviation of the relative noise of a forecast.
imitation:
The probability that a trader looks at a neighbour in a step.
random_state:
The seed or generator of the initial strategies.
Returns:
system:
The model, with the group `traders` and the book `book`, whose reporters are `book.price` and
`book.volume`, and the imitation `imitation` of the strategies of more successful neighbours.
Examples:
>>> from skcomplex.models.economics.finance import order_book
>>> order_book(5, 5).elements[0].check().prepare().size_
25
"""
rng = np.random.default_rng(random_state)
cells = [(x, y) for x in range(width) for y in range(height)]
traders = GroupElement(
'traders',
element=StockTraderElement,
size=len(cells),
values={
'strategy': (rng.random(len(cells)) < chartists).astype(float),
'money': 10 * fundamental,
'shares': 10.0,
'wealth': 20 * fundamental,
'fundamental': fundamental,
'reversion': reversion,
'extrapolation': extrapolation,
'noise': noise,
},
location=cells,
)
accounts = [
MapInteraction(
f'{name}_account',
target='traders',
state=name,
function=function,
form='agent',
phase='account',
)
for name, function in (('gain', StockTraderElement.get_gain), ('wealth', StockTraderElement.get_wealth))
]
return System(
elements=[traders],
interactions=[
BookInteraction(
'book',
requesters='traders',
providers='traders',
rule=ContinuousMatchingSpec(),
opening=fundamental,
),
*accounts,
ImitationInteraction(
'imitation',
target='traders',
state='strategy',
payoff='gain',
rule='better',
form='agent',
probability=imitation,
phase='imitation',
),
],
space=LatticeSpace((width, height), boundary='periodic', neighborhood='von_neumann'),
time=DiscreteTime(
schedule=PhasesScheduleSpec(
[
PhaseSpec('quote'),
PhaseSpec('trade'),
PhaseSpec('account'),
PhaseSpec('imitation', activation='simultaneous'),
],
),
),
)
Components
class StockTraderElement(Element):
"""Trader who forecasts the price as a fundamentalist or a chartist, quotes a limit order, and imitates.
A fundamentalist expects the price `p` to revert to the fundamental value `F`, `p + reversion (F - p)`, and a
chartist expects the last change to go on, `p + extrapolation (p - p')`, both with a relative noise. The trader
bids its expectation for one share when it is above the price and it can pay, and asks its expectation for one
share when it is below and it holds one. After trading, an `ImitationInteraction` of rule `'better'` lets a trader
whose wealth grew less than that of a random neighbour copy the neighbour's strategy.
Args:
label:
The label of the agent or group.
visual:
How the element is drawn, or `None` for `default_visual_`.
strategy:
0 for a fundamentalist, 1 for a chartist.
money:
The money of the trader.
shares:
The shares of the trader.
bid:
The limit price of a buy order.
ask:
The limit price of a sell order.
demand:
The shares wanted in the step.
supply:
The shares offered in the step.
wealth:
The money plus the shares at the last price.
gain:
The change of the wealth in the last step.
fundamental:
The fundamental value of the share.
reversion:
The share of the gap to the fundamental value a fundamentalist expects to close.
extrapolation:
The share of the last change of the price a chartist expects to repeat.
noise:
The standard deviation of the relative noise of a forecast.
Examples:
>>> from skcomplex.core.elements import GroupElement
>>> from skcomplex.models.economics.finance import StockTraderElement
>>> GroupElement('traders', element=StockTraderElement, size=4, values={'strategy': 1.0}).size
4
"""
def __init__(
self,
label: str,
visual: VisualSpec | None = VisualSpec('agents', {'color_by': 'strategy', 'color_range': (0.0, 1.0)}),
*,
strategy: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
money: Annotated[float, StateSpec()] = 1000.0,
shares: Annotated[float, StateSpec()] = 10.0,
bid: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
ask: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
demand: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
supply: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
wealth: Annotated[float, StateSpec()] = 0.0,
gain: Annotated[float, StateSpec()] = 0.0,
fundamental: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 100.0,
reversion: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.2,
extrapolation: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
noise: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.02,
) -> None:
super().__init__(label, visual=visual)
self.strategy = strategy
self.money = money
self.shares = shares
self.bid = bid
self.ask = ask
self.demand = demand
self.supply = supply
self.wealth = wealth
self.gain = gain
self.fundamental = fundamental
self.reversion = reversion
self.extrapolation = extrapolation
self.noise = noise
@classmethod
def _get_view(cls) -> type[StockTraderView]:
"""Return the view of the agents of the element."""
return StockTraderView
def _get_laws(self, s: SimpleNamespace) -> list[BaseLaw]:
"""Return the quoting rule in its phase.
Args:
s:
The namespace of the symbols the laws can use.
Returns:
laws:
The laws of the component.
"""
return [RuleLaw(self.quote, phase='quote')]
@staticmethod
def _prices(agent: StockTraderView, model: ModelView) -> tuple[float, float]:
"""Return the last two prices of the book, the fundamental value before the first."""
history = model.history('book.price', 2)
prices = [value for value in history if math.isfinite(value)] or [agent.fundamental]
return prices[-1], prices[0]
def quote(self, agent: StockTraderView, model: ModelView) -> None:
"""Forecast the price and quote a limit order for one share.
Args:
agent:
The trader.
model:
The model.
"""
price, previous = self._prices(agent, model)
if agent.strategy:
expected = heuristics.trend([previous, price], strength=model.parameter('extrapolation'))
else:
reversion = 1 - model.parameter('reversion')
expected = heuristics.fundamental([price], fundamental=model.parameter('fundamental'), reversion=reversion)
expected = max(LOWEST_PRICE, expected * (1 + model.parameter('noise') * model.rng.normal()))
buying = expected > price
new_bid, new_ask = (expected, 0.0) if buying else (0.0, expected)
agent.bid = new_bid
agent.ask = new_ask
agent.demand = float(buying and agent.money >= expected)
agent.supply = float(not buying and agent.shares >= 1)
@staticmethod
def get_gain(agent: StockTraderView, model: ModelView) -> float:
"""Return the change of the wealth valued at the last price, zero before the first valuation.
Args:
agent:
The trader.
model:
The model view of the accounting.
Returns:
gain:
The change of the wealth in the step.
"""
return StockTraderElement.get_wealth(agent, model) - agent.wealth if agent.wealth else 0.0
@staticmethod
def get_wealth(agent: StockTraderView, model: ModelView) -> float:
"""Return the money plus the shares valued at the last price.
Args:
agent:
The trader.
model:
The model view of the accounting.
Returns:
wealth:
The wealth of the trader.
"""
price, _ = StockTraderElement._prices(agent, model)
return agent.money + agent.shares * price
class BookInteraction(MatchingInteraction[ModelView]):
"""Limit order book that matches the orders of the traders and reports the last traded price.
The book is a matching of the traders with themselves, which binds the requesters' limit to their `bid` and the
providers' to their `ask`, and every trade is two transfers, money from buyer to seller and shares from seller to
buyer, so the total money and the total shares stay the same, which the default run checks as invariants. The book
reports the last traded `price`, the previous price when nothing trades, and the traded `volume`.
Args:
label:
The label of the book.
requesters:
The path of the group of buying traders.
providers:
The path of the group of selling traders.
rule:
The spec of the algorithm that matches the orders, such as `ContinuousMatchingSpec()`.
limit:
The states with the limit prices of the buyers and of the sellers.
quantity:
The states with the quantities the buyers want and the sellers offer.
allowance:
The state of a buyer with the money it can spend.
transfers:
The states every trade moves, money paid by the buyer and shares delivered to it.
phase:
The phase the book clears in.
on_match:
A function of every trade and the model view, run after the transfers, or `None`.
opening:
The price before the first trade.
Examples:
>>> from skcomplex.models.economics.finance import BookInteraction
>>> BookInteraction('book', requesters='traders', providers='traders').reports
('price', 'volume')
"""
reports: ClassVar[tuple[str, ...]] = ('price', 'volume')
def __init__(
self,
label: str,
*,
requesters: Annotated[str, RoleSpec()],
providers: Annotated[str, RoleSpec()],
rule: MatchingRule = RandomSearchSpec(),
limit: str | tuple[str, str] = ('bid', 'ask'),
quantity: str | tuple[str, str] = ('demand', 'supply'),
allowance: str = 'money',
transfers: Sequence[tuple[str, int]] = (('money', -1), ('shares', 1)),
phase: str | None = 'trade',
on_match: Callable[[MatchSpec, ModelView], None] | None = None,
opening: float = 100.0,
) -> None:
super().__init__(
label,
requesters=requesters,
providers=providers,
rule=rule,
limit=limit,
quantity=quantity,
allowance=allowance,
transfers=transfers,
phase=phase,
on_match=on_match,
)
self.opening = opening
def _report_matches(self, matches: list[MatchSpec], unmatched: float, model: ModelView) -> None:
"""Report the last traded price, the previous price when nothing trades, and the traded volume.
Args:
matches:
The trades of the clearing.
unmatched:
The quantity of both sides left after the clearing, which the book does not report.
model:
The model view of the book.
"""
del unmatched
last = model.history('price', 1)
price = matches[-1].value if matches else (float(last[-1]) if len(last) else self.opening)
model.report('price', price)
model.report('volume', sum(match.quantity for match in matches))
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 ImitationInteraction(NeighborInteraction):
"""Imitation of a discrete state of the neighbours of every member, at a rate or with a probability per step.
By the rule `'random'` a member copies a neighbour chosen at random, by `'majority'` it takes the option that more
of its neighbours hold than any other, and by `'best'` it copies the neighbour whose payoff is largest, when that
payoff is larger than its own. By `'better'` it picks a neighbour at random and copies it when the neighbour's
payoff is larger than its own. By `'fermi'` it picks a neighbour at random and copies it with probability
`1 / (1 + exp(-beta (P_j - P_i)))`, where `P_j` and `P_i` are the payoffs of the neighbour and of the member and
`beta` is the intensity. A member without such a neighbour keeps its state. The rules read the neighbourhood of
the space, the graph of a network or the neighbourhood of a lattice.
In the `'rate'` form every member imitates at the rate `rate`, by jumps between every pair of options in continuous
time. In the `'probability'` form every member imitates with probability `probability` in every step of a
discrete-time system, all members at once in its `phase`, reading the states the phases before it set. In the
`'agent'` form the target is an agent group, and every agent imitates its neighbours in the space with probability
`probability` in its `phase`. The agents imitate at once, reading the states of the step before, 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'`. The rule `'better'` runs in the agent form only. Among neighbours of equal largest payoff, the rule
`'best'` of the agent form copies the first under the ties `'first'` and one drawn uniformly under `'random'`.
Args:
label:
The label of the interaction, unique among the parts of its system.
target:
The path of the group on the nodes or sites.
state:
The name of the discrete state of the members.
payoff:
The name of the state that holds the payoff of the members, which the rules `'best'` and `'fermi'` read.
options:
The values the state takes.
rate:
The rate at which every member imitates, in the rate form.
rule:
`'random'`, `'majority'`, `'best'`, or `'fermi'`.
intensity:
The intensity `beta` of the Fermi rule.
form:
`'rate'` for imitation in continuous time, `'probability'` for imitation in discrete time, or `'agent'`
for imitation of the neighbours of the agents of an agent group.
probability:
The probability that a member imitates in a step, in the probability form.
layer:
The layer of the network whose edges the interaction runs on, or `None` for the graph of the network.
phase:
The phase of the step the agent and the probability forms run in, or `None` for the default phase.
schedule:
The order of the agents in the agent form, `'synchronous'`, `'sequential'`, `'random-sequential'`, or
`'member'`.
ties:
The neighbour the rule `'best'` of the agent form copies among ties, `'first'` or `'random'`.
Raises:
ValidationError:
The rule, the form, the schedule, or the ties are unknown, a schedule or ties other than the defaults are
given outside the agent form, or the options are not distinct.
Examples:
>>> from skcomplex.core.interactions import ImitationInteraction
>>> ImitationInteraction('imitation', target='voters', rule='majority').check().options
(0.0, 1.0)
>>> ImitationInteraction('imitation', target='players', rule='fermi', form='probability').check().form
'probability'
>>> ImitationInteraction('imitation', target='traders', rule='better', form='agent', phase='learn').check().rule
'better'
"""
def __init__(
self,
label: str,
*,
target: Annotated[str, RoleSpec()],
state: str = 'state',
payoff: str = 'payoff',
options: Sequence[float] = (0.0, 1.0),
rate: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
rule: str = 'random',
intensity: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
form: str = 'rate',
probability: Annotated[float, ParameterSpec(domain=RealSet.closed(0, 1))] = 1.0,
layer: str | None = None,
phase: str | None = None,
schedule: str = 'synchronous',
ties: str = 'first',
) -> None:
super().__init__(label, layer=layer)
self.target = target
self.state = state
self.payoff = payoff
self.options = options
self.rate = rate
self.rule = rule
self.intensity = intensity
self.form = form
self.probability = probability
self.phase = phase
self.schedule = schedule
self.ties = ties
def _check_agent_options(self) -> None:
"""Raise an error for an unknown schedule or ties, or either of them not default outside the agent form."""
self._check_variant('schedule', self.schedule, SCHEDULES)
self._check_variant('ties', self.ties, TIES)
if self.form != 'agent' and (self.schedule != 'synchronous' or self.ties != 'first'):
error_msg = (
f'The imitation {self.label!r} has the schedule {self.schedule!r} and the ties {self.ties!r}, which '
'only the agent form runs. Use form="agent", or schedule="synchronous" and ties="first".'
)
raise ValidationError(error_msg)
def _check(self) -> None:
"""Raise an error for an unknown rule, form, schedule, or ties, or options that are not distinct.
Raises:
ValidationError:
The rule, the form, the schedule, or the ties are unknown, a schedule or ties other than the defaults
are given outside the agent form, or the options are not distinct.
"""
super()._check()
self._check_variant('rule', self.rule, RULES, (BaseSpec,))
if self.form not in FORMS:
error_msg = f'The form of imitation is {self.form!r}. Use one of {", ".join(FORMS)}.'
raise ValidationError(error_msg)
if self.rule == 'better' and self.form != 'agent':
error_msg = f'The rule of imitation {self.rule!r} runs in the agent form only. Use form `agent`.'
raise ValidationError(error_msg)
self._check_agent_options()
if len(set(self.options)) < max(len(self.options), 2):
error_msg = f'The options of imitation are {self.options!r}. Give at least two distinct values.'
raise ValidationError(error_msg)
def _get_neighborhood(self) -> AdjacencyNeighborhood | None:
"""Return the neighbourhood of the layer, or `None` for the neighbourhood of the space."""
return None if self.layer is None else AdjacencyNeighborhood(layer=self.layer)
def _get_fermi(self, member: SimpleNamespace, other: SimpleNamespace, intensity: Expression) -> Expression:
"""Return the Fermi probability that a member copies another of a different payoff."""
return 1 / (1 + exp(-intensity * (getattr(other, self.payoff) - getattr(member, self.payoff))))
def _get_counts(self, namespace: SimpleNamespace) -> dict[float, Expression]:
"""Return the number of the neighbours of the target that hold every option."""
source, over = self._get_member(namespace, 'source'), self._get_neighborhood()
return {
option: neighbors(where(getattr(source, self.state) == option, 1, 0), over=over) for option in self.options
}
def _get_majority(self, namespace: SimpleNamespace) -> dict[float, Expression]:
"""Return the indicator of every option that more of the neighbours of the target hold than any other."""
counts = self._get_counts(namespace)
leads = {}
for option, count in counts.items():
lead = SR(1)
for other, other_count in counts.items():
if other != option:
lead = lead * where(count > other_count, 1, 0)
leads[option] = lead
return leads
def _get_best(self, namespace: SimpleNamespace) -> tuple[Expression, Expression]:
"""Return the option of the neighbour of largest payoff and whether that payoff is larger than the target's."""
source, over = self._get_member(namespace, 'source'), self._get_neighborhood()
payoff = getattr(source, self.payoff)
best = neighbors(getattr(source, self.state), over=over, reduce='argmax', weight=payoff)
largest = neighbors(payoff, over=over, reduce='max')
return best, where(largest > getattr(self._get_member(namespace, 'target'), self.payoff), 1, 0)
def _switches(self, value: Expression, rate: Expression, chosen: Expression) -> list[BaseLaw]:
"""Return the jumps of a state from every option to the chosen option, at a rate when it is chosen."""
return [
JumpLaw(rate * where(value == first, 1, 0) * where(chosen == second, 1, 0), {value: second - first})
for first, second in itertools.permutations(self.options, 2)
]
def _copies(self, member: SimpleNamespace, other: SimpleNamespace, namespace: SimpleNamespace) -> list[BaseLaw]:
"""Return the copies of the state of the other end of an edge by a member, chosen among its neighbours."""
value, rate = getattr(member, self.state), self._get_parameter_symbol('rate', namespace) / member.degree
if self.rule == 'fermi':
rate = rate * self._get_fermi(member, other, self._get_parameter_symbol('intensity', namespace))
return self._switches(value, rate, getattr(other, self.state))
def _get_jumps(self, namespace: SimpleNamespace) -> list[BaseLaw]:
"""Return the jumps of the state at the rate of imitation, in continuous time."""
value = getattr(self._get_member(namespace, 'target'), self.state)
if self.rule == 'majority':
leads = self._get_majority(namespace)
return [
JumpLaw(
self._get_parameter_symbol('rate', namespace) * where(value == first, 1, 0) * leads[second],
{value: second - first},
)
for first, second in itertools.permutations(self.options, 2)
]
if self.rule == 'best':
best, better = self._get_best(namespace)
return self._switches(value, self._get_parameter_symbol('rate', namespace) * better, best)
if self.rule not in ('random', 'fermi'):
raise self._get_variant_error(self.rule, 'rule', '_get_jumps')
source, target = self._get_member(namespace, 'source'), self._get_member(namespace, 'target')
return [*self._copies(target, source, namespace), *self._copies(source, target, namespace)]
def _draw_copy(self, namespace: SimpleNamespace) -> Expression:
"""Return the next state of the target by the random or the Fermi rule, from the option it copies or its own."""
source, target = self._get_member(namespace, 'source'), self._get_member(namespace, 'target')
value = getattr(target, self.state)
probability, over = self._get_parameter_symbol('probability', namespace), self._get_neighborhood()
counts = self._get_counts(namespace)
degree = sum(counts.values(), SR(0))
if self.rule == 'fermi':
fermi = self._get_fermi(target, source, self._get_parameter_symbol('intensity', namespace))
weights = [
probability * neighbors(where(getattr(source, self.state) == option, 1, 0) * fermi, over=over)
for option in self.options
]
else:
weights = [probability * counts[option] for option in self.options]
stay = degree - sum(weights, SR(0)) + where(degree == 0, 1, 0)
index = categorical(f'{value}_imitation', [stay, *weights])
return value + sum(
(where(index == position + 1, option - value, 0) for position, option in enumerate(self.options)),
SR(0),
)
def _get_update(self, namespace: SimpleNamespace) -> list[BaseLaw]:
"""Return the update of the state of the target with the probability of imitation, in discrete time."""
value = getattr(self._get_member(namespace, 'target'), self.state)
if self.rule in ('random', 'fermi'):
return [UpdateLaw(value, self._draw_copy(namespace), phase=self.phase)]
if self.rule == 'majority':
leads = self._get_majority(namespace)
imitated = value + sum((lead * (option - value) for option, lead in leads.items()), SR(0))
elif self.rule != 'best':
raise self._get_variant_error(self.rule, 'rule', '_get_update')
else:
best, better = self._get_best(namespace)
imitated = where(better == 1, best, value)
return [
UpdateLaw(
value,
where(
uniform(f'{value}_imitation') < self._get_parameter_symbol('probability', namespace),
imitated,
value,
),
phase=self.phase,
),
]
def _get_laws(self, namespace: SimpleNamespace) -> list[BaseLaw]:
"""Return the jumps or the update of the state by the rule of the imitation.
Args:
namespace:
The namespace of an edge.
Returns:
laws:
The jumps of the state between every pair of options, or the update of the state of every member.
"""
if self.form == 'agent' and self.schedule == 'member':
return [RuleLaw(self._imitate_member, phase=self.phase, members=self._get_role('target'))]
if self.form == 'agent':
return [RuleLaw(self._imitate, phase=self.phase)]
if self.form == 'probability':
return self._get_update(namespace)
return self._get_jumps(namespace)
def _choose(
self,
agent: AgentViewProtocol,
neighbours: Sequence[AgentViewProtocol],
model: ModelViewProtocol,
) -> float | None:
"""Return the option an agent copies from its neighbours by the rule, or `None` to keep its own."""
rng = model.rng
if self.rule == 'majority':
counts = [sum(getattr(other, self.state) == option for other in neighbours) for option in self.options]
leaders = np.flatnonzero(np.asarray(counts) == max(counts))
return float(self.options[int(leaders[0])]) if len(leaders) == 1 else None
if self.rule == 'random':
return float(getattr(neighbours[int(rng.integers(len(neighbours)))], self.state))
own = float(getattr(agent, self.payoff))
if self.rule == 'best':
payoffs = np.array([getattr(other, self.payoff) for other in neighbours], dtype=float)
best = self._pick_best(payoffs, self.ties, model)
return float(getattr(neighbours[best], self.state)) if payoffs[best] > own else None
other = neighbours[int(rng.integers(len(neighbours)))]
difference = float(getattr(other, self.payoff)) - own
if self.rule == 'better':
return float(getattr(other, self.state)) if difference > 0 else None
if self.rule != 'fermi':
raise self._get_variant_error(self.rule, 'rule', '_choose')
fermi = 1 / (1 + np.exp(-self._get_parameter_value('intensity', model) * difference))
return float(getattr(other, self.state)) if rng.random() < fermi else None
def _get_choice(self, agent: AgentViewProtocol, model: ModelViewProtocol) -> float | None:
"""Return the option an agent copies with the probability of imitation, or `None` to keep its own."""
groups = set(model.groups(self._get_role('target')))
neighbours = [other for other in agent.neighbors(layer=self.layer) if other.group in groups]
if not neighbours or model.rng.random() >= self._get_parameter_value('probability', model):
return None
return self._choose(agent, neighbours, model)
def _set_choice(self, agent: AgentViewProtocol, choice: float | None) -> None:
"""Set the state of an agent to the option it copies, if any."""
if choice is not None:
setattr(agent, self.state, choice)
def _imitate_member(self, agent: AgentViewProtocol, model: ModelViewProtocol) -> None:
"""Let one agent imitate its neighbours with the probability of imitation, in its own activation."""
if agent.alive:
self._set_choice(agent, self._get_choice(agent, model))
def _imitate(self, model: ModelViewProtocol) -> None:
"""Let every agent imitate its neighbours with the probability of imitation, by the schedule."""
self._update_agents(model, self.schedule, self._get_choice, self._set_choice)