El Farol
Arthur's El Farol bar problem, whose attendance self-organizes around the capacity.
- Name:
el_farol - Field:
economics/econophysics - Formalism: agent-based model
- References: Arthur 1994; Yakovenko and Rosser 2009; Carmichael and Hadzikadic 2019 ch. 9
Factory
@register_model(
field='economics/econophysics',
tags=('el farol', 'agents', 'inductive reasoning', 'bounded rationality'),
references=('Arthur 1994', 'Yakovenko and Rosser 2009', 'Carmichael and Hadzikadic 2019 ch. 9'),
defaults={'until': 200},
observables=lambda system: [ObservableSpec('attendance', group='patrons', variable='attends', reduce='sum')],
)
def el_farol(
n_agents: int = 100,
*,
capacity: float = 60.0,
n_predictors: int = 6,
discount: float = 0.9,
random_state: int | np.random.Generator | None = 0,
) -> System:
"""Return Arthur's El Farol bar problem, whose attendance self-organizes around the capacity.
Every patron holds a random set of predictors of next week's attendance, drawn from a pool of 46, and acts on
its currently most accurate one, going when it forecasts room in the bar. The attendance is random-looking, yet
its mean settles near the capacity. The patrons read the past attendances from the observable `attendance` of
the run, which the catalog lists, so a run passes it in `observables`.
Args:
n_agents:
The number of patrons.
capacity:
The attendance above which the bar is too crowded to enjoy.
n_predictors:
The number of predictors every patron holds.
discount:
The weight of past errors in the accuracy of a predictor.
random_state:
The seed or generator of the predictors that every patron holds.
Returns:
system:
The model, with the group `patrons`, the one-member group `pool` of the forecasts and the errors of the
predictors, the reinforcement `scoring` of the errors, and the choice `choice` to go when the forecast is
below the capacity of the patrons.
Examples:
>>> from skcomplex.models.economics.econophysics import el_farol
>>> el_farol(10).elements[0].check().prepare().size_
10
"""
rng = np.random.default_rng(random_state)
masks = [
float(sum(1 << int(index) for index in rng.choice(len(PREDICTORS), n_predictors, replace=False)))
for _ in range(n_agents)
]
patrons = GroupElement(
'patrons',
element=PatronElement,
size=n_agents,
values={'predictors': masks, 'capacity': capacity},
)
pool = GroupElement(
'pool',
element=PredictorPoolElement,
size=1,
values={'forecasts': np.zeros((1, len(PREDICTORS))), 'errors': np.zeros((1, len(PREDICTORS)))},
)
scoring = ReinforcementInteraction(
'scoring',
target='pool',
scores='errors',
payoff=PredictorPoolElement.error,
rate=1 - discount,
form='agent',
phase='learn',
)
choice = ChoiceInteraction(
'choice',
target='patrons',
state='attends',
options=(0.0, 1.0),
utility=lambda agent, model, option: option if agent.forecast < agent.capacity else -option,
rule='best',
form='agent',
phase='decide',
)
return System(
elements=[patrons, pool],
interactions=[scoring, choice],
time=DiscreteTime(
schedule=PhasesScheduleSpec(
[PhaseSpec(name) for name in ('learn', 'forecast', 'predict', 'decide')],
),
),
)
Components
class PatronElement(Element):
"""Patron of the El Farol bar, who forecasts the attendance with its most accurate predictor.
Every patron holds a few predictors of the pool, the set bits of `predictors`, and forecasts with the one of
smallest error in the pool. A choice then sends the patron to the bar when the forecast is below the capacity.
Args:
label:
The label of the agent or group.
visual:
How the element is drawn, or `None` for `default_visual_`.
predictors:
The bit mask of the predictors of the pool that the patron holds.
forecast:
The last forecast of the patron's most accurate predictor.
attends:
Whether the patron went to the bar this week, 1 or 0.
capacity:
The attendance above which the bar is too crowded to enjoy.
Examples:
>>> from skcomplex.core.elements import GroupElement
>>> from skcomplex.models.economics.econophysics import PatronElement
>>> GroupElement('patrons', element=PatronElement, size=100, values={'predictors': 7}).size
100
"""
pool: ClassVar[str] = 'pool'
def __init__(
self,
label: str,
visual: VisualSpec | None = None,
*,
predictors: Annotated[float, StateSpec(domain=NN)] = 1,
forecast: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
attends: Annotated[float, StateSpec(domain=FiniteEnumeratedSet([0, 1]))] = 0.0,
capacity: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 60.0,
) -> None:
super().__init__(label, visual=visual)
self.predictors = predictors
self.forecast = forecast
self.attends = attends
self.capacity = capacity
@classmethod
def _get_view(cls) -> type[PatronView]:
"""Return the view of the agents of the element."""
return PatronView
def _get_laws(self, s: SimpleNamespace) -> list[BaseLaw]:
"""Return the predicting rule.
Args:
s:
The namespace of the symbols the laws can use.
Returns:
laws:
The laws of the component.
"""
return [RuleLaw(self.predict, phase='predict')]
def predict(self, agent: PatronView, model: ModelView) -> None:
"""Forecast the attendance with the most accurate predictor held.
Args:
agent:
The patron.
model:
The model.
"""
pool = model.agents(self.pool, PredictorPoolView)[0]
errors, predicted = np.ravel(pool.errors), np.ravel(pool.forecasts)
mask = int(agent.predictors)
held = [index for index in range(len(PREDICTORS)) if mask >> index & 1]
agent.forecast = float(predicted[held[int(np.argmin(errors[held]))]])
class PredictorPoolElement(Element):
"""Pool of Arthur's predictors, with the forecast and the accuracy of every predictor that the patrons share.
The accuracy of a predictor depends only on the history of attendance, the observable `attendance` of the run,
so one member holds it for every patron. Every week the pool forecasts the attendance with every predictor, from
ten random weeks it draws at the first week followed by the observed attendances, and a
`ReinforcementInteraction` moves the error of every predictor towards its last absolute error, a discounted mean
of the errors.
Args:
label:
The label of the agent or group.
visual:
How the element is drawn, or `None` for `default_visual_`.
forecasts:
The forecast of next week's attendance by every predictor of the pool.
errors:
The discounted mean absolute error of every predictor of the pool.
Examples:
>>> from skcomplex.core.elements import GroupElement
>>> from skcomplex.models.economics.econophysics import PredictorPoolElement
>>> GroupElement('pool', element=PredictorPoolElement, size=1).size
1
"""
attendance: ClassVar[str] = 'attendance'
def __init__(
self,
label: str,
visual: VisualSpec | None = None,
*,
forecasts: Annotated[npt.ArrayLike, StateSpec(vector=True)] = 0.0,
errors: Annotated[npt.ArrayLike, StateSpec(vector=True)] = 0.0,
) -> None:
super().__init__(label, visual=visual)
self.forecasts = forecasts
self.errors = errors
@classmethod
def _get_view(cls) -> type[PredictorPoolView]:
"""Return the view of the pool."""
return PredictorPoolView
def _get_laws(self, s: SimpleNamespace) -> list[BaseLaw]:
"""Return the forecasting rule.
Args:
s:
The namespace of the symbols the laws can use.
Returns:
laws:
The laws of the component.
"""
return [RuleLaw(self.forecast, phase='forecast')]
def _history(self, model: ModelView) -> Array:
"""Return the attendances, after ten random weeks drawn at the first week."""
seeded = model.storage.get('seeded')
if not isinstance(seeded, np.ndarray):
seeded = model.rng.integers(0, int(MAX_ATTENDANCE) + 1, N_SEEDED_WEEKS).astype(float)
model.storage['seeded'] = seeded
return np.concatenate([seeded, model.history(self.attendance)])
def forecast(self, agent: PredictorPoolView, model: ModelView) -> None:
"""Forecast next week's attendance with every predictor of the pool.
Args:
agent:
The pool.
model:
The model.
"""
agent.forecasts = forecasts(self._history(model))
@staticmethod
def error(agent: PredictorPoolView, model: ModelView, index: int) -> float:
"""Return the absolute error of a predictor last week, or its error so far before the first attendance.
Args:
agent:
The pool.
model:
The model view of the rule that reads the error.
index:
The index of the predictor in the pool.
Returns:
error:
The absolute error of the forecast of the predictor.
"""
attendance = model.history(PredictorPoolElement.attendance)
if not len(attendance):
return float(np.ravel(agent.errors)[index])
return float(abs(np.ravel(agent.forecasts)[index] - attendance[-1]))
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)]