Minority game and its evolutionary version
The minority game of an odd number of agents with two random strategies each.
- Name:
minority_game - Field:
economics/econophysics - Formalism: agent-based model
- References: Challet and Zhang 1997; Yakovenko and Rosser 2009; Carmichael and Hadzikadic 2019 ch. 9
Factory
@register_model(
field='economics/econophysics',
tags=('minority game', 'agents', 'inductive reasoning', 'evolution'),
references=('Challet and Zhang 1997', 'Yakovenko and Rosser 2009', 'Carmichael and Hadzikadic 2019 ch. 9'),
defaults={'until': 500},
)
def minority_game(
n_agents: int = 101,
*,
memory: int = 3,
evolutionary: bool = False,
period: int = 50,
fraction: float = 0.1,
random_state: int | np.random.Generator | None = 0,
) -> System:
"""Return the minority game of an odd number of agents with two random strategies each.
Args:
n_agents:
The number of agents, odd so that there is always a minority.
memory:
The memory `m` of the strategies, so `alpha = 2^m / N`.
evolutionary:
Whether every `period` rounds the poorest agents take new random strategies.
period:
The number of rounds between two evolutions of the evolutionary version.
fraction:
The fraction of poorest agents that take new strategies at an evolution.
random_state:
The seed or generator of the strategies.
Returns:
system:
The model, with the group `players`, the choice `choice` of the strategy of higher virtual score, the
interaction `game`, which reports the attendance `game.attendance`, the cumulative reinforcement
`scoring` of the virtual scores, and in the evolutionary version the truncation `evolution`, which gives
the poorest agents new random strategies and no wealth every `period` rounds.
Raises:
ValueError:
The number of agents is even.
Examples:
>>> from skcomplex.models.economics.econophysics import minority_game
>>> minority_game(11, memory=2).interactions[1].memory
2
"""
if n_agents % 2 == 0:
error_msg = f'The minority game needs an odd number of agents, got {n_agents}.'
raise ValueError(error_msg)
codes = np.random.default_rng(random_state).integers(MAX_SEED, size=(2, n_agents)).astype(float)
players = GroupElement(
'players',
element=MinorityPlayerElement,
size=n_agents,
values={'first': codes[0], 'second': codes[1]},
)
choice = ChoiceInteraction(
'choice',
target='players',
state='strategy',
options=(0.0, 1.0),
utility=lambda agent, model, option: agent.second_score if option else agent.first_score,
rule='best',
form='agent',
phase='choose',
)
game = MinorityGameInteraction('game', players='players', memory=memory, phase='play')
scoring = ReinforcementInteraction(
'scoring',
target='players',
scores=('first_score', 'second_score'),
payoff=game.payoff,
rule='cumulative',
form='agent',
phase='score',
)
evolution = EvolutionInteraction(
'evolution',
target='players',
strategy=('first', 'second'),
fitness='wealth',
replacement='truncation',
fraction=fraction,
mutation=1.0,
change='any',
alphabet=MAX_SEED,
period=period,
reset={'first_score': 0.0, 'second_score': 0.0, 'wealth': 0.0},
phase='score',
)
return System(
elements=[players],
interactions=[choice, game, scoring, *([evolution] if evolutionary else [])],
time=DiscreteTime(
schedule=PhasesScheduleSpec([PhaseSpec('choose'), PhaseSpec('play'), PhaseSpec('score')]),
),
)
Components
class MinorityPlayerElement(Element):
"""Agent of the minority game, who holds two strategies and plays the one it chose.
Args:
label:
The label of the agent or group.
visual:
How the element is drawn, or `None` for `default_visual_`.
first:
The code of the first strategy.
second:
The code of the second strategy.
first_score:
The virtual score of the first strategy, the number of times it would have been in the minority.
second_score:
The virtual score of the second strategy.
strategy:
The strategy played, 0 for the first and 1 for the second.
action:
The last action, `+1` or `-1`.
wealth:
The net number of times the agent was in the minority.
Examples:
>>> from skcomplex.core.elements import GroupElement
>>> from skcomplex.models.economics.econophysics import MinorityPlayerElement
>>> GroupElement('players', element=MinorityPlayerElement, size=11, values={'first': 3, 'second': 5}).size
11
"""
def __init__(
self,
label: str,
visual: VisualSpec | None = None,
*,
first: Annotated[float, StateSpec(domain=NN)] = 0,
second: Annotated[float, StateSpec(domain=NN)] = 1,
first_score: Annotated[float, StateSpec()] = 0.0,
second_score: Annotated[float, StateSpec()] = 0.0,
strategy: Annotated[float, StateSpec(domain=FiniteEnumeratedSet([0, 1]))] = 0.0,
action: Annotated[float, StateSpec(domain=ZZ)] = 0.0,
wealth: Annotated[float, StateSpec()] = 0.0,
) -> None:
super().__init__(label, visual=visual)
self.first = first
self.second = second
self.first_score = first_score
self.second_score = second_score
self.strategy = strategy
self.action = action
self.wealth = wealth
@classmethod
def _get_view(cls) -> type[MinorityPlayerView]:
"""Return the view of the agents of the element."""
return MinorityPlayerView
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 MinorityGameInteraction(Interaction):
"""Rounds of the minority game of Challet and Zhang, where the agents on the minority side win.
Every round each agent plays the action, `+1` or `-1`, that its chosen strategy assigns to the last `m` winning
sides, the strategy of higher virtual score that a `ChoiceInteraction` of rule `'best'` chooses before. The side
with fewer agents wins, and every agent on it gains one point and every other loses one. The game reports the
attendance `A = sum_i a_i` and the code of the history the strategies read, so a cumulative
`ReinforcementInteraction` gives every strategy one virtual point when it would have played the winning side, by
the payoff `payoff`. The volatility `sigma^2 / N` of the attendance has a minimum below the value of random play
near `alpha = 2^m / N ~ 0.34`.
Args:
label:
The label of the interaction.
players:
The path of the group of agents.
memory:
The memory `m`, the number of past winning sides the strategies read.
phase:
The phase the rounds run in, or `None` for the default phase.
Examples:
>>> from skcomplex.models.economics.econophysics import MinorityGameInteraction
>>> MinorityGameInteraction('game', players='players', memory=4).memory
4
"""
reports: ClassVar[tuple[str, ...]] = ('attendance', 'history')
def __init__(
self,
label: str,
*,
players: Annotated[str, RoleSpec()],
memory: int = 3,
phase: str | None = None,
) -> None:
super().__init__(label)
self.players = players
self.memory = memory
self.phase = phase
def _get_laws(self, s: SimpleNamespace) -> list[BaseLaw]:
"""Return the rule of a round.
Args:
s:
The namespace of the symbols the laws can use.
Returns:
laws:
The laws of the component.
"""
return [RuleLaw(self.play, phase=self.phase)]
def _table(self, code: float, model: ModelView) -> IntArray:
"""Return the table of a strategy, cached for the run in the state of the game."""
state = self._state(model)
if code not in state.tables:
state = replace(state, tables={**state.tables, code: strategy_table(code, self.memory)})
model.storage['game'] = state
return state.tables[code]
def _state(self, model: ModelView) -> _GameStateSpec:
"""Return the state of the game in the storage of the model, starting a random history at the first round."""
state = model.storage.get('game')
if not isinstance(state, _GameStateSpec):
state = _GameStateSpec(int(model.rng.integers(2**self.memory)))
model.storage['game'] = state
return state
def play(self, model: ModelView) -> None:
"""Play one round, pay the agents, and update and report the history.
Args:
model:
The model.
"""
agents = model.agents('players', MinorityPlayerView)
history = self._state(model).history
for agent in agents:
code = agent.second if agent.strategy else agent.first
agent.action = float(2 * int(self._table(code, model)[history]) - 1)
attendance = sum(agent.action for agent in agents)
winner = -1 if attendance > 0 else 1
for agent in agents:
agent.wealth += agent.action * winner
state = self._state(model)
model.storage['game'] = replace(state, history=(2 * history + (winner > 0)) % 2**self.memory)
model.report('attendance', attendance)
model.report('history', float(history))
def payoff(self, agent: MinorityPlayerView, model: ModelView, option: int) -> float:
"""Return the virtual payoff of a strategy of an agent in the last round, `+1` if it played the winning side.
Args:
agent:
The agent.
model:
The model view of the rule that reads the payoff.
option:
The strategy, 0 for the first and 1 for the second.
Returns:
payoff:
`+1` when the strategy would have played the winning side, or `-1`.
"""
history = int(model.history(f'{self.label}.history')[-1])
winner = -1 if model.history(f'{self.label}.attendance')[-1] > 0 else 1
code = agent.second if option else agent.first
return float((2 * int(strategy_table(code, self.memory)[history]) - 1) * winner)
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