Skip to content

Econophysics

Econophysics: wealth exchange between agents.

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.models.economics.econophysics import CrowdAvoidanceInteraction
>>> CrowdAvoidanceInteraction('choice', target='diners', options=(0.0, 1.0)).check().rule
'metropolis'
Source code in src/skcomplex/models/economics/econophysics/_crowd_avoidance.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
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.models.economics.econophysics import DebtExchangeInteraction
>>> DebtExchangeInteraction('trades', source='traders', target='traders', debt=5.0).check().debt
5.0
Source code in src/skcomplex/models/economics/econophysics/_debt_exchange.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
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.elements import GroupElement
>>> from skcomplex.models.economics.econophysics import DinerElement
>>> GroupElement('diners', element=DinerElement, size=10).size
10
Source code in src/skcomplex/models/economics/econophysics/_diner.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
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.models.economics.econophysics import MinorityGameInteraction
>>> MinorityGameInteraction('game', players='players', memory=4).memory
4
Source code in src/skcomplex/models/economics/econophysics/_minority_game.py 
78
79
80
81
82
83
84
85
86
87
88
89
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, model, 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
model ModelView

The model 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/models/economics/econophysics/_minority_game.py 
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
def payoff(self, agent: MinorityPlayerView, model: ModelView, option: int) -> float:
    """Return the virtual payoff of a strategy of an agent in the last round, `+1` if it played the winning side.

    Args:
        agent:
            The agent.

        model:
            The model view of the rule that reads the payoff.

        option:
            The strategy, 0 for the first and 1 for the second.

    Returns:
        payoff:
            `+1` when the strategy would have played the winning side, or `-1`.
    """
    history = int(model.history(f'{self.label}.history')[-1])
    winner = -1 if model.history(f'{self.label}.attendance')[-1] > 0 else 1
    code = agent.second if option else agent.first
    return float((2 * int(strategy_table(code, self.memory)[history]) - 1) * winner)

play(model)

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

Parameters:

Name Type Description Default
model ModelView

The model.

 required
Source code in src/skcomplex/models/economics/econophysics/_minority_game.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
def play(self, model: ModelView) -> None:
    """Play one round, pay the agents, and update and report the history.

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

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.elements import GroupElement
>>> from skcomplex.models.economics.econophysics import MinorityPlayerElement
>>> GroupElement('players', element=MinorityPlayerElement, size=11, values={'first': 3, 'second': 5}).size
11
Source code in src/skcomplex/models/economics/econophysics/_minority_player.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
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.elements import GroupElement
>>> from skcomplex.models.economics.econophysics import PatronElement
>>> GroupElement('patrons', element=PatronElement, size=100, values={'predictors': 7}).size
100
Source code in src/skcomplex/models/economics/econophysics/_patron.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
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, model)

Forecast the attendance with the most accurate predictor held.

Parameters:

Name Type Description Default
agent PatronView

The patron.

 required
model ModelView

The model.

 required
Source code in src/skcomplex/models/economics/econophysics/_patron.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def predict(self, agent: PatronView, model: ModelView) -> None:
    """Forecast the attendance with the most accurate predictor held.

    Args:
        agent:
            The patron.

        model:
            The model.
    """
    pool = model.agents(self.pool, PredictorPoolView)[0]
    errors, predicted = np.ravel(pool.errors), np.ravel(pool.forecasts)
    mask = int(agent.predictors)
    held = [index for index in range(len(PREDICTORS)) if mask >> index & 1]
    agent.forecast = float(predicted[held[int(np.argmin(errors[held]))]])

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.elements import GroupElement
>>> from skcomplex.models.economics.econophysics import PredictorPoolElement
>>> GroupElement('pool', element=PredictorPoolElement, size=1).size
1
Source code in src/skcomplex/models/economics/econophysics/_predictor_pool.py 
110
111
112
113
114
115
116
117
118
119
120
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, model, 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
model ModelView

The model 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/models/economics/econophysics/_predictor_pool.py 
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
@staticmethod
def error(agent: PredictorPoolView, model: ModelView, index: int) -> float:
    """Return the absolute error of a predictor last week, or its error so far before the first attendance.

    Args:
        agent:
            The pool.

        model:
            The model view of the rule that reads the error.

        index:
            The index of the predictor in the pool.

    Returns:
        error:
            The absolute error of the forecast of the predictor.
    """
    attendance = model.history(PredictorPoolElement.attendance)
    if not len(attendance):
        return float(np.ravel(agent.errors)[index])
    return float(abs(np.ravel(agent.forecasts)[index] - attendance[-1]))

forecast(agent, model)

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

Parameters:

Name Type Description Default
agent PredictorPoolView

The pool.

 required
model ModelView

The model.

 required
Source code in src/skcomplex/models/economics/econophysics/_predictor_pool.py 
148
149
150
151
152
153
154
155
156
157
158
def forecast(self, agent: PredictorPoolView, model: ModelView) -> None:
    """Forecast next week's attendance with every predictor of the pool.

    Args:
        agent:
            The pool.

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

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.models.economics.econophysics import RestaurantsInteraction
>>> RestaurantsInteraction('restaurants', target='diners', n_restaurants=5).check().groups
'state'
Source code in src/skcomplex/models/economics/econophysics/_restaurants.py 
45
46
47
48
49
50
51
52
53
54
55
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, model)

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
model ModelViewProtocol

The model view of the restaurants.

 required
Source code in src/skcomplex/models/economics/econophysics/_restaurants.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
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)

report(model)

Report the fraction of restaurants with at least one diner.

Parameters:

Name Type Description Default
model ModelViewProtocol

The model view of the restaurants.

 required
Source code in src/skcomplex/models/economics/econophysics/_restaurants.py 
72
73
74
75
76
77
78
79
80
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)

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.models.economics.econophysics import SavingExchangeInteraction
>>> SavingExchangeInteraction('trades', source='traders', target='traders').check().saving
0.0
Source code in src/skcomplex/models/economics/econophysics/_saving_exchange.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
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 models, 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.elements import GroupElement
>>> from skcomplex.models.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/models/economics/econophysics/_saving_trader.py 
34
35
36
37
38
39
40
41
42
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 model, 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.models.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/models/economics/econophysics/_exchange.py 
37
38
39
40
41
42
43
44
45
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.models.economics.econophysics import WealthExchangeInteraction
>>> WealthExchangeInteraction('exchange', candidate='wealth', field='wealth').check().competition
0.3
Source code in src/skcomplex/models/economics/econophysics/_wealth_exchange.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
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.models.economics.econophysics import WelfarePolicyInteraction
>>> WelfarePolicyInteraction('welfare', candidate='wealth', field='state').check().tax
0.1
Source code in src/skcomplex/models/economics/econophysics/_welfare_policy.py 
78
79
80
81
82
83
84
85
86
87
88
89
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 model 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 model, with the group traders and the transfer gifts.

Examples:

>>> from skcomplex.models.economics.econophysics import boltzmann_wealth
>>> boltzmann_wealth(10).elements[0].check().prepare().size_
10
Source code in src/skcomplex/models/economics/econophysics/_exchange.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
@register_model(
    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 model 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 model, with the group `traders` and the transfer `gifts`.

    Examples:
        >>> from skcomplex.models.economics.econophysics import boltzmann_wealth
        >>> boltzmann_wealth(10).elements[0].check().prepare().size_
        10
    """
    gifts = TransferInteraction(
        'gifts',
        source='traders',
        target='traders',
        state='wealth',
        amount=1.0,
        pairing='random',
        guard=lambda giver, receiver, model: 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 model, with the group patrons, the one-member group pool of the forecasts and the errors of the predictors, the reinforcement scoring of the errors, and the choice choice to go when the forecast is below the capacity of the patrons.

Examples:

>>> from skcomplex.models.economics.econophysics import el_farol
>>> el_farol(10).elements[0].check().prepare().size_
10
Source code in src/skcomplex/models/economics/econophysics/_el_farol.py 
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
@register_model(
    field='economics/econophysics',
    tags=('el farol', 'agents', 'inductive reasoning', 'bounded rationality'),
    references=('Arthur 1994', 'Yakovenko and Rosser 2009', 'Carmichael and Hadzikadic 2019 ch. 9'),
    defaults={'until': 200},
    observables=lambda system: [ObservableSpec('attendance', group='patrons', variable='attends', reduce='sum')],
)
def el_farol(
    n_agents: int = 100,
    *,
    capacity: float = 60.0,
    n_predictors: int = 6,
    discount: float = 0.9,
    random_state: int | np.random.Generator | None = 0,
) -> System:
    """Return Arthur's El Farol bar problem, whose attendance self-organizes around the capacity.

    Every patron holds a random set of predictors of next week's attendance, drawn from a pool of 46, and acts on
    its currently most accurate one, going when it forecasts room in the bar. The attendance is random-looking, yet
    its mean settles near the capacity. The patrons read the past attendances from the observable `attendance` of
    the run, which the catalog lists, so a run passes it in `observables`.

    Args:
        n_agents:
            The number of patrons.

        capacity:
            The attendance above which the bar is too crowded to enjoy.

        n_predictors:
            The number of predictors every patron holds.

        discount:
            The weight of past errors in the accuracy of a predictor.

        random_state:
            The seed or generator of the predictors that every patron holds.

    Returns:
        system:
            The model, with the group `patrons`, the one-member group `pool` of the forecasts and the errors of the
            predictors, the reinforcement `scoring` of the errors, and the choice `choice` to go when the forecast is
            below the capacity of the patrons.

    Examples:
        >>> from skcomplex.models.economics.econophysics import el_farol
        >>> el_farol(10).elements[0].check().prepare().size_
        10
    """
    rng = np.random.default_rng(random_state)
    masks = [
        float(sum(1 << int(index) for index in rng.choice(len(PREDICTORS), n_predictors, replace=False)))
        for _ in range(n_agents)
    ]
    patrons = GroupElement(
        'patrons',
        element=PatronElement,
        size=n_agents,
        values={'predictors': masks, 'capacity': capacity},
    )
    pool = GroupElement(
        'pool',
        element=PredictorPoolElement,
        size=1,
        values={'forecasts': np.zeros((1, len(PREDICTORS))), 'errors': np.zeros((1, len(PREDICTORS)))},
    )
    scoring = ReinforcementInteraction(
        'scoring',
        target='pool',
        scores='errors',
        payoff=PredictorPoolElement.error,
        rate=1 - discount,
        form='agent',
        phase='learn',
    )
    choice = ChoiceInteraction(
        'choice',
        target='patrons',
        state='attends',
        options=(0.0, 1.0),
        utility=lambda agent, model, option: option if agent.forecast < agent.capacity else -option,
        rule='best',
        form='agent',
        phase='decide',
    )
    return System(
        elements=[patrons, pool],
        interactions=[scoring, choice],
        time=DiscreteTime(
            schedule=PhasesScheduleSpec(
                [PhaseSpec(name) for name in ('learn', 'forecast', 'predict', 'decide')],
            ),
        ),
    )

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.models.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/models/economics/econophysics/_wealth_exchange.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
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.models.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.models.economics.econophysics import forecasts
>>> forecasts([60.0] * 10)[:3].tolist()
[60.0, 60.0, 60.0]
Source code in src/skcomplex/models/economics/econophysics/_predictor_pool.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
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.models.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/models/economics/econophysics/_welfare.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
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 model 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 model, 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.models.economics.econophysics import kinetic_wealth_exchange
>>> type(kinetic_wealth_exchange(10, debt=2.0).interactions[0]).__name__
'DebtExchangeInteraction'
Source code in src/skcomplex/models/economics/econophysics/_kinetic.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
@register_model(
    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 model 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 model, 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.models.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 model, 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.models.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/models/economics/econophysics/_welfare.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
@register_model(
    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 model, 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.models.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 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:

Type Description
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'
Source code in src/skcomplex/models/economics/econophysics/_kolkata.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
@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])

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/models/economics/econophysics/_welfare.py 
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
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 model, with the group players, the choice choice of the strategy of higher virtual score, the interaction game, which reports the attendance game.attendance, the cumulative reinforcement scoring of the virtual scores, and in the evolutionary version the truncation evolution, which gives the poorest agents new random strategies and no wealth every period rounds.

Raises:

Type Description
ValueError

The number of agents is even.

Examples:

>>> from skcomplex.models.economics.econophysics import minority_game
>>> minority_game(11, memory=2).interactions[1].memory
2
Source code in src/skcomplex/models/economics/econophysics/_minority.py 
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
@register_model(
    field='economics/econophysics',
    tags=('minority game', 'agents', 'inductive reasoning', 'evolution'),
    references=('Challet and Zhang 1997', 'Yakovenko and Rosser 2009', 'Carmichael and Hadzikadic 2019 ch. 9'),
    defaults={'until': 500},
)
def minority_game(
    n_agents: int = 101,
    *,
    memory: int = 3,
    evolutionary: bool = False,
    period: int = 50,
    fraction: float = 0.1,
    random_state: int | np.random.Generator | None = 0,
) -> System:
    """Return the minority game of an odd number of agents with two random strategies each.

    Args:
        n_agents:
            The number of agents, odd so that there is always a minority.

        memory:
            The memory `m` of the strategies, so `alpha = 2^m / N`.

        evolutionary:
            Whether every `period` rounds the poorest agents take new random strategies.

        period:
            The number of rounds between two evolutions of the evolutionary version.

        fraction:
            The fraction of poorest agents that take new strategies at an evolution.

        random_state:
            The seed or generator of the strategies.

    Returns:
        system:
            The model, with the group `players`, the choice `choice` of the strategy of higher virtual score, the
            interaction `game`, which reports the attendance `game.attendance`, the cumulative reinforcement
            `scoring` of the virtual scores, and in the evolutionary version the truncation `evolution`, which gives
            the poorest agents new random strategies and no wealth every `period` rounds.

    Raises:
        ValueError:
            The number of agents is even.

    Examples:
        >>> from skcomplex.models.economics.econophysics import minority_game
        >>> minority_game(11, memory=2).interactions[1].memory
        2
    """
    if n_agents % 2 == 0:
        error_msg = f'The minority game needs an odd number of agents, got {n_agents}.'
        raise ValueError(error_msg)
    codes = np.random.default_rng(random_state).integers(MAX_SEED, size=(2, n_agents)).astype(float)
    players = GroupElement(
        'players',
        element=MinorityPlayerElement,
        size=n_agents,
        values={'first': codes[0], 'second': codes[1]},
    )
    choice = ChoiceInteraction(
        'choice',
        target='players',
        state='strategy',
        options=(0.0, 1.0),
        utility=lambda agent, model, option: agent.second_score if option else agent.first_score,
        rule='best',
        form='agent',
        phase='choose',
    )
    game = MinorityGameInteraction('game', players='players', memory=memory, phase='play')
    scoring = ReinforcementInteraction(
        'scoring',
        target='players',
        scores=('first_score', 'second_score'),
        payoff=game.payoff,
        rule='cumulative',
        form='agent',
        phase='score',
    )
    evolution = EvolutionInteraction(
        'evolution',
        target='players',
        strategy=('first', 'second'),
        fitness='wealth',
        replacement='truncation',
        fraction=fraction,
        mutation=1.0,
        change='any',
        alphabet=MAX_SEED,
        period=period,
        reset={'first_score': 0.0, 'second_score': 0.0, 'wealth': 0.0},
        phase='score',
    )
    return System(
        elements=[players],
        interactions=[choice, game, scoring, *([evolution] if evolutionary else [])],
        time=DiscreteTime(
            schedule=PhasesScheduleSpec([PhaseSpec('choose'), PhaseSpec('play'), PhaseSpec('score')]),
        ),
    )

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.models.economics.econophysics import policy_table
>>> policy_table(3)[0].ravel().tolist()
[1.0, 0.0, 1.0]
Source code in src/skcomplex/models/economics/econophysics/_welfare_policy.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
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.models.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.models.economics.econophysics import strategy_table
>>> strategy_table(7, 2).shape
(4,)
Source code in src/skcomplex/models/economics/econophysics/_minority_game.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
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.models.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.models.economics.econophysics import wealth_classes
>>> wealth_classes(3).tolist()
[0.0, 0.5, 1.0]
Source code in src/skcomplex/models/economics/econophysics/_welfare.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
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.models.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/models/economics/econophysics/_welfare.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
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.models.economics.econophysics import wealth_observables
>>> [item.label for item in wealth_observables(9)]
['density', 'mean_wealth', 'gini']
Source code in src/skcomplex/models/economics/econophysics/_welfare.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
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.models.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'),
    ]