Kolkata Paise Restaurant
The Kolkata Paise Restaurant problem, where diners spread over restaurants that serve one meal each.
- Name:
kolkata_restaurant - Field:
economics/econophysics - Formalism: agent-based model
- References: Chakrabarti et al. 2009; Ghosh et al. 2010; Carmichael and Hadzikadic 2019 ch. 9
Factory
@register_model(
field='economics/econophysics',
tags=('kolkata paise restaurant', 'agents', 'resource allocation', 'minority game'),
references=('Chakrabarti et al. 2009', 'Ghosh et al. 2010', 'Carmichael and Hadzikadic 2019 ch. 9'),
defaults={'until': 100},
)
def kolkata_restaurant(
n_agents: int = 100,
*,
n_restaurants: int | None = None,
strategy: str = 'random',
random_state: int | np.random.Generator | None = 0,
) -> System:
"""Return the Kolkata Paise Restaurant problem, where diners spread over restaurants that serve one meal each.
With random choice the expected utilization of `N` restaurants by `N` diners is `1 - 1/e`, about 0.632, and with
the stochastic crowd avoiding strategy it settles near 0.8.
Args:
n_agents:
The number of diners.
n_restaurants:
The number of restaurants, or `None` for as many as diners.
strategy:
`'random'` or `'crowd_avoiding'`.
random_state:
The seed or generator of the first choices.
Returns:
system:
The model, with the group `diners`, the choice `choice` of a restaurant every evening, uniform or by crowd
avoidance, and the interaction `restaurants`, which reports the fraction of
restaurants that serve a meal `restaurants.utilization`.
Raises:
ValueError:
The strategy is unknown.
Examples:
>>> from skcomplex.models.economics.econophysics import kolkata_restaurant
>>> type(kolkata_restaurant(10, strategy='crowd_avoiding').interactions[0]).__name__
'CrowdAvoidanceInteraction'
"""
if strategy not in STRATEGIES:
error_msg = f'Unknown strategy {strategy!r}. Use one of {STRATEGIES}.'
raise ValueError(error_msg)
n_restaurants = n_agents if n_restaurants is None else n_restaurants
options = [float(restaurant) for restaurant in range(n_restaurants)]
choices = np.random.default_rng(random_state).integers(n_restaurants, size=n_agents).astype(float)
diners = GroupElement('diners', element=DinerElement, size=n_agents, values={'restaurant': choices})
choice = (
CrowdAvoidanceInteraction('choice', target='diners', options=options)
if strategy == 'crowd_avoiding'
else ChoiceInteraction(
'choice',
target='diners',
state='restaurant',
options=options,
utility=lambda agent, model, option: 0.0,
intensity=0.0,
form='agent',
)
)
restaurants = RestaurantsInteraction('restaurants', target='diners', n_restaurants=n_restaurants)
return System(elements=[diners], interactions=[choice, restaurants])
Components
class DinerElement(Element):
"""Diner of the Kolkata Paise Restaurant problem, who goes to one restaurant every evening.
Args:
label:
The label of the agent or group.
visual:
How the element is drawn, or `None` for `default_visual_`.
restaurant:
The restaurant chosen last, from 0 to the number of restaurants minus one.
crowd:
The number of diners at that restaurant, 0 before the first evening.
served:
Whether the diner got a meal, 1 or 0.
Examples:
>>> from skcomplex.core.elements import GroupElement
>>> from skcomplex.models.economics.econophysics import DinerElement
>>> GroupElement('diners', element=DinerElement, size=10).size
10
"""
def __init__(
self,
label: str,
visual: VisualSpec | None = None,
*,
restaurant: Annotated[float, StateSpec(domain=NN)] = 0,
crowd: Annotated[float, StateSpec(domain=NN)] = 0,
served: Annotated[float, StateSpec(domain=FiniteEnumeratedSet([0, 1]))] = 0.0,
) -> None:
super().__init__(label, visual=visual)
self.restaurant = restaurant
self.crowd = crowd
self.served = served
@classmethod
def _get_view(cls) -> type[DinerView]:
"""Return the view of the agents of the element."""
return DinerView
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 RestaurantsInteraction(GroupingInteraction):
"""Restaurants that each serve one meal every evening, to one of the diners who come chosen at random.
The diners are grouped by their restaurant. Every diner learns the crowd at its restaurant, and the first of the
group in a random order is served. The interaction reports the fraction of restaurants that serve a meal.
Args:
label:
The label of the interaction.
target:
The path of the group of diners.
n_restaurants:
The number of restaurants.
by:
The state of the diners that holds their restaurant.
phase:
The phase the restaurants serve in, or `None` for the default phase.
Examples:
>>> from skcomplex.models.economics.econophysics import RestaurantsInteraction
>>> RestaurantsInteraction('restaurants', target='diners', n_restaurants=5).check().groups
'state'
"""
reports: ClassVar[tuple[str, ...]] = ('utilization',)
def __init__(
self,
label: str,
*,
target: str,
n_restaurants: int = 100,
by: str = 'restaurant',
phase: str | None = None,
) -> None:
super().__init__(label, target=target, groups='state', by=by, phase=phase)
self.n_restaurants = n_restaurants
def act_on(self, members: list[AgentViewProtocol], model: ModelViewProtocol) -> None:
"""Tell every diner of a restaurant its crowd and serve the first diner.
Args:
members:
The diners at one restaurant, in random order.
model:
The model view of the restaurants.
"""
diners = [member for member in members if isinstance(member, DinerView)]
for position, member in enumerate(diners):
member.crowd = float(len(diners))
member.served = float(position == 0)
def report(self, model: ModelViewProtocol) -> None:
"""Report the fraction of restaurants with at least one diner.
Args:
model:
The model view of the restaurants.
"""
visited = {float(getattr(agent, str(self.by))) for agent in model.agents('target')}
model.report('utilization', len(visited) / self.n_restaurants)
def _get_laws(self, namespace: SimpleNamespace) -> list[BaseLaw]:
"""Return the rule that serves the diners and the rule that reports the utilization.
Args:
namespace:
The namespace of the symbols the laws can use.
Returns:
laws:
The serving and the reporting rules.
"""
return [*super()._get_laws(namespace), RuleLaw(self.report, phase=self.phase)]