Brock-Hommes heuristic switching
Brock and Hommes' asset market with traders switching between fundamentalists and trend chasers.
- Name:
brock_hommes - Field:
economics/markets - Formalism: agent-based model
- References: Brock and Hommes 1998
Factory
@register_model(
field='economics/markets',
tags=('asset pricing', 'heuristics', 'bounded rationality', 'agents'),
references=('Brock and Hommes 1998',),
defaults={'until': 100},
)
def brock_hommes(
n_traders: int = 1000,
*,
intensity: float = 1.0,
rate: float = 1.1,
strength: float = 1.2,
cost: float = 1.0,
risk: float = 1.0,
noise: float = 0.01,
initial: float = 0.1,
) -> System:
"""Return Brock and Hommes' asset market with traders switching between fundamentalists and trend chasers.
Every step the traders score every heuristic by its last realised fitness and choose a heuristic by a logit choice
of intensity `beta` on its last realised fitness, and the market maker clears the deviation of the price from the
fundamental at `R x_t = sum_h n_h f_h`. The fundamental steady state is stable while the share of trend chasers
there, `1 / (1 + exp(-beta C))`, is below `R / g`, and beyond the critical intensity `beta* = ln(R / (g - R)) / C`
the price fluctuates persistently around the fundamental.
Args:
n_traders:
The number of traders.
intensity:
The intensity of choice `beta`.
rate:
The gross return `R` of the riskless asset.
strength:
The extrapolation `g` of the trend chasers.
cost:
The cost `C` of the fundamentalist heuristic.
risk:
The risk aversion times the variance of returns, `a sigma^2`.
noise:
The standard deviation of the noise of the price.
initial:
The initial deviation of the price from its fundamental value.
Returns:
system:
The model, with the group `traders`, their scores `learning` and choice `choice`, and the institution
`pricing`, whose reporters are the deviation of the price `pricing.deviation`, the share of trend chasers
`pricing.trend_share`, and the fitnesses `pricing.fitness_fundamentalist` and `pricing.fitness_trend`.
Examples:
>>> from skcomplex.models.economics.markets import brock_hommes
>>> brock_hommes(100).elements[0].check().prepare().size_
100
"""
traders = GroupElement('traders', element=HeuristicTraderElement, size=n_traders)
learning = ReinforcementInteraction(
'learning',
target='traders',
scores=tuple(f'{name}_fitness' for name in PREDICTORS),
payoff=HeuristicTraderElement.get_realised_fitness,
rate=1.0,
form='agent',
phase='learning',
)
choice = ChoiceInteraction(
'choice',
target='traders',
state='predictor',
options=(0, 1),
utility=HeuristicTraderElement.get_fitness,
intensity=intensity,
form='agent',
phase='choice',
)
pricing = HeuristicPricingInteraction(
PRICING,
traders='traders',
phase='pricing',
rate=rate,
strength=strength,
cost=cost,
risk=risk,
noise=noise,
initial=initial,
)
return System(
elements=[traders],
interactions=[learning, choice, pricing],
time=DiscreteTime(
schedule=PhasesScheduleSpec([PhaseSpec('learning'), PhaseSpec('choice'), PhaseSpec('pricing')]),
),
)
Components
class HeuristicTraderElement(Element):
"""Trader who holds the index of the forecasting heuristic in use, 0 for the fundamentalist and 1 for the trend.
The trader scores every heuristic by its last realised fitness, `fundamentalist_fitness` and `trend_fitness`, which
a reinforcement of rate 1 sets every step and a logit choice reads.
Args:
label:
The label of the agent or group.
visual:
How the element is drawn, or `None` for `default_visual_`.
predictor:
The index of the heuristic in use.
fundamentalist_fitness:
The last realised fitness of the fundamentalist heuristic.
trend_fitness:
The last realised fitness of the trend heuristic.
Examples:
>>> from skcomplex.core.elements import GroupElement
>>> from skcomplex.models.economics.markets import HeuristicTraderElement
>>> GroupElement('traders', element=HeuristicTraderElement, size=10, values={'predictor': 1.0}).size
10
"""
def __init__(
self,
label: str,
visual: VisualSpec | None = None,
*,
predictor: Annotated[float, StateSpec(domain=NN)] = 0.0,
fundamentalist_fitness: Annotated[float, StateSpec()] = 0.0,
trend_fitness: Annotated[float, StateSpec()] = 0.0,
) -> None:
super().__init__(label, visual=visual)
self.predictor = predictor
self.fundamentalist_fitness = fundamentalist_fitness
self.trend_fitness = trend_fitness
@classmethod
def _get_view(cls) -> type[HeuristicTraderView]:
"""Return the view of a speculator, with its heuristic."""
return HeuristicTraderView
@staticmethod
def get_fitness(agent: HeuristicTraderView, model: ModelView, option: float) -> float:
"""Return the score of a heuristic of a trader, its last realised fitness.
Args:
agent:
The trader.
model:
The model view of the choice.
option:
The index of the heuristic.
Returns:
fitness:
The last realised fitness of the heuristic.
"""
return float(getattr(agent, f'{PREDICTORS[int(option)]}_fitness'))
@staticmethod
def get_realised_fitness(agent: HeuristicTraderView, model: ModelView, index: int) -> float:
"""Return the fitness of a heuristic that the market maker reported last, zero before the first report.
Args:
agent:
The trader.
model:
The model view of the reinforcement.
index:
The index of the heuristic.
Returns:
fitness:
The last reported fitness of the heuristic.
"""
reported = model.history(f'{PRICING}.fitness_{PREDICTORS[index]}', 1)
return float(reported[-1]) if len(reported) else 0.0
class ReinforcementInteraction(Interaction):
"""Reinforcement of the scores of options, each moving towards the payoff its option realized in the step.
Every step, the score `s_k` of option `k` of every member moves towards the realized payoff `P_k` at the rate `a`,
`s_k <- s_k + a (P_k - s_k)`, so a score is an exponentially weighted mean of past payoffs, and the last payoff at
the rate 1, by the rule `'average'`. By the rule `'cumulative'` the score adds the payoff, `s_k <- s_k + P_k`, so
it is the sum of past payoffs, such as a virtual score. In the agent form the scores may be the components of one
vector state, one per option, and a group of one member holds scores that several groups share. In the
`'symbolic'` form, `payoff(namespace, k)` returns the payoff of option `k` as a symbolic
expression, where `namespace.target` holds the states and parameters of the group. In the `'agent'` form, the
group is an agent group and `payoff(agent, model, k)` returns the payoff as a number, so it can read the agent and
the history and storage of the model view. With `chosen`, the name of a state that holds the index of the option a
member chose, only the score of that option moves, and the payoff of the other options is not read. In the agent
form every agent reinforces at once under the schedule `'synchronous'`, one after the other in their order or in a
new random order every step under `'sequential'` or `'random-sequential'`, and 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, such as after a choice and a transfer in one visit.
Args:
label:
The label of the interaction, unique among the parts of its system.
target:
The path of the element or group whose members hold the scores.
scores:
The names of the states that hold the score of every option, in the order of the options, or in the agent
form the name of one vector state whose components are the scores.
payoff:
The function `payoff(namespace, k)` in the symbolic form, or `payoff(agent, model, k)` in the agent form,
that returns the realized payoff of option `k`.
rate:
The rate `a` at which the scores move towards the payoffs, between 0 and 1, by the rule `'average'`.
rule:
`'average'` to move the scores towards the payoffs, or `'cumulative'` to add the payoffs to the scores.
form:
`'symbolic'` for update laws of the states of a group, 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.
chosen:
The name of the state that holds the index of the option a member chose, so only its score moves, or
`None` to move every score.
schedule:
The order of the agent form, `'synchronous'` for every agent at once, `'sequential'` or
`'random-sequential'` for one agent after the other, or `'member'` for every agent in its own activation.
Raises:
ValidationError:
The form, the rule, or the schedule is unknown, the symbolic form runs member by member, there are no
scores, or the payoff is not callable.
Examples:
>>> from skcomplex.core.interactions import ReinforcementInteraction
>>> reinforcement = ReinforcementInteraction(
... 'reinforcement', target='learners', scores=('first', 'second'), payoff=lambda s, k: k, rate=0.5
... )
>>> reinforcement.check().rate
0.5
>>> ReinforcementInteraction(
... 'virtual', target='players', scores='scores', payoff=lambda a, m, k: k, rule='cumulative', form='agent'
... ).check().rule
'cumulative'
>>> ReinforcementInteraction(
... 'satisfaction', target='customers', scores=('s1', 's2'), payoff=lambda a, m, k: 1.0, chosen='shop',
... schedule='member', form='agent',
... ).check().chosen
'shop'
"""
def __init__(
self,
label: str,
*,
target: Annotated[str, RoleSpec()],
scores: str | Sequence[str],
payoff: Callable[..., float | Expression],
rate: Annotated[float, ParameterSpec(domain=RealSet.closed(0, 1))] = 0.1,
rule: str = 'average',
form: str = 'symbolic',
phase: str | None = None,
chosen: str | None = None,
schedule: str = 'synchronous',
) -> None:
super().__init__(label)
self.target = target
self.scores = scores
self.payoff = payoff
self.rate = rate
self.rule = rule
self.form = form
self.phase = phase
self.chosen = chosen
self.schedule = schedule
def _check(self) -> None:
"""Raise an error for an unknown variant, no scores, or a payoff that cannot be called.
Raises:
ValidationError:
The type declares states, the form, the rule, or the schedule is unknown, the symbolic form runs member
by member, there are no scores, or the payoff is not callable.
"""
super()._check()
for name, value, variants in (
('form', self.form, FORMS),
('rule', self.rule, RULES),
('schedule', self.schedule, SCHEDULES),
):
if value not in variants:
error_msg = f'Unknown {name} {value!r} of {self.label!r}. Use one of {variants}.'
raise ValidationError(error_msg)
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)
vector = isinstance(self.scores, str) and bool(self.scores) and self.form == 'agent'
if not vector and (isinstance(self.scores, str) or not self.scores):
error_msg = (
f'{self.label!r} has the scores {self.scores!r}. Give a sequence of state names, one per option, or '
'in the agent form the name of a vector state.'
)
raise ValidationError(error_msg)
if not callable(self.payoff):
error_msg = f'The payoff of {self.label!r} must be callable, but it is {self.payoff!r}.'
raise ValidationError(error_msg)
def _read(self, agent: AgentViewProtocol) -> Array:
"""Return the scores of an agent, one per option."""
if isinstance(self.scores, str):
return np.array(getattr(agent, self.scores), dtype=float).ravel()
return np.array([getattr(agent, name) for name in self.scores], dtype=float)
def _get_scores(self, agent: AgentViewProtocol, model: ModelViewProtocol) -> Array:
"""Return the new scores of an agent, every score or only that of its chosen option moved by its payoff."""
old = self._read(agent)
options = range(len(old)) if self.chosen is None else [int(getattr(agent, self.chosen))]
rate = 1.0 if self.rule == 'cumulative' else self._get_parameter_value('rate', model)
change = np.zeros_like(old)
for k in options:
payoff = float(self.payoff(agent, model, k))
change[k] = payoff if self.rule == 'cumulative' else rate * (payoff - old[k])
return old + change
def _write(self, agent: AgentViewProtocol, scores: Array) -> None:
"""Set the scores of an agent."""
if isinstance(self.scores, str):
setattr(agent, self.scores, scores)
return
for name, value in zip(self.scores, scores, strict=True):
setattr(agent, name, float(value))
def _reinforce_member(self, agent: AgentViewProtocol, model: ModelViewProtocol) -> None:
"""Move the scores of one agent towards its payoffs, in its own activation."""
if agent.alive:
self._write(agent, self._get_scores(agent, model))
def reinforce(self, model: ModelViewProtocol) -> None:
"""Move the scores of every agent towards the payoffs of its options, by the schedule.
Args:
model:
The model view of the interaction.
"""
self._update_agents(model, self.schedule, self._get_scores, self._write)
def _get_laws(self, namespace: SimpleNamespace) -> list[BasePart]:
"""Return the update of every score towards its payoff, or the rule of the agent form.
Args:
namespace:
The namespace of the symbols the laws can use, with `namespace.target` and `namespace.rate`.
Returns:
laws:
The update laws of the scores, or the rule that reinforces the scores of the agents.
"""
if self.form == 'agent' and self.schedule == 'member':
return [RuleLaw(self._reinforce_member, phase=self.phase, members=self._get_role('target'))]
if self.form == 'agent':
return [RuleLaw(self.reinforce, phase=self.phase)]
laws: list[BasePart] = []
for index, name in enumerate(self.scores):
score = getattr(self._get_member(namespace, 'target'), name)
payoff = self.payoff(namespace, index)
change = (
payoff
if self.rule == 'cumulative'
else self._get_parameter_symbol('rate', namespace) * (payoff - score)
)
if self.chosen is not None:
change = where(getattr(self._get_member(namespace, 'target'), self.chosen) == index, change, 0)
laws.append(UpdateLaw(score, score + change))
return laws
class ChoiceInteraction(Interaction):
"""Choice of every member of a group among options, drawn every step from the utilities of the options by a rule.
The rule `'logit'`, which is the heat-bath rule, draws option `k` of utility `u_k` with probability
`exp(beta u_k) / sum_j exp(beta u_j)`, where `beta` is the intensity of choice. The rules `'metropolis'` and
`'fermi'` propose one of the other options uniformly, or with the proposal `'all'` one of all the options
including the current one, and accept it with probability `min(1, exp(beta (u_new - u_old)))` or
`1 / (1 + exp(-beta (u_new - u_old)))`, so the rule `'fermi'` with the proposal `'all'` is the heat-bath rule of a
uniform proposal. The rule `'best'` draws uniformly among the options of largest utility under the ties
`'random'`, and takes the first of them under `'first'`. Every step, each member revises its choice with probability
`activity` and keeps it otherwise.
In the `'symbolic'` form, `utility(namespace, option)` returns the utility as a symbolic expression, where
`namespace.target` holds the states and parameters of the group, and every member draws at once. In the `'agent'`
form, the group is an agent group and `utility(agent, model, option)` returns the utility as a number, so it can
read the agent, its neighbours, and the history and storage of the model view. Its members draw at once, from the
choices of the last step, under the `'synchronous'` schedule, one after the other in their order or in a new random
order, each seeing the choices made before it, under the `'sequential'` or `'random-sequential'` schedule, and each
in its own activation under the `'member'` schedule, together with the rules of its element and the other member
interactions, in the order the system lists them. An option of utility `-inf` is never drawn in the agent form.
Args:
label:
The label of the interaction, unique among the parts of its system.
target:
The path of the element or group whose members choose.
state:
The name of the discrete state of the members that holds the chosen option.
options:
The values the state can take, at least two.
utility:
The function `utility(namespace, option)` in the symbolic form, or `utility(agent, model, option)` in the
agent form, that returns the utility of an option.
rule:
The rule of the draw, `'logit'`, `'metropolis'`, `'fermi'`, or `'best'`.
intensity:
The intensity of choice `beta`, zero for uniform choice.
schedule:
The order in which the members revise, `'synchronous'` or, in the agent form, `'sequential'`,
`'random-sequential'`, or `'member'`.
activity:
The probability that a member revises its choice in a step.
form:
`'symbolic'` for update laws of the states of a group, 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.
proposal:
The options the rules `'metropolis'` and `'fermi'` propose from, `'others'` for the options other than the
current one, or `'all'` for every option.
ties:
The option the rule `'best'` takes among options of equal largest utility, `'random'` for one drawn
uniformly or `'first'` for the first in the order of the options.
Raises:
ValidationError:
The rule, the schedule, the form, the proposal, or the ties are unknown, the schedule is not synchronous in
the symbolic form, there are fewer than two options, or the utility is not callable.
Examples:
>>> from skcomplex.core.interactions import ChoiceInteraction
>>> from skcomplex.core.laws.utils import mean
>>> choice = ChoiceInteraction(
... 'choice', target='spins', state='spin', options=(-1, 1), utility=lambda s, o: o * mean(s.target.spin)
... )
>>> choice.check().rule
'logit'
>>> agents = ChoiceInteraction(
... 'choice', target='traders', state='plan', options=(0, 1), utility=lambda a, m, o: o, form='agent'
... )
>>> type(agents.check().prepare().run(SimpleNamespace()).laws_[0]).__name__
'RuleLaw'
>>> ChoiceInteraction(
... 'heat-bath', target='spins', state='spin', options=(-1, 0, 1), utility=lambda s, o: o, rule='fermi',
... proposal='all'
... ).check().proposal
'all'
"""
def __init__(
self,
label: str,
*,
target: Annotated[str, RoleSpec()],
state: str,
options: Sequence[float],
utility: Callable[..., float | Expression],
rule: str = 'logit',
intensity: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
schedule: str = 'synchronous',
activity: Annotated[float, ParameterSpec(domain=RealSet.closed(0, 1))] = 1.0,
form: str = 'symbolic',
phase: str | None = None,
proposal: str = 'others',
ties: str = 'random',
) -> None:
super().__init__(label)
self.target = target
self.state = state
self.options = options
self.utility = utility
self.rule = rule
self.intensity = intensity
self.schedule = schedule
self.activity = activity
self.form = form
self.phase = phase
self.proposal = proposal
self.ties = ties
def _check_variants(self) -> None:
"""Raise an error for an unknown variant, or a symbolic choice that is not synchronous."""
self._check_variant('rule', self.rule, RULES, (BaseSpec,))
for name, value, variants in (
('schedule', self.schedule, SCHEDULES),
('form', self.form, FORMS),
('proposal', self.proposal, PROPOSALS),
('ties', self.ties, TIES),
):
if value not in variants:
error_msg = f'Unknown {name} {value!r} of {self.label!r}. Use one of {variants}.'
raise ValidationError(error_msg)
if self.form == 'symbolic' and self.schedule != 'synchronous':
error_msg = (
f'{self.label!r} has the schedule {self.schedule!r}, which only the agent form runs. Use '
"form='agent' or schedule='synchronous'."
)
raise ValidationError(error_msg)
def _check(self) -> None:
"""Raise an error for an unknown variant, fewer than two options, or a utility that cannot be called.
Raises:
ValidationError:
The type declares states, the rule, the schedule, the form, the proposal, or the ties are unknown, the
schedule is not synchronous in the symbolic form, there are fewer than two options, or the utility is
not callable.
"""
super()._check()
self._check_variants()
if len(self.options) < MIN_OPTIONS:
error_msg = f'{self.label!r} chooses among {list(self.options)}. Give at least {MIN_OPTIONS} options.'
raise ValidationError(error_msg)
if not callable(self.utility):
error_msg = f'The utility of {self.label!r} must be callable, but it is {self.utility!r}.'
raise ValidationError(error_msg)
def _select(self, index: Expression, values: Sequence[Expression]) -> Expression:
"""Return the value at a symbolic index."""
return sum((where(index == position, value, 0) for position, value in enumerate(values)), SR(0))
def _draw_softmax(self, state: Expression, utilities: Sequence[Expression], intensity: Expression) -> Expression:
"""Return the option drawn with probabilities proportional to `exp(beta u_k)`."""
index = categorical(f'{state}_option', [exp(intensity * utility) for utility in utilities])
return self._select(index, [SR(option) for option in self.options])
def _draw_best(self, state: Expression, utilities: Sequence[Expression]) -> Expression:
"""Return an option drawn uniformly among the options of largest utility."""
largest = utilities[0]
for utility in utilities[1:]:
largest = where(utility > largest, utility, largest)
weights = [where(utility >= largest, 1, 0) for utility in utilities]
if self.ties == 'first':
weights = [
weight * where(sum(weights[:position], SR(0)) > 0, 0, 1) for position, weight in enumerate(weights)
]
index = categorical(f'{state}_best', weights)
return self._select(index, [SR(option) for option in self.options])
def _draw_proposal(self, state: Expression, utilities: Sequence[Expression], intensity: Expression) -> Expression:
"""Return a uniform proposal of an option, accepted by the Metropolis or the Fermi probability."""
weights = [SR(1) if self.proposal == 'all' else where(state == option, 0, 1) for option in self.options]
index = categorical(f'{state}_proposal', weights)
proposal = self._select(index, [SR(option) for option in self.options])
gain = self._select(index, utilities) - sum(
(where(state == option, utility, 0) for option, utility in zip(self.options, utilities, strict=True)),
SR(0),
)
if self.rule == 'metropolis':
acceptance = where(gain >= 0, 1, exp(intensity * gain))
else:
acceptance = 1 / (1 + exp(-intensity * gain))
return where(uniform(f'{state}_acceptance') < acceptance, proposal, state)
def _draw_symbolic(self, namespace: SimpleNamespace) -> Expression:
"""Return the option a member draws, as a symbolic expression of the namespace."""
state = getattr(self._get_member(namespace, 'target'), self.state)
utilities = [SR(self.utility(namespace, option)) for option in self.options]
if self.rule == 'logit':
return self._draw_softmax(state, utilities, self._get_parameter_symbol('intensity', namespace))
if self.rule == 'best':
return self._draw_best(state, utilities)
if self.rule in ('metropolis', 'fermi'):
return self._draw_proposal(state, utilities, self._get_parameter_symbol('intensity', namespace))
raise self._get_variant_error(self.rule, 'rule', '_draw_symbolic')
def _get_utilities(self, agent: AgentViewProtocol, model: ModelViewProtocol) -> np.ndarray:
"""Return the utility of every option for an agent."""
return np.array([float(self.utility(agent, model, option)) for option in self.options])
def _draw_agent_proposal(self, agent: AgentViewProtocol, model: ModelViewProtocol, intensity: float) -> float:
"""Return a uniform proposal of an option of an agent if accepted, or its current option."""
current = float(getattr(agent, self.state))
others = [option for option in self.options if self.proposal == 'all' or option != current]
proposal = float(others[int(model.rng.integers(len(others)))])
before = float(self.utility(agent, model, current)) if current in self.options else 0.0
gain = float(self.utility(agent, model, proposal)) - before
if self.rule == 'metropolis':
acceptance = 1.0 if gain >= 0 else math.exp(intensity * gain)
else:
acceptance = float(expit(intensity * gain))
return proposal if model.rng.random() < acceptance else current
def _draw_agent(self, agent: AgentViewProtocol, model: ModelViewProtocol, intensity: float) -> float:
"""Return the option an agent draws by the rule."""
if self.rule in ('metropolis', 'fermi'):
return self._draw_agent_proposal(agent, model, intensity)
utilities = self._get_utilities(agent, model)
if self.rule == 'best':
return float(self.options[self._pick_best(utilities, self.ties, model)])
if self.rule != 'logit':
raise self._get_variant_error(self.rule, 'rule', '_draw_agent')
scaled = np.where(np.isneginf(utilities), -np.inf, intensity * utilities)
probabilities = softmax(scaled)
return float(self.options[int(model.rng.choice(len(self.options), p=probabilities))])
def _is_active(self, model: ModelViewProtocol, activity: float) -> bool:
"""Return whether a member revises its choice in this step."""
return activity >= 1 or bool(model.rng.random() < activity)
def _get_choice(self, agent: AgentViewProtocol, model: ModelViewProtocol) -> float | None:
"""Return the option an agent draws if it revises in this step, or `None` to keep its own."""
if not self._is_active(model, self._get_parameter_value('activity', model)):
return None
return self._draw_agent(agent, model, self._get_parameter_value('intensity', model))
def _set_choice(self, agent: AgentViewProtocol, option: float | None) -> None:
"""Set the state of an agent to the option it draws, if any."""
if option is not None:
setattr(agent, self.state, option)
def _choose_member(self, agent: AgentViewProtocol, model: ModelViewProtocol) -> None:
"""Set the state of one agent to the option it draws if it revises, in its own activation."""
if agent.alive:
self._set_choice(agent, self._get_choice(agent, model))
def choose(self, model: ModelViewProtocol) -> None:
"""Set the state of every agent that revises in this step to the option it draws, by the schedule.
Args:
model:
The model view of the interaction.
"""
self._update_agents(model, self.schedule, self._get_choice, self._set_choice)
def _get_laws(self, namespace: SimpleNamespace) -> list[BasePart]:
"""Return the update of the state of every member that revises, or the rule of the agent form.
Args:
namespace:
The namespace of the symbols the laws can use, with `namespace.target`, `namespace.intensity`, and
`namespace.activity`.
Returns:
laws:
The update law of the state, or the rule that runs the choice of the agents.
"""
if self.form == 'agent' and self.schedule == 'member':
return [RuleLaw(self._choose_member, phase=self.phase, members=self._get_role('target'))]
if self.form == 'agent':
return [RuleLaw(self.choose, phase=self.phase)]
state = getattr(self._get_member(namespace, 'target'), self.state)
revised = where(
uniform(f'{state}_activity') < self._get_parameter_symbol('activity', namespace),
self._draw_symbolic(namespace),
state,
)
return [UpdateLaw(state, revised)]
class HeuristicPricingInteraction(FeedbackInteraction[ModelView]):
"""Market maker that clears the risky asset at the price where the mean forecast of the traders is met.
The deviation `x` of the price from its fundamental value clears at `R x_t = sum_h n_h f_h + eps_t`, where `n_h`
is the share of traders using heuristic `h`. The fundamentalists forecast the fundamental, `f = 0`, and the trend
chasers extrapolate the last deviation, `f = g x_{t-1}`, both as `heuristics.fundamental` with a reversion of 0
and `g`. The realised fitness of a heuristic is the profit of its last forecast,
`(x_t - R x_{t-1}) (f_h(t - 1) - R x_{t-1}) / (a sigma^2)`, minus its cost.
Args:
label:
The label of the institution.
rule:
Unused, since the institution overrides `act`.
phase:
The phase the institution clears in.
traders:
The path of the group of traders.
rate:
The gross return `R` of the riskless asset.
strength:
The extrapolation `g` of the trend chasers.
cost:
The cost of the fundamentalist heuristic.
risk:
The risk aversion times the variance of returns, `a sigma^2`.
noise:
The standard deviation of the noise of the price.
initial:
The initial deviation of the price from its fundamental value.
Examples:
>>> from skcomplex.models.economics.markets import HeuristicPricingInteraction
>>> HeuristicPricingInteraction('pricing', traders='traders').reports
('deviation', 'trend_share', 'fitness_fundamentalist', 'fitness_trend')
"""
reports: ClassVar[tuple[str, ...]] = ('deviation', 'trend_share', 'fitness_fundamentalist', 'fitness_trend')
def __init__(
self,
label: str,
*,
rule: Callable[[ModelView], None] | None = None,
phase: str | None = None,
traders: Annotated[str, RoleSpec()],
rate: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 1.1,
strength: Annotated[float, ParameterSpec()] = 1.2,
cost: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
risk: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 1.0,
noise: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.01,
initial: Annotated[float, ParameterSpec()] = 0.1,
) -> None:
super().__init__(label, rule=rule, phase=phase)
self.traders = traders
self.rate = rate
self.strength = strength
self.cost = cost
self.risk = risk
self.noise = noise
self.initial = initial
def act(self, model: ModelView) -> None:
"""Clear the price from the forecasts of the traders' heuristics and report the fitnesses.
Args:
model:
The model view of the institution.
"""
rate, strength = model.parameter('rate'), model.parameter('strength')
state = model.storage.get('market')
if not isinstance(state, _MarketStateSpec):
state = _MarketStateSpec((model.parameter('initial'),))
history = list(state.history)
forecasts = np.array(
[heuristics.fundamental(history), heuristics.fundamental(history, fundamental=0.0, reversion=strength)],
)
traders = model.agents('traders', HeuristicTraderView)
share = float(np.mean([trader.predictor for trader in traders])) if traders else 0.0
price = float(forecasts @ [1 - share, share] + model.rng.normal(0.0, model.parameter('noise'))) / rate
previous = state.forecasts
excess = price - rate * history[-1]
if previous is None:
fitness = np.zeros(len(PREDICTORS))
else:
fitness = excess * (previous - rate * history[-1]) / model.parameter('risk')
fitness -= [model.parameter('cost'), 0.0]
model.storage['market'] = _MarketStateSpec((history[-1], price), forecasts)
model.report('deviation', price)
model.report('trend_share', share)
for name, value in zip(PREDICTORS, fitness, strict=True):
model.report(f'fitness_{name}', value)