Econophysics

Econophysics: wealth exchange between agents.

CoalitionBarInteraction(label, *, rule=None, phase=None, patrons, strategies=(), capacity=60.0, smoothing=0.1, coalitions=True)

Bases: FeedbackInteraction[SystemView]

El Farol bar of Rand and Wilensky whose patrons share strategies within coalitions that split and merge.

A strategy s in [-1, 1]^(m + 1) predicts the attendance n s_1 + sum_j s_(j + 1) h_j from the last m attendances h, and its error is the total absolute error of its predictions over the last m nights. Every night every coalition uses the strategy of the lowest error among the strategies of its members, and every member goes when the prediction is at most the capacity. Every member's return moves by smoothing towards 1 after a good night and towards 0 after a bad one. With coalitions, after every night (Collins 2019, Section 3.2), every coalition in random order that attended an overcrowded night within the memory, when the smallest excess x of those nights is below its size N, splits off a random subgroup of at most N - x members that holds the owner of its best strategy. Then, at most once per coalition and night, two coalitions that attended an uncrowded night merge unless their joint size would have overcrowded an uncrowded night either attended within the memory. A merged coalition keeps the longer history.

Parameters:

Name	Type	Description	Default
label	str	The label of the bar.	required
rule	None	Unused, since the bar overrides act.	None
phase	str | None	The phase the bar acts in, or None for the default phase.	None
patrons	Annotated[str, RoleSpec()]	The path of the group of patrons.	required
strategies	tuple[tuple[tuple[float, ...], ...], ...]	The strategies of every patron, strategies[i][k] the m + 1 weights of strategy k of patron i.	()
capacity	Annotated[float, ParameterSpec(domain=open(0, oo))]	The attendance above which the bar is overcrowded.	60.0
smoothing	Annotated[float, ParameterSpec(domain=closed(0, 1))]	The weight lambda of the last night in the return.	0.1
coalitions	bool	Whether patrons form coalitions.	True

Examples:

>>> from skcomplex.systems.economics.econophysics import CoalitionBarInteraction
>>> CoalitionBarInteraction('bar', patrons='patrons').reports[:2]
('attendance', 'coalitions')

Source code in src/skcomplex/systems/economics/econophysics/_coalitions.py

def __init__(
    self,
    label: str,
    *,
    rule: None = None,
    phase: str | None = None,
    patrons: Annotated[str, RoleSpec()],
    strategies: tuple[tuple[tuple[float, ...], ...], ...] = (),
    capacity: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 60.0,
    smoothing: Annotated[float, ParameterSpec(domain=RealSet.closed(0, 1))] = 0.1,
    coalitions: bool = True,
) -> None:
    super().__init__(label, rule=rule, phase=phase)
    self.patrons = patrons
    self.strategies = strategies
    self.capacity = capacity
    self.smoothing = smoothing
    self.coalitions = coalitions

act(system)

Decide the night, score the patrons, and let the coalitions split and merge.

Parameters:

Name	Type	Description	Default
system	SystemView	The system view of the bar.	required

Source code in src/skcomplex/systems/economics/econophysics/_coalitions.py

def act(self, system: SystemView) -> None:
    """Decide the night, score the patrons, and let the coalitions split and merge.

    Args:
        system:
            The system view of the bar.
    """
    patrons = sorted(system.agents('patrons', CoalitionPatronView), key=lambda patron: patron.id)
    strategies = np.asarray(self.strategies, dtype=float)
    memory, size, capacity = strategies.shape[-1] - 1, len(patrons), system.parameter('capacity')
    stored = system.storage.get('history')
    history = np.asarray(stored) if isinstance(stored, np.ndarray) else system.rng.integers(0, size + 1, 2 * memory)
    records = system.storage.get('records')
    records = records if isinstance(records, dict) else {}
    labels = np.array([int(patron.coalition) for patron in patrons])
    if not self.coalitions:
        labels = np.arange(size)
    errors = self._errors(strategies, history.astype(float))
    best: dict[int, int] = {}
    attends = np.zeros(size, dtype=bool)
    for label in np.unique(labels):
        members = np.flatnonzero(labels == label)
        owner, strategy = np.unravel_index(np.argmin(errors[members]), errors[members].shape)
        best[int(label)] = int(members[owner])
        weights = strategies[members[owner], strategy]
        attends[members] = size * weights[0] + weights[1:] @ history[:memory] <= capacity
    attendance = float(attends.sum())
    crowded = attendance > capacity
    good = np.where(attends, not crowded, crowded)
    smoothing = system.parameter('smoothing')
    scores = np.array([patron.score for patron in patrons]) * (1 - smoothing) + smoothing * good
    for label in np.unique(labels[attends]):
        records.setdefault(int(label), []).append((attendance, float(np.sum(labels == label))))
    for label in list(records):
        records[label] = records[label][-memory:]
    splits = joins = 0
    if self.coalitions:
        splits = self._split(system, labels, best, records)
        joins = 0 if crowded else self._join(system, labels, attends, records)
    system.storage.update(history=np.concatenate(([attendance], history[:-1])), records=records)
    for patron, label, went, score in zip(patrons, labels, attends, scores, strict=True):
        patron.coalition, patron.attends, patron.score = float(label), float(went), float(score)
    system.report('attendance', attendance)
    system.report('coalitions', float(len(np.unique(labels))))
    system.report('mean_score', float(scores.mean()))
    system.report('splits', float(splits))
    system.report('joins', float(joins))

CoalitionPatronElement(label, visual=VisualSpec('agents', {'color_by': 'coalition'}), *, coalition=0, attends=0.0, score=0.0)

Bases: Element

Patron of the El Farol bar who belongs to a coalition and keeps a smoothed return of its nights.

Parameters:

Name	Type	Description	Default
label	str	The label of the agent or group.	required
visual	VisualSpec | None	How the element is drawn, or None for default_visual_.	VisualSpec('agents', {'color_by': 'coalition'})
coalition	Annotated[float, StateSpec(domain=NN)]	The label of the coalition of the patron.	0
attends	Annotated[float, StateSpec(domain=closed(0, 1))]	1 when the patron went to the bar on the last night, 0 otherwise.	0.0
score	Annotated[float, StateSpec(domain=closed(0, 1))]	The exponentially smoothed share of good nights, a night at the bar that was not overcrowded or a night away from an overcrowded bar.	0.0

Examples:

>>> from skcomplex.core.components.elements import GroupElement
>>> from skcomplex.systems.economics.econophysics import CoalitionPatronElement
>>> GroupElement('patrons', element=CoalitionPatronElement, size=3).size
3

Source code in src/skcomplex/systems/economics/econophysics/_coalitions.py

def __init__(
    self,
    label: str,
    visual: VisualSpec | None = VisualSpec('agents', {'color_by': 'coalition'}),
    *,
    coalition: Annotated[float, StateSpec(domain=NN)] = 0,
    attends: Annotated[float, StateSpec(domain=RealSet.closed(0, 1))] = 0.0,
    score: Annotated[float, StateSpec(domain=RealSet.closed(0, 1))] = 0.0,
) -> None:
    super().__init__(label, visual=visual)
    self.coalition = coalition
    self.attends = attends
    self.score = score

CrowdAvoidanceInteraction(label, *, target, options, state='restaurant', intensity=1.0, activity=1.0, phase=None)

Bases: ChoiceInteraction

Stochastic crowd avoidance of diners, who leave crowded restaurants.

A diner returns to last evening's restaurant with probability 1 / n, with n the crowd it found there, and otherwise moves to one of the other restaurants at random, which raises the utilization to about 0.8. It is a Metropolis choice of intensity 1 where every other restaurant has utility 0 and staying has the utility -log(1 - 1 / n), so a proposed move is accepted with probability 1 - 1 / n.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction.	required
target	str	The path of the group of diners.	required
options	Sequence[float]	The restaurants, from 0 to their number minus one.	required
state	str	The state of the diners that holds their restaurant.	'restaurant'
intensity	float	The intensity of choice, 1 for the stay probability 1 / n.	1.0
activity	float	The probability that a diner revises its choice in an evening.	1.0
phase	str | None	The phase the choice runs in, or None for the default phase.	None

Examples:

>>> from skcomplex.systems.economics.econophysics import CrowdAvoidanceInteraction
>>> CrowdAvoidanceInteraction('choice', target='diners', options=(0.0, 1.0)).rule
'metropolis'

Source code in src/skcomplex/systems/economics/econophysics/_kolkata.py

def __init__(
    self,
    label: str,
    *,
    target: str,
    options: Sequence[float],
    state: str = 'restaurant',
    intensity: float = 1.0,
    activity: float = 1.0,
    phase: str | None = None,
) -> None:
    super().__init__(
        label,
        target=target,
        state=state,
        options=options,
        utility=self._get_utility,
        rule='metropolis',
        intensity=intensity,
        activity=activity,
        form='agent',
        phase=phase,
    )

DebtExchangeInteraction(label, *, source, target, debt=0.0, amount=1.0, phase=None)

Bases: TransferInteraction

Random pairwise trades of Dragulescu and Yakovenko of a fixed amount, allowing debt down to a limit.

In every trade an agent pays a fixed amount to another agent drawn at random, unless its wealth would fall below minus the debt limit. Every step, every agent pays once, in random order. The stationary wealth distribution is the exponential P(w) = exp(-(w + d) / T) / T shifted to the debt limit d, with temperature T = <w> + d.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction.	required
source	str	The path of the group of traders.	required
target	str	The path of the group of traders, the source again.	required
debt	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The debt limit d, the largest debt an agent can have.	0.0
amount	Annotated[float, ParameterSpec(domain=open(0, oo))]	The amount paid in every trade.	1.0
phase	str | None	The phase the trades run in, or None for the default phase.	None

Examples:

>>> from skcomplex.systems.economics.econophysics import DebtExchangeInteraction
>>> DebtExchangeInteraction('trades', source='traders', target='traders', debt=5.0).debt
5.0

Source code in src/skcomplex/systems/economics/econophysics/_kinetic.py

def __init__(
    self,
    label: str,
    *,
    source: str,
    target: str,
    debt: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
    amount: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 1.0,
    phase: str | None = None,
) -> None:
    super().__init__(
        label,
        source=source,
        target=target,
        state='wealth',
        amount=amount,
        pairing='random',
        guard=self._is_solvent,
        form='agent',
        phase=phase,
    )
    self.debt = debt

DinerElement(label, visual=None, *, restaurant=0, crowd=0, served=0.0)

Bases: Element

Diner of the Kolkata Paise Restaurant problem, who goes to one restaurant every evening.

Parameters:

Name	Type	Description	Default
label	str	The label of the agent or group.	required
visual	VisualSpec | None	How the element is drawn, or None for default_visual_.	None
restaurant	Annotated[float, StateSpec(domain=NN)]	The restaurant chosen last, from 0 to the number of restaurants minus one.	0
crowd	Annotated[float, StateSpec(domain=NN)]	The number of diners at that restaurant, 0 before the first evening.	0
served	Annotated[float, StateSpec(domain=FiniteEnumeratedSet([0, 1]))]	Whether the diner got a meal, 1 or 0.	0.0

Examples:

>>> from skcomplex.core.components.elements import GroupElement
>>> from skcomplex.systems.economics.econophysics import DinerElement
>>> GroupElement('diners', element=DinerElement, size=10).size
10

Source code in src/skcomplex/systems/economics/econophysics/_kolkata.py

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

MinorityGameInteraction(label, *, players, memory=3, phase=None)

Bases: 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.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction.	required
players	Annotated[str, RoleSpec()]	The path of the group of agents.	required
memory	int	The memory m, the number of past winning sides the strategies read.	3
phase	str | None	The phase the rounds run in, or None for the default phase.	None

Examples:

>>> from skcomplex.systems.economics.econophysics import MinorityGameInteraction
>>> MinorityGameInteraction('game', players='players', memory=4).memory
4

Source code in src/skcomplex/systems/economics/econophysics/_minority.py

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

payoff(agent, system, option)

Return the virtual payoff of a strategy of an agent in the last round, +1 if it played the winning side.

Parameters:

Name	Type	Description	Default
agent	MinorityPlayerView	The agent.	required
system	SystemView	The system view of the rule that reads the payoff.	required
option	int	The strategy, 0 for the first and 1 for the second.	required

Returns:

Name	Type	Description
payoff	float	+1 when the strategy would have played the winning side, or -1.

Source code in src/skcomplex/systems/economics/econophysics/_minority.py

def payoff(self, agent: MinorityPlayerView, system: SystemView, 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.

        system:
            The system 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(system.history(f'{self.label}.history')[-1])
    winner = -1 if system.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)

play(system)

Play one round, pay the agents, and update and report the history.

Parameters:

Name	Type	Description	Default
system	SystemView	The system.	required

Source code in src/skcomplex/systems/economics/econophysics/_minority.py

def play(self, system: SystemView) -> None:
    """Play one round, pay the agents, and update and report the history.

    Args:
        system:
            The system.
    """
    agents = system.agents('players', MinorityPlayerView)
    history = self._state(system).history
    for agent in agents:
        code = agent.second if agent.strategy else agent.first
        agent.action = float(2 * int(self._table(code, system)[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(system)
    system.storage['game'] = replace(state, history=(2 * history + (winner > 0)) % 2**self.memory)
    system.report('attendance', attendance)
    system.report('history', float(history))

MinorityPlayerElement(label, visual=None, *, first=0, second=1, first_score=0.0, second_score=0.0, strategy=0.0, action=0.0, wealth=0.0)

Bases: Element

Agent of the minority game, who holds two strategies and plays the one it chose.

Parameters:

Name	Type	Description	Default
label	str	The label of the agent or group.	required
visual	VisualSpec | None	How the element is drawn, or None for default_visual_.	None
first	Annotated[float, StateSpec(domain=NN)]	The code of the first strategy.	0
second	Annotated[float, StateSpec(domain=NN)]	The code of the second strategy.	1
first_score	Annotated[float, StateSpec()]	The virtual score of the first strategy, the number of times it would have been in the minority.	0.0
second_score	Annotated[float, StateSpec()]	The virtual score of the second strategy.	0.0
strategy	Annotated[float, StateSpec(domain=FiniteEnumeratedSet([0, 1]))]	The strategy played, 0 for the first and 1 for the second.	0.0
action	Annotated[float, StateSpec(domain=ZZ)]	The last action, +1 or -1.	0.0
wealth	Annotated[float, StateSpec()]	The net number of times the agent was in the minority.	0.0

Examples:

>>> from skcomplex.core.components.elements import GroupElement
>>> from skcomplex.systems.economics.econophysics import MinorityPlayerElement
>>> GroupElement('players', element=MinorityPlayerElement, size=11, values={'first': 3, 'second': 5}).size
11

Source code in src/skcomplex/systems/economics/econophysics/_minority.py

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

PatronElement(label, visual=None, *, predictors=1, forecast=0.0, attends=0.0, capacity=60.0)

Bases: 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.

Parameters:

Name	Type	Description	Default
label	str	The label of the agent or group.	required
visual	VisualSpec | None	How the element is drawn, or None for default_visual_.	None
predictors	Annotated[float, StateSpec(domain=NN)]	The bit mask of the predictors of the pool that the patron holds.	1
forecast	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The last forecast of the patron's most accurate predictor.	0.0
attends	Annotated[float, StateSpec(domain=FiniteEnumeratedSet([0, 1]))]	Whether the patron went to the bar this week, 1 or 0.	0.0
capacity	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The attendance above which the bar is too crowded to enjoy.	60.0

Examples:

>>> from skcomplex.core.components.elements import GroupElement
>>> from skcomplex.systems.economics.econophysics import PatronElement
>>> GroupElement('patrons', element=PatronElement, size=100, values={'predictors': 7}).size
100

Source code in src/skcomplex/systems/economics/econophysics/_el_farol.py

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

predict(agent, system)

Forecast the attendance with the most accurate predictor held.

Parameters:

Name	Type	Description	Default
agent	PatronView	The patron.	required
system	SystemView	The system.	required

Source code in src/skcomplex/systems/economics/econophysics/_el_farol.py

def predict(self, agent: PatronView, system: SystemView) -> None:
    """Forecast the attendance with the most accurate predictor held.

    Args:
        agent:
            The patron.

        system:
            The system.
    """
    pool = system.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]))]])

PredictorPoolElement(label, visual=None, *, forecasts=0.0, errors=0.0)

Bases: 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.

Parameters:

Name	Type	Description	Default
label	str	The label of the agent or group.	required
visual	VisualSpec | None	How the element is drawn, or None for default_visual_.	None
forecasts	Annotated[ArrayLike, StateSpec(vector=True)]	The forecast of next week's attendance by every predictor of the pool.	0.0
errors	Annotated[ArrayLike, StateSpec(vector=True)]	The discounted mean absolute error of every predictor of the pool.	0.0

Examples:

>>> from skcomplex.core.components.elements import GroupElement
>>> from skcomplex.systems.economics.econophysics import PredictorPoolElement
>>> GroupElement('pool', element=PredictorPoolElement, size=1).size
1

Source code in src/skcomplex/systems/economics/econophysics/_el_farol.py

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

error(agent, system, index) staticmethod

Return the absolute error of a predictor last week, or its error so far before the first attendance.

Parameters:

Name	Type	Description	Default
agent	PredictorPoolView	The pool.	required
system	SystemView	The system view of the rule that reads the error.	required
index	int	The index of the predictor in the pool.	required

Returns:

Name	Type	Description
error	float	The absolute error of the forecast of the predictor.

Source code in src/skcomplex/systems/economics/econophysics/_el_farol.py

@staticmethod
def error(agent: PredictorPoolView, system: SystemView, 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.

        system:
            The system 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 = system.history(PredictorPoolElement.attendance)
    if not len(attendance):
        return float(np.ravel(agent.errors)[index])
    return float(abs(np.ravel(agent.forecasts)[index] - attendance[-1]))

forecast(agent, system)

Forecast next week's attendance with every predictor of the pool.

Parameters:

Name	Type	Description	Default
agent	PredictorPoolView	The pool.	required
system	SystemView	The system.	required

Source code in src/skcomplex/systems/economics/econophysics/_el_farol.py

def forecast(self, agent: PredictorPoolView, system: SystemView) -> None:
    """Forecast next week's attendance with every predictor of the pool.

    Args:
        agent:
            The pool.

        system:
            The system.
    """
    agent.forecasts = forecasts(self._history(system))

RestaurantsInteraction(label, *, target, n_restaurants=100, by='restaurant', phase=None)

Bases: 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.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction.	required
target	str	The path of the group of diners.	required
n_restaurants	int	The number of restaurants.	100
by	str	The state of the diners that holds their restaurant.	'restaurant'
phase	str | None	The phase the restaurants serve in, or None for the default phase.	None

Examples:

>>> from skcomplex.systems.economics.econophysics import RestaurantsInteraction
>>> RestaurantsInteraction('restaurants', target='diners', n_restaurants=5).groups
'state'

Source code in src/skcomplex/systems/economics/econophysics/_kolkata.py

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

act_on(members, system)

Tell every diner of a restaurant its crowd and serve the first diner.

Parameters:

Name	Type	Description	Default
members	list[AgentViewProtocol]	The diners at one restaurant, in random order.	required
system	SystemViewProtocol	The system view of the restaurants.	required

Source code in src/skcomplex/systems/economics/econophysics/_kolkata.py

def act_on(self, members: list[AgentViewProtocol], system: SystemViewProtocol) -> None:
    """Tell every diner of a restaurant its crowd and serve the first diner.

    Args:
        members:
            The diners at one restaurant, in random order.

        system:
            The system 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)

report(system)

Report the fraction of restaurants with at least one diner.

Parameters:

Name	Type	Description	Default
system	SystemViewProtocol	The system view of the restaurants.	required

Source code in src/skcomplex/systems/economics/econophysics/_kolkata.py

def report(self, system: SystemViewProtocol) -> None:
    """Report the fraction of restaurants with at least one diner.

    Args:
        system:
            The system view of the restaurants.
    """
    visited = {float(getattr(agent, str(self.by))) for agent in system.agents('target')}
    system.report('utilization', len(visited) / self.n_restaurants)

SavingExchangeInteraction(label, *, source, target, saving=0.0, phase=None)

Bases: TransferInteraction

Random pairwise trades of Chakraborti and Chakrabarti, where every agent saves a fraction of its wealth.

A trade between agents i and j keeps total wealth and gives w_i' = lambda w_i + eps (1 - lambda) (w_i + w_j) and w_j' = lambda w_j + (1 - eps) (1 - lambda) (w_i + w_j) with eps uniform on [0, 1], so i transfers (1 - lambda) (w_i - eps (w_i + w_j)) to j, an amount of either sign. Every step, every agent trades once with another agent drawn at random, in random order. Without saving the stationary wealth distribution is exponential, and with saving propensity lambda it is close to a Gamma distribution of shape n = 1 + 3 lambda / (1 - lambda).

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction.	required
source	str	The path of the group of traders.	required
target	str	The path of the group of traders, the source again.	required
saving	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The saving propensity lambda, the fraction of its wealth that an agent keeps out of a trade.	0.0
phase	str | None	The phase the trades run in, or None for the default phase.	None

Examples:

>>> from skcomplex.systems.economics.econophysics import SavingExchangeInteraction
>>> SavingExchangeInteraction('trades', source='traders', target='traders').saving
0.0

Source code in src/skcomplex/systems/economics/econophysics/_kinetic.py

def __init__(
    self,
    label: str,
    *,
    source: str,
    target: str,
    saving: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
    phase: str | None = None,
) -> None:
    super().__init__(
        label,
        source=source,
        target=target,
        state='wealth',
        amount=self._get_amount,
        pairing='random',
        form='agent',
        phase=phase,
    )
    self.saving = saving

SavingTraderElement(label, visual=None, *, wealth=1.0)

Bases: Element

Agent of the kinetic wealth exchange systems, whose wealth changes only in pairwise trades.

Parameters:

Name	Type	Description	Default
label	str	The label of the agent or group.	required
visual	VisualSpec | None	How the element is drawn, or None for default_visual_.	None
wealth	Annotated[float, StateSpec()]	The initial wealth, negative down to the debt limit.	1.0

Examples:

>>> from skcomplex.core.components.elements import GroupElement
>>> from skcomplex.systems.economics.econophysics import SavingTraderElement
>>> group = GroupElement('traders', element=SavingTraderElement, size=10, values={'wealth': 2.0})
>>> group.label, group.element.__name__, group.size
('traders', 'SavingTraderElement', 10)

Source code in src/skcomplex/systems/economics/econophysics/_kinetic.py

def __init__(
    self,
    label: str,
    visual: VisualSpec | None = None,
    *,
    wealth: Annotated[float, StateSpec()] = 1.0,
) -> None:
    super().__init__(label, visual=visual)
    self.wealth = wealth

TraderElement(label, visual=None, *, wealth=1.0)

Bases: Element

Agent of the Boltzmann wealth system, whose wealth in whole units changes only by gifts of one unit.

Parameters:

Name	Type	Description	Default
label	str	The label of the agent or group.	required
visual	VisualSpec | None	How the element is drawn, or None for default_visual_.	None
wealth	Annotated[float, StateSpec(domain=NN)]	The initial wealth.	1.0

Examples:

>>> from skcomplex.systems.economics.econophysics import TraderElement
>>> group = GroupElement('traders', element=TraderElement, size=100)
>>> group.label, group.element.__name__, group.size
('traders', 'TraderElement', 100)

Source code in src/skcomplex/systems/economics/econophysics/_exchange.py

def __init__(
    self,
    label: str,
    visual: VisualSpec | None = None,
    *,
    wealth: Annotated[float, StateSpec(domain=NN)] = 1.0,
) -> None:
    super().__init__(label, visual=visual)
    self.wealth = wealth

WealthExchangeInteraction(label, *, candidate, field, classes=9, eta=1.0, cooperation=0.2, competition=0.3)

Bases: EncounterInteraction

Conservative encounters between wealth classes, with cooperation and competition.

A candidate of class h meets a field particle of class k at the rate eta and moves by the transition table of exchange_table.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction.	required
candidate	str	The path of the distribution over wealth classes whose particles move.	required
field	str	The path of the distribution over wealth classes they meet, the candidate again.	required
classes	Annotated[int, ParameterSpec(domain=open(0, oo))]	The number of wealth classes.	9
eta	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The encounter rate.	1.0
cooperation	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The probability of the move that narrows the gap between the classes of an encounter.	0.2
competition	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The probability of the move that widens it.	0.3

Examples:

>>> from skcomplex.systems.economics.econophysics import WealthExchangeInteraction
>>> WealthExchangeInteraction('exchange', candidate='wealth', field='wealth').competition
0.3

Source code in src/skcomplex/systems/economics/econophysics/_welfare.py

def __init__(
    self,
    label: str,
    *,
    candidate: str,
    field: str,
    classes: Annotated[int, ParameterSpec(domain=RealSet.open(0, oo))] = 9,
    eta: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
    cooperation: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.2,
    competition: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.3,
) -> None:
    super().__init__(label, candidate=candidate, field=field, rate=self._get_rate, kernel=self._get_kernel)
    self.classes = classes
    self.eta = eta
    self.cooperation = cooperation
    self.competition = competition

WelfarePolicyInteraction(label, *, candidate, field, classes=9, tax=0.1)

Bases: EncounterInteraction

Welfare policy that taxes the richest classes and helps the poorest, a micro-scale external action.

The policy is an encounter whose field role binds a group with a fixed density, so it acts on the transitions of the population at the rate tax times that density, by the pattern and the table of policy_table.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction.	required
candidate	str	The path of the distribution over wealth classes.	required
field	str	The path of the group of one member whose fixed density is the strength of the policy.	required
classes	Annotated[int, ParameterSpec(domain=open(0, oo))]	The number of wealth classes.	9
tax	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The rate at which the policy moves the richest classes down and the poorest up.	0.1

Examples:

>>> from skcomplex.systems.economics.econophysics import WelfarePolicyInteraction
>>> WelfarePolicyInteraction('welfare', candidate='wealth', field='state').tax
0.1

Source code in src/skcomplex/systems/economics/econophysics/_welfare.py

def __init__(
    self,
    label: str,
    *,
    candidate: str,
    field: str,
    classes: Annotated[int, ParameterSpec(domain=RealSet.open(0, oo))] = 9,
    tax: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.1,
) -> None:
    super().__init__(label, candidate=candidate, field=field, rate=self._get_rate, kernel=self._get_kernel)
    self.classes = classes
    self.tax = tax

boltzmann_wealth(n_agents=100)

Return a Boltzmann wealth exchange system where every agent starts with one unit.

Every step, every agent with any wealth gives one unit to another agent drawn at random, in random order.

Parameters:

Name	Type	Description	Default
n_agents	int	The number of agents.	100

Returns:

Name	Type	Description
system	System	The system, with the group traders and the transfer gifts.

Examples:

>>> from skcomplex.systems.economics.econophysics import boltzmann_wealth
>>> boltzmann_wealth(10).elements[0].prepare().size_
10

Source code in src/skcomplex/systems/economics/econophysics/_exchange.py

@register_system(
    field='economics/econophysics',
    tags=('wealth', 'agents', 'exchange'),
    references=('Dragulescu and Yakovenko 2000',),
    defaults={'until': 100},
)
def boltzmann_wealth(n_agents: int = 100) -> System:
    """Return a Boltzmann wealth exchange system where every agent starts with one unit.

    Every step, every agent with any wealth gives one unit to another agent drawn at random, in random order.

    Args:
        n_agents:
            The number of agents.

    Returns:
        system:
            The system, with the group `traders` and the transfer `gifts`.

    Examples:
        >>> from skcomplex.systems.economics.econophysics import boltzmann_wealth
        >>> boltzmann_wealth(10).elements[0].prepare().size_
        10
    """
    gifts = TransferInteraction(
        'gifts',
        source='traders',
        target='traders',
        state='wealth',
        amount=1.0,
        pairing='random',
        guard=lambda giver, receiver, system: giver.wealth > 0,
        form='agent',
    )
    return System(elements=[GroupElement('traders', element=TraderElement, size=n_agents)], interactions=[gifts])

el_farol(n_agents=100, *, capacity=60.0, n_predictors=6, discount=0.9, random_state=0)

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.

Parameters:

Name	Type	Description	Default
n_agents	int	The number of patrons.	100
capacity	float	The attendance above which the bar is too crowded to enjoy.	60.0
n_predictors	int	The number of predictors every patron holds.	6
discount	float	The weight of past errors in the accuracy of a predictor.	0.9
random_state	int | Generator | None	The seed or generator of the predictors that every patron holds.	0

Returns:

Name	Type	Description
system	System	The system, 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.systems.economics.econophysics import el_farol
>>> el_farol(10).elements[0].prepare().size_
10

Source code in src/skcomplex/systems/economics/econophysics/_el_farol.py

@register_system(
    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 system, 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.systems.economics.econophysics import el_farol
        >>> el_farol(10).elements[0].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, system, 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')],
            ),
        ),
    )

el_farol_coalitions(n_agents=100, *, capacity=60.0, memory=5, n_strategies=10, smoothing=0.1, coalitions=True, random_state=0)

Return Collins' El Farol bar with strategic group formation, whose coalitions share their best strategy.

Patrons of Rand and Wilensky's El Farol model predict the attendance with linear strategies of the last memory attendances and go when they predict room. In Collins' version (Carmichael et al. 2019, chapter 9) patrons form coalitions that use the best strategy of all their members, so a larger coalition has better predictions but adds more to the crowd. Coalitions split off the members with the best strategy after overcrowded nights, the core condition that no subgroup gains by leaving, and coalitions that attended together merge when it would not have crowded an earlier good night. Without coalitions the mean attendance stays near the capacity and patrons enjoy about half their nights, while coalitions make the attendance swing and lower the returns of all, because the patrons are boundedly rational. The return of a patron is exponentially smoothed with the weight smoothing, which the source leaves open.

Parameters:

Name	Type	Description	Default
n_agents	int	The number of patrons.	100
capacity	float	The attendance above which the bar is overcrowded.	60.0
memory	int	The number of past attendances a strategy reads.	5
n_strategies	int	The number of strategies of every patron.	10
smoothing	float	The weight of the last night in the return.	0.1
coalitions	bool	Whether patrons form coalitions.	True
random_state	int | Generator | None	The seed or generator of the strategies.	0

Returns:

Name	Type	Description
system	System	The system, with the group patrons and the bar bar, whose reporters are bar.attendance, the number of coalitions bar.coalitions, the mean return bar.mean_score, and the splits bar.splits and joins bar.joins of the night.

Examples:

>>> from skcomplex.systems.economics.econophysics import el_farol_coalitions
>>> el_farol_coalitions(10).elements[0].prepare().size_
10

Source code in src/skcomplex/systems/economics/econophysics/_coalitions.py

@register_system(
    field='economics/econophysics',
    tags=('el farol', 'agents', 'coalitions', 'group formation', 'cooperative game theory', 'bounded rationality'),
    references=('Carmichael et al. 2019 ch. 9',),
    defaults={'until': 500},
)
def el_farol_coalitions(
    n_agents: int = 100,
    *,
    capacity: float = 60.0,
    memory: int = 5,
    n_strategies: int = 10,
    smoothing: float = 0.1,
    coalitions: bool = True,
    random_state: int | np.random.Generator | None = 0,
) -> System:
    """Return Collins' El Farol bar with strategic group formation, whose coalitions share their best strategy.

    Patrons of Rand and Wilensky's El Farol model predict the attendance with linear strategies of the last `memory`
    attendances and go when they predict room. In Collins' version (Carmichael et al. 2019, chapter 9) patrons form
    coalitions that use the best strategy of all their members, so a larger coalition has better predictions but adds
    more to the crowd. Coalitions split off the members with the best strategy after overcrowded nights, the core
    condition that no subgroup gains by leaving, and coalitions that attended together merge when it would not have
    crowded an earlier good night. Without coalitions the mean attendance stays near the capacity and patrons enjoy
    about half their nights, while coalitions make the attendance swing and lower the returns of all, because the
    patrons are boundedly rational. The return of a patron is exponentially smoothed with the weight `smoothing`,
    which the source leaves open.

    Args:
        n_agents:
            The number of patrons.

        capacity:
            The attendance above which the bar is overcrowded.

        memory:
            The number of past attendances a strategy reads.

        n_strategies:
            The number of strategies of every patron.

        smoothing:
            The weight of the last night in the return.

        coalitions:
            Whether patrons form coalitions.

        random_state:
            The seed or generator of the strategies.

    Returns:
        system:
            The system, with the group `patrons` and the bar `bar`, whose reporters are `bar.attendance`, the number of
            coalitions `bar.coalitions`, the mean return `bar.mean_score`, and the splits `bar.splits` and joins
            `bar.joins` of the night.

    Examples:
        >>> from skcomplex.systems.economics.econophysics import el_farol_coalitions
        >>> el_farol_coalitions(10).elements[0].prepare().size_
        10
    """
    rng = np.random.default_rng(random_state)
    strategies = rng.uniform(-1.0, 1.0, (n_agents, n_strategies, memory + 1))
    patrons = GroupElement(
        'patrons',
        element=CoalitionPatronElement,
        size=n_agents,
        values={'coalition': np.arange(n_agents, dtype=float)},
    )
    bar = CoalitionBarInteraction(
        'bar',
        patrons='patrons',
        strategies=tuple(tuple(map(tuple, rows)) for rows in strategies.tolist()),
        capacity=capacity,
        smoothing=smoothing,
        coalitions=coalitions,
    )
    return System(elements=[patrons], interactions=[bar])

exchange_table(classes, cooperation, competition)

Return the transition table of encounters between wealth classes, A[h, k, i].

A candidate of class h that meets a richer field particle moves one class up with probability cooperation and one class down with probability competition. One that meets a poorer field particle moves one class up with probability competition and one class down with probability cooperation. So competition makes the rich richer and the poor poorer, and cooperation does the opposite. A move past the poorest or the richest class is not made.

Parameters:

Name	Type	Description	Default
classes	int	The number of wealth classes.	required
cooperation	object	The probability of the move that narrows the gap between the two classes, a number or a parameter.	required
competition	object	The probability of the move that widens it, a number or a parameter.	required

Returns:

Name	Type	Description
table	NDArray[object_]	The transition probabilities, which sum to one over the outcomes.

Examples:

>>> from skcomplex.systems.economics.econophysics import exchange_table
>>> exchange_table(3, 0.2, 0.3)[0, 2].astype(float).tolist()
[0.8, 0.2, 0.0]

Source code in src/skcomplex/systems/economics/econophysics/_welfare.py

def exchange_table(classes: int, cooperation: object, competition: object) -> npt.NDArray[np.object_]:
    """Return the transition table of encounters between wealth classes, `A[h, k, i]`.

    A candidate of class `h` that meets a richer field particle moves one class up with probability `cooperation`
    and one class down with probability `competition`. One that meets a poorer field particle moves one class up with
    probability `competition` and one class down with probability `cooperation`. So competition makes the rich richer
    and the poor poorer, and cooperation does the opposite. A move past the poorest or the richest class is not made.

    Args:
        classes:
            The number of wealth classes.

        cooperation:
            The probability of the move that narrows the gap between the two classes, a number or a parameter.

        competition:
            The probability of the move that widens it, a number or a parameter.

    Returns:
        table:
            The transition probabilities, which sum to one over the outcomes.

    Examples:
        >>> from skcomplex.systems.economics.econophysics import exchange_table
        >>> exchange_table(3, 0.2, 0.3)[0, 2].astype(float).tolist()
        [0.8, 0.2, 0.0]
    """
    table = np.zeros((classes, classes, classes), dtype=object)
    for candidate in range(classes):
        for field in range(classes):
            gap = int(np.sign(field - candidate))
            moves = {candidate + gap: cooperation, candidate - gap: competition} if gap else {}
            for outcome, probability in moves.items():
                if 0 <= outcome < classes:
                    table[candidate, field, outcome] += probability
            table[candidate, field, candidate] += 1.0 - table[candidate, field].sum()
    return table

forecasts(history)

Return the forecast of next week's attendance of every predictor of the pool, between 0 and 100.

Parameters:

Name	Type	Description	Default
history	ArrayLike	The past attendances, the last one last, at least ten of them.	required

Returns:

Name	Type	Description
forecasts	Array	One forecast per predictor of the pool.

Examples:

>>> from skcomplex.systems.economics.econophysics import forecasts
>>> forecasts([60.0] * 10)[:3].tolist()
[60.0, 60.0, 60.0]

Source code in src/skcomplex/systems/economics/econophysics/_el_farol.py

def forecasts(history: npt.ArrayLike) -> Array:
    """Return the forecast of next week's attendance of every predictor of the pool, between 0 and 100.

    Args:
        history:
            The past attendances, the last one last, at least ten of them.

    Returns:
        forecasts:
            One forecast per predictor of the pool.

    Examples:
        >>> from skcomplex.systems.economics.econophysics import forecasts
        >>> forecasts([60.0] * 10)[:3].tolist()
        [60.0, 60.0, 60.0]
    """
    values = np.asarray(history, dtype=float)
    predicted = [function(values, **arguments) for function, arguments in PREDICTORS]
    return np.clip(predicted, 0.0, MAX_ATTENDANCE)

gini_coefficient(s, wealth)

Return the Gini coefficient of a distribution over wealth classes, read by every member of its group.

The coefficient is sum_ij f_i f_j |w_i - w_j| / (2 F M), with the density F and the total wealth M.

Parameters:

Name	Type	Description	Default
s	SimpleNamespace	The namespace of one member of the distribution.	required
wealth	NDArray[float64]	The wealth of every class.	required

Returns:

Name	Type	Description
gini	Expression	The coefficient as an expression of reductions.

Source code in src/skcomplex/systems/economics/econophysics/_welfare.py

def gini_coefficient(s: SimpleNamespace, wealth: npt.NDArray[np.float64]) -> Expression:
    """Return the Gini coefficient of a distribution over wealth classes, read by every member of its group.

    The coefficient is `sum_ij f_i f_j |w_i - w_j| / (2 F M)`, with the density `F` and the total wealth `M`.

    Args:
        s:
            The namespace of one member of the distribution.

        wealth:
            The wealth of every class.

    Returns:
        gini:
            The coefficient as an expression of reductions.
    """
    spread = total(s.f, weights=np.abs(wealth[:, None] - wealth[None, :]))
    return total(s.f * spread) / (2 * moment(s.f, 0, wealth) * moment(s.f, 1, wealth))

kinetic_wealth_exchange(n_agents=500, *, saving=0.0, debt=None, wealth=1.0, amount=0.1)

Return a kinetic wealth exchange system with a saving propensity or a debt limit.

Without a debt limit, agents trade random shares of their pooled wealth after saving a fraction saving of it, as in Chakraborti and Chakrabarti. With a debt limit, they pay a fixed amount in every trade and may go into debt down to the limit, as in Dragulescu and Yakovenko.

Parameters:

Name	Type	Description	Default
n_agents	int	The number of agents.	500
saving	float	The saving propensity lambda, between 0 and 1, of the trades without a debt limit.	0.0
debt	float | None	The debt limit of the fixed-amount trades, or None for trades of random shares with saving.	None
wealth	float	The initial wealth of every agent, the mean wealth.	1.0
amount	float	The amount paid in every fixed-amount trade.	0.1

Returns:

Name	Type	Description
system	System	The system, with the group traders, the interaction trades, and the variance of wealth variance.

Raises:

Type	Description
ValueError	Both a saving propensity and a debt limit are given, or the saving propensity is not below 1.

Examples:

>>> from skcomplex.systems.economics.econophysics import kinetic_wealth_exchange
>>> type(kinetic_wealth_exchange(10, debt=2.0).interactions[0]).__name__
'DebtExchangeInteraction'

Source code in src/skcomplex/systems/economics/econophysics/_kinetic.py

@register_system(
    field='economics/econophysics',
    tags=('wealth', 'agents', 'exchange', 'kinetic'),
    references=('Chakraborti and Chakrabarti 2000', 'Dragulescu and Yakovenko 2000', 'Yakovenko and Rosser 2009'),
    defaults={'until': 200},
    observables=lambda system: [ObservableSpec('variance', group='traders', variable='wealth', reduce='var')],
)
def kinetic_wealth_exchange(
    n_agents: int = 500,
    *,
    saving: float = 0.0,
    debt: float | None = None,
    wealth: float = 1.0,
    amount: float = 0.1,
) -> System:
    """Return a kinetic wealth exchange system with a saving propensity or a debt limit.

    Without a debt limit, agents trade random shares of their pooled wealth after saving a fraction `saving` of it,
    as in Chakraborti and Chakrabarti. With a debt limit, they pay a fixed amount in every trade and may go into debt
    down to the limit, as in Dragulescu and Yakovenko.

    Args:
        n_agents:
            The number of agents.

        saving:
            The saving propensity `lambda`, between 0 and 1, of the trades without a debt limit.

        debt:
            The debt limit of the fixed-amount trades, or `None` for trades of random shares with saving.

        wealth:
            The initial wealth of every agent, the mean wealth.

        amount:
            The amount paid in every fixed-amount trade.

    Returns:
        system:
            The system, with the group `traders`, the interaction `trades`, and the variance of wealth `variance`.

    Raises:
        ValueError:
            Both a saving propensity and a debt limit are given, or the saving propensity is not below 1.

    Examples:
        >>> from skcomplex.systems.economics.econophysics import kinetic_wealth_exchange
        >>> type(kinetic_wealth_exchange(10, debt=2.0).interactions[0]).__name__
        'DebtExchangeInteraction'
    """
    if debt is not None and saving:
        error_msg = 'Give either a saving propensity or a debt limit, not both.'
        raise ValueError(error_msg)
    if not 0 <= saving < 1:
        error_msg = f'The saving propensity must lie in [0, 1), got {saving}.'
        raise ValueError(error_msg)
    trades = (
        SavingExchangeInteraction('trades', source='traders', target='traders', saving=saving)
        if debt is None
        else DebtExchangeInteraction('trades', source='traders', target='traders', debt=debt, amount=amount)
    )
    return System(
        elements=[GroupElement('traders', element=SavingTraderElement, size=n_agents, values={'wealth': wealth})],
        interactions=[trades],
    )

kinetic_wealth_welfare(classes=9, *, cooperation=0.2, competition=0.3, eta=1.0, tax=0.0)

Return the kinetic theory of wealth classes, with a welfare policy when the tax is positive.

The population is a distribution over wealth classes, a site group on a lattice of the classes. Encounters between classes move particles one class up or down, with cooperation narrowing the gaps and competition widening them, so the total density stays constant. When competition wins, wealth polarises toward the poorest and the richest classes, and the Gini coefficient grows. A welfare policy is an encounter with a group of fixed density, which moves the richest third of the classes down and the poorest third up and lowers the stationary Gini coefficient.

Parameters:

Name	Type	Description	Default
classes	int	The number of wealth classes.	9
cooperation	float	The probability of the move that narrows the gap between the classes of an encounter.	0.2
competition	float	The probability of the move that widens it.	0.3
eta	float	The encounter rate.	1.0
tax	float	The rate of the welfare policy, 0 for none.	0.0

Returns:

Name	Type	Description
system	System	The system, with the distribution wealth, the encounters exchange, and with a tax the group state and the policy welfare. Its standard observables are the density, the mean wealth, and the Gini coefficient of wealth.

Examples:

>>> from skcomplex.systems.economics.econophysics import kinetic_wealth_welfare
>>> [item.label for item in kinetic_wealth_welfare(tax=0.2).interactions]
['exchange', 'welfare']

Source code in src/skcomplex/systems/economics/econophysics/_welfare.py

@register_system(
    field='economics/econophysics',
    tags=('wealth', 'kinetic', 'distribution', 'welfare'),
    references=WELFARE_REFERENCES,
    defaults={'until': 100, 'dt': 1.0},
    observables=lambda system: wealth_observables(_classes_of(system)),
)
def kinetic_wealth_welfare(
    classes: int = 9,
    *,
    cooperation: float = 0.2,
    competition: float = 0.3,
    eta: float = 1.0,
    tax: float = 0.0,
) -> System:
    """Return the kinetic theory of wealth classes, with a welfare policy when the tax is positive.

    The population is a distribution over wealth classes, a site group on a lattice of the classes. Encounters
    between classes move particles one class up or down, with cooperation narrowing the gaps and competition widening
    them, so the total density stays constant. When competition wins, wealth polarises toward the poorest and the
    richest classes, and the Gini coefficient grows. A welfare policy is an encounter with a group of fixed density,
    which moves the richest third of the classes down and the poorest third up and lowers the stationary Gini
    coefficient.

    Args:
        classes:
            The number of wealth classes.

        cooperation:
            The probability of the move that narrows the gap between the classes of an encounter.

        competition:
            The probability of the move that widens it.

        eta:
            The encounter rate.

        tax:
            The rate of the welfare policy, 0 for none.

    Returns:
        system:
            The system, with the distribution `wealth`, the encounters `exchange`, and with a tax the group `state` and
            the policy `welfare`. Its standard observables are the density, the mean wealth, and the Gini coefficient
            of `wealth`.

    Examples:
        >>> from skcomplex.systems.economics.econophysics import kinetic_wealth_welfare
        >>> [item.label for item in kinetic_wealth_welfare(tax=0.2).interactions]
        ['exchange', 'welfare']
    """
    initial = np.exp(-((wealth_classes(classes) - 0.5) ** 2) / 0.05)
    components: list[GroupElement] = [
        GroupElement(
            'wealth',
            element=DensityElement,
            size=classes,
            on='sites',
            values={'f': initial / initial.sum()},
        ),
    ]
    interactions: list[BaseInteraction] = [
        WealthExchangeInteraction(
            'exchange',
            candidate='wealth',
            field='wealth',
            classes=classes,
            eta=eta,
            cooperation=cooperation,
            competition=competition,
        ),
    ]
    if tax > 0:
        components.append(GroupElement('state', element=DensityElement, size=1, values={'f': 1.0}))
        interactions.append(
            WelfarePolicyInteraction('welfare', candidate='wealth', field='state', classes=classes, tax=tax),
        )
    return System(
        elements=components,
        interactions=interactions,
        space=LatticeSpace(classes, boundary='fixed'),
        time=ContinuousTime(),
    )

kolkata_restaurant(n_agents=100, *, n_restaurants=None, strategy='random', random_state=0)

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.

Parameters:

Name	Type	Description	Default
n_agents	int	The number of diners.	100
n_restaurants	int | None	The number of restaurants, or None for as many as diners.	None
strategy	str	'random' or 'crowd_avoiding'.	'random'
random_state	int | Generator | None	The seed or generator of the first choices.	0

Returns:

Name	Type	Description
system	System	The system, 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:

Type	Description
ValueError	The strategy is unknown.

Examples:

>>> from skcomplex.systems.economics.econophysics import kolkata_restaurant
>>> type(kolkata_restaurant(10, strategy='crowd_avoiding').interactions[0]).__name__
'CrowdAvoidanceInteraction'

Source code in src/skcomplex/systems/economics/econophysics/_kolkata.py

@register_system(
    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 system, 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.systems.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, system, option: 0.0,
            intensity=0.0,
            form='agent',
        )
    )
    restaurants = RestaurantsInteraction('restaurants', target='diners', n_restaurants=n_restaurants)
    return System(elements=[diners], interactions=[choice, restaurants])

mean_wealth(s, wealth)

Return the mean wealth of a distribution over wealth classes.

Parameters:

Name	Type	Description	Default
s	SimpleNamespace	The namespace of one member of the distribution.	required
wealth	NDArray[float64]	The wealth of every class.	required

Returns:

Name	Type	Description
mean	Expression	The normalized first moment.

Source code in src/skcomplex/systems/economics/econophysics/_welfare.py

def mean_wealth(s: SimpleNamespace, wealth: npt.NDArray[np.float64]) -> Expression:
    """Return the mean wealth of a distribution over wealth classes.

    Args:
        s:
            The namespace of one member of the distribution.

        wealth:
            The wealth of every class.

    Returns:
        mean:
            The normalized first moment.
    """
    return moment(s.f, 1, wealth, normalized=True)

minority_game(n_agents=101, *, memory=3, evolutionary=False, period=50, fraction=0.1, random_state=0)

Return the minority game of an odd number of agents with two random strategies each.

Parameters:

Name	Type	Description	Default
n_agents	int	The number of agents, odd so that there is always a minority.	101
memory	int	The memory m of the strategies, so alpha = 2^m / N.	3
evolutionary	bool	Whether every period rounds the poorest agents take new random strategies.	False
period	int	The number of rounds between two evolutions of the evolutionary version.	50
fraction	float	The fraction of poorest agents that take new strategies at an evolution.	0.1
random_state	int | Generator | None	The seed or generator of the strategies.	0

Returns:

Name	Type	Description
system	System	The system, 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:

Type	Description
ValueError	The number of agents is even.

Examples:

>>> from skcomplex.systems.economics.econophysics import minority_game
>>> minority_game(11, memory=2).interactions[1].memory
2

Source code in src/skcomplex/systems/economics/econophysics/_minority.py

@register_system(
    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 system, 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.systems.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, system, 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')]),
        ),
    )

policy_table(classes)

Return the rate pattern and the transition table of a welfare policy that taxes the rich and helps the poor.

The policy moves a candidate of the richest third of the classes one class down, and a candidate of the poorest third one class up, and leaves the middle classes alone.

Parameters:

Name	Type	Description	Default
classes	int	The number of wealth classes.	required

Returns:

Name	Type	Description
pattern	NDArray[float64]	The rate pattern eta[h, 0], 1 for the classes the policy acts on and 0 for the others.
table	NDArray[float64]	The transition table A[h, 0, i] of the policy.

Examples:

>>> from skcomplex.systems.economics.econophysics import policy_table
>>> policy_table(3)[0].ravel().tolist()
[1.0, 0.0, 1.0]

Source code in src/skcomplex/systems/economics/econophysics/_welfare.py

def policy_table(classes: int) -> tuple[npt.NDArray[np.float64], npt.NDArray[np.float64]]:
    """Return the rate pattern and the transition table of a welfare policy that taxes the rich and helps the poor.

    The policy moves a candidate of the richest third of the classes one class down, and a candidate of the poorest
    third one class up, and leaves the middle classes alone.

    Args:
        classes:
            The number of wealth classes.

    Returns:
        pattern:
            The rate pattern `eta[h, 0]`, 1 for the classes the policy acts on and 0 for the others.

        table:
            The transition table `A[h, 0, i]` of the policy.

    Examples:
        >>> from skcomplex.systems.economics.econophysics import policy_table
        >>> policy_table(3)[0].ravel().tolist()
        [1.0, 0.0, 1.0]
    """
    pattern = np.zeros((classes, 1))
    table = np.zeros((classes, 1, classes))
    third = max(1, classes // 3)
    for candidate in range(classes):
        outcome = candidate + 1 if candidate < third else candidate - 1 if candidate >= classes - third else candidate
        pattern[candidate, 0] = float(outcome != candidate)
        table[candidate, 0, outcome] = 1.0
    return pattern, table

strategy_table(code, memory)

Return the table of a minority game strategy, the bit it plays after each of the 2^m histories.

A table has 2^m bits, too many to store as one number beyond small memories, so a strategy is stored as the integer code that seeds a generator of its table.

Parameters:

Name	Type	Description	Default
code	float	The integer code of the strategy.	required
memory	int	The memory m, the number of past outcomes the strategy reads.	required

Returns:

Name	Type	Description
table	IntArray	The bit of the strategy for every history, from 0 to 2^m - 1.

Examples:

>>> from skcomplex.systems.economics.econophysics import strategy_table
>>> strategy_table(7, 2).shape
(4,)

Source code in src/skcomplex/systems/economics/econophysics/_minority.py

def strategy_table(code: float, memory: int) -> IntArray:
    """Return the table of a minority game strategy, the bit it plays after each of the `2^m` histories.

    A table has `2^m` bits, too many to store as one number beyond small memories, so a strategy is stored as the
    integer code that seeds a generator of its table.

    Args:
        code:
            The integer code of the strategy.

        memory:
            The memory `m`, the number of past outcomes the strategy reads.

    Returns:
        table:
            The bit of the strategy for every history, from 0 to `2^m - 1`.

    Examples:
        >>> from skcomplex.systems.economics.econophysics import strategy_table
        >>> strategy_table(7, 2).shape
        (4,)
    """
    return np.random.default_rng(int(code)).integers(0, 2, 2**memory)

wealth_classes(classes)

Return the wealth of every class, evenly spaced from 0 for the poorest to 1 for the richest.

Parameters:

Name	Type	Description	Default
classes	int	The number of wealth classes.	required

Returns:

Name	Type	Description
wealth	NDArray[float64]	The wealth of every class.

Examples:

>>> from skcomplex.systems.economics.econophysics import wealth_classes
>>> wealth_classes(3).tolist()
[0.0, 0.5, 1.0]

Source code in src/skcomplex/systems/economics/econophysics/_welfare.py

def wealth_classes(classes: int) -> npt.NDArray[np.float64]:
    """Return the wealth of every class, evenly spaced from 0 for the poorest to 1 for the richest.

    Args:
        classes:
            The number of wealth classes.

    Returns:
        wealth:
            The wealth of every class.

    Examples:
        >>> from skcomplex.systems.economics.econophysics import wealth_classes
        >>> wealth_classes(3).tolist()
        [0.0, 0.5, 1.0]
    """
    return np.linspace(0.0, 1.0, classes)

wealth_density(s, wealth)

Return the density of a distribution over wealth classes, its order-0 moment.

Parameters:

Name	Type	Description	Default
s	SimpleNamespace	The namespace of one member of the distribution.	required
wealth	NDArray[float64]	The wealth of every class.	required

Returns:

Name	Type	Description
density	Expression	The order-0 moment.

Source code in src/skcomplex/systems/economics/econophysics/_welfare.py

def wealth_density(s: SimpleNamespace, wealth: npt.NDArray[np.float64]) -> Expression:
    """Return the density of a distribution over wealth classes, its order-0 moment.

    Args:
        s:
            The namespace of one member of the distribution.

        wealth:
            The wealth of every class.

    Returns:
        density:
            The order-0 moment.
    """
    return moment(s.f, 0, wealth)

wealth_observables(classes)

Return the density, the mean wealth, and the Gini coefficient of a distribution wealth over wealth classes.

Parameters:

Name	Type	Description	Default
classes	int	The number of wealth classes.	required

Returns:

Name	Type	Description
observables	list[ObservableSpec]	The observables, each reading the moments of the distribution.

Examples:

>>> from skcomplex.systems.economics.econophysics import wealth_observables
>>> [item.label for item in wealth_observables(9)]
['density', 'mean_wealth', 'gini']

Source code in src/skcomplex/systems/economics/econophysics/_welfare.py

def wealth_observables(classes: int) -> list[ObservableSpec]:
    """Return the density, the mean wealth, and the Gini coefficient of a distribution `wealth` over wealth classes.

    Args:
        classes:
            The number of wealth classes.

    Returns:
        observables:
            The observables, each reading the moments of the distribution.

    Examples:
        >>> from skcomplex.systems.economics.econophysics import wealth_observables
        >>> [item.label for item in wealth_observables(9)]
        ['density', 'mean_wealth', 'gini']
    """
    wealth = wealth_classes(classes)
    return [
        ObservableSpec('density', partial(wealth_density, wealth=wealth), group='wealth'),
        ObservableSpec('mean_wealth', partial(mean_wealth, wealth=wealth), group='wealth'),
        ObservableSpec('gini', partial(gini_coefficient, wealth=wealth), group='wealth'),
    ]