Skip to content

Finance

Finance: asset prices, growth processes, wealth, systemic risk, and arbitrage.

ArbitrageDensityElement(label, visual=None, *, rho=0.0, D=0.5, A_x=0.0, E=0.5, A_t=0.0)

Bases: Element

Density of a good carried by random-walk traders, minimally coupled to an exchange rate connection.

Parameters:

Name Type Description Default
label str

The label of the density.

required
visual VisualSpec | None

How the element is drawn, or None for default_visual_.

None
rho Annotated[object, StateSpec(domain=closed_open(0, oo))]

The density of the good.

0.0
D Annotated[object, ParameterSpec(domain=open(0, oo))]

The diffusion constant of the traders.

0.5
A_x Annotated[object, ParameterSpec()]

The initial spatial component of the exchange rate connection.

0.0
E Annotated[object, ParameterSpec()]

The arbitrage field strength, the growth rate of the spatial component.

0.5
A_t Annotated[object, ParameterSpec()]

The temporal component of the exchange rate connection.

0.0

Examples:

>>> from skcomplex.models.economics.finance import ArbitrageDensityElement
>>> ArbitrageDensityElement('market', E=1.0).E
1.0
Source code in src/skcomplex/models/economics/finance/_arbitrage.py
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def __init__(
    self,
    label: str,
    visual: VisualSpec | None = None,
    *,
    rho: Annotated[object, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
    D: Annotated[object, ParameterSpec(domain=RealSet.open(0, oo))] = 0.5,
    A_x: Annotated[object, ParameterSpec()] = 0.0,
    E: Annotated[object, ParameterSpec()] = 0.5,
    A_t: Annotated[object, ParameterSpec()] = 0.0,
) -> None:
    super().__init__(label, visual=visual)
    self.rho = rho
    self.D = D
    self.A_x = A_x
    self.E = E
    self.A_t = A_t

AssetElement(label, visual=None, *, price=1.0)

Bases: Element

Traded asset described by its price.

Parameters:

Name Type Description Default
label str

The label of the asset.

required
visual VisualSpec | None

How the element is drawn, or None for default_visual_.

None
price Annotated[float, StateSpec(domain=open(0, oo))]

The initial price.

1.0

Examples:

>>> from skcomplex.models.economics.finance import AssetElement
>>> AssetElement('stock', price=100.0).price
100.0
Source code in src/skcomplex/models/economics/finance/_asset.py
34
35
36
37
38
39
40
41
42
def __init__(
    self,
    label: str,
    visual: VisualSpec | None = None,
    *,
    price: Annotated[float, StateSpec(domain=RealSet.open(0, oo))] = 1.0,
) -> None:
    super().__init__(label, visual=visual)
    self.price = price

BookInteraction(label, *, requesters, providers, rule=RandomSearchSpec(), limit=('bid', 'ask'), quantity=('demand', 'supply'), allowance='money', transfers=(('money', -1), ('shares', 1)), phase='trade', on_match=None, opening=100.0)

Bases: MatchingInteraction[ModelView]

Limit order book that matches the orders of the traders and reports the last traded price.

The book is a matching of the traders with themselves, which binds the requesters' limit to their bid and the providers' to their ask, and every trade is two transfers, money from buyer to seller and shares from seller to buyer, so the total money and the total shares stay the same, which the default run checks as invariants. The book reports the last traded price, the previous price when nothing trades, and the traded volume.

Parameters:

Name Type Description Default
label str

The label of the book.

required
requesters Annotated[str, RoleSpec()]

The path of the group of buying traders.

required
providers Annotated[str, RoleSpec()]

The path of the group of selling traders.

required
rule MatchingRule

The spec of the algorithm that matches the orders, such as ContinuousMatchingSpec().

RandomSearchSpec()
limit str | tuple[str, str]

The states with the limit prices of the buyers and of the sellers.

('bid', 'ask')
quantity str | tuple[str, str]

The states with the quantities the buyers want and the sellers offer.

('demand', 'supply')
allowance str

The state of a buyer with the money it can spend.

'money'
transfers Sequence[tuple[str, int]]

The states every trade moves, money paid by the buyer and shares delivered to it.

(('money', -1), ('shares', 1))
phase str | None

The phase the book clears in.

'trade'
on_match Callable[[MatchSpec, ModelView], None] | None

A function of every trade and the model view, run after the transfers, or None.

None
opening float

The price before the first trade.

100.0

Examples:

>>> from skcomplex.models.economics.finance import BookInteraction
>>> BookInteraction('book', requesters='traders', providers='traders').reports
('price', 'volume')
Source code in src/skcomplex/models/economics/finance/_book.py
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
def __init__(
    self,
    label: str,
    *,
    requesters: Annotated[str, RoleSpec()],
    providers: Annotated[str, RoleSpec()],
    rule: MatchingRule = RandomSearchSpec(),
    limit: str | tuple[str, str] = ('bid', 'ask'),
    quantity: str | tuple[str, str] = ('demand', 'supply'),
    allowance: str = 'money',
    transfers: Sequence[tuple[str, int]] = (('money', -1), ('shares', 1)),
    phase: str | None = 'trade',
    on_match: Callable[[MatchSpec, ModelView], None] | None = None,
    opening: float = 100.0,
) -> None:
    super().__init__(
        label,
        requesters=requesters,
        providers=providers,
        rule=rule,
        limit=limit,
        quantity=quantity,
        allowance=allowance,
        transfers=transfers,
        phase=phase,
        on_match=on_match,
    )
    self.opening = opening

DebtRankCascadeInteraction(label, *, banks, impact=())

Bases: Interaction

Synchronous spread of distress, h_i = min(1, h_i + sum_j W_ji h_j) over the distressed banks j.

A bank whose distress becomes positive turns distressed, and a distressed bank turns inactive after one step, so every bank passes its distress on once.

Parameters:

Name Type Description Default
label str

The label of the interaction.

required
banks Annotated[str, RoleSpec()]

The path of the group of banks.

required
impact Sequence[Sequence[float]]

The impact matrix, impact[i][j] the relative loss of j when i defaults.

()

Examples:

>>> from skcomplex.models.economics.finance import DebtRankCascadeInteraction
>>> DebtRankCascadeInteraction('cascade', banks='banks', impact=[[0.0, 0.5], [0.0, 0.0]]).impact
[[0.0, 0.5], [0.0, 0.0]]
Source code in src/skcomplex/models/economics/finance/_debt_rank_cascade.py
47
48
49
50
51
52
53
54
55
56
def __init__(
    self,
    label: str,
    *,
    banks: Annotated[str, RoleSpec()],
    impact: Sequence[Sequence[float]] = (),
) -> None:
    super().__init__(label)
    self.banks = banks
    self.impact = impact

spread(model)

Advance the distress and the states of every bank by one step.

Parameters:

Name Type Description Default
model ModelView

The model.

required
Source code in src/skcomplex/models/economics/finance/_debt_rank_cascade.py
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def spread(self, model: ModelView) -> None:
    """Advance the distress and the states of every bank by one step.

    Args:
        model:
            The model.
    """
    banks = sorted(model.agents('banks', InterbankView), key=lambda bank: bank.id)
    h = np.array([bank.h for bank in banks])
    status = np.array([bank.status for bank in banks])
    distressed = status == DISTRESSED
    new_h = np.minimum(1.0, h + np.asarray(self.impact, dtype=float).T @ (h * distressed))
    hit = (status == UNDISTRESSED) & (new_h > 0)
    new_status = np.where(distressed, INACTIVE, np.where(hit, DISTRESSED, status))
    for bank, level, code in zip(banks, new_h, new_status, strict=True):
        bank.h, bank.status = float(level), float(code)

GeometricBrownianMotionInteraction(label, *, target, mu=0.05, sigma=0.2, process=None, index=None)

Bases: RateInteraction

Geometric Brownian motion, dS = mu * S * dt + sigma * S * dW, of the price of an asset.

The Wiener process dW is independent by default, or process index of a WienerProcessSpec that correlates it with the motions of other assets.

Parameters:

Name Type Description Default
label str

The label of the interaction.

required
target Annotated[str, RoleSpec()]

The path of the asset.

required
mu Annotated[float, ParameterSpec()]

The drift.

0.05
sigma Annotated[float, ParameterSpec(domain=closed_open(0, oo))]

The volatility.

0.2
process Annotated[WienerProcessSpec | None, RoleSpec(noise=True)]

The Wiener processes, one of which drives the price, or None for an independent process.

None
index int | None

The process of process that drives the price, or None without processes.

None

Examples:

>>> from skcomplex.models.economics.finance import GeometricBrownianMotionInteraction
>>> from skcomplex.specs import WienerProcessSpec
>>> GeometricBrownianMotionInteraction('gbm', target='stock', sigma=0.3).sigma
0.3
>>> market = WienerProcessSpec('market', dimension=2)
>>> GeometricBrownianMotionInteraction('gbm', target='stock', process=market, index=1).check().index
1
Source code in src/skcomplex/models/economics/finance/_geometric_brownian_motion.py
51
52
53
54
55
56
57
58
59
60
61
62
63
def __init__(
    self,
    label: str,
    *,
    target: Annotated[str, RoleSpec()],
    mu: Annotated[float, ParameterSpec()] = 0.05,
    sigma: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.2,
    process: Annotated[WienerProcessSpec | None, RoleSpec(noise=True)] = None,
    index: int | None = None,
) -> None:
    super().__init__(label, target=target, state='price', process=process, index=index)
    self.mu = mu
    self.sigma = sigma

GibratFirmElement(label, visual=None, *, x=100.0, previous=100.0, mu=1.02, sigma=0.05)

Bases: Element

Firm whose size is multiplied every step by an independent normal growth factor, x = lambda x.

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
x Annotated[float, StateSpec()]

The initial size.

100.0
previous Annotated[float, StateSpec()]

The size before the last step.

100.0
mu Annotated[float, ParameterSpec()]

The mean of the growth factor.

1.02
sigma Annotated[float, ParameterSpec(domain=closed_open(0, oo))]

The standard deviation of the growth factor.

0.05

Examples:

>>> from skcomplex.core.elements import GroupElement
>>> from skcomplex.models.economics.finance import GibratFirmElement
>>> GroupElement('firms', element=GibratFirmElement, size=10, values={'mu': 1.0}).size
10
Source code in src/skcomplex/models/economics/finance/_gibrat_firm.py
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def __init__(
    self,
    label: str,
    visual: VisualSpec | None = None,
    *,
    x: Annotated[float, StateSpec()] = 100.0,
    previous: Annotated[float, StateSpec()] = 100.0,
    mu: Annotated[float, ParameterSpec()] = 1.02,
    sigma: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.05,
) -> None:
    super().__init__(label, visual=visual)
    self.x = x
    self.previous = previous
    self.mu = mu
    self.sigma = sigma

get_growth(agent, model) staticmethod

Return the size multiplied by a growth factor drawn from a normal distribution.

Parameters:

Name Type Description Default
agent GibratFirmView

The firm.

required
model ModelView

The model view of the growth.

required

Returns:

Name Type Description
size float

The size after the step.

Source code in src/skcomplex/models/economics/finance/_gibrat_firm.py
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
@staticmethod
def get_growth(agent: GibratFirmView, model: ModelView) -> float:
    """Return the size multiplied by a growth factor drawn from a normal distribution.

    Args:
        agent:
            The firm.

        model:
            The model view of the growth.

    Returns:
        size:
            The size after the step.
    """
    return agent.x * float(model.rng.normal(agent.mu, agent.sigma))

InterbankElement(label, visual=None, *, h=0.0, status=UNDISTRESSED, value=0.0)

Bases: Element

Bank of the DebtRank cascade, with its level of distress, its state, and its economic value.

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
h Annotated[float, StateSpec(domain=closed_open(0, oo))]

The level of distress, between 0 and 1, where 1 is default.

0.0
status Annotated[float, StateSpec(domain=NN)]

0 for undistressed, 1 for distressed, and 2 for inactive.

UNDISTRESSED
value Annotated[float, StateSpec(domain=closed_open(0, oo))]

The economic value, the bank's share of all interbank lending.

0.0

Examples:

>>> from skcomplex.core.elements import GroupElement
>>> from skcomplex.models.economics.finance import InterbankElement
>>> GroupElement('banks', element=InterbankElement, size=3, values={'h': [1.0, 0.0, 0.0]}).size
3
Source code in src/skcomplex/models/economics/finance/_interbank.py
46
47
48
49
50
51
52
53
54
55
56
57
58
def __init__(
    self,
    label: str,
    visual: VisualSpec | None = None,
    *,
    h: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
    status: Annotated[float, StateSpec(domain=NN)] = UNDISTRESSED,
    value: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
) -> None:
    super().__init__(label, visual=visual)
    self.h = h
    self.status = status
    self.value = value

KestenFirmElement(label, visual=None, *, x=100.0, lambda_low=0.48, lambda_high=1.48, b_low=0.0, b_high=1.0)

Bases: Element

Firm whose size follows Kesten's process, x = lambda x + b, with uniform random lambda and b.

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
x Annotated[float, StateSpec(domain=closed_open(0, oo))]

The initial size.

100.0
lambda_low Annotated[float, ParameterSpec(domain=closed_open(0, oo))]

The lower bound of the multiplicative factor.

0.48
lambda_high Annotated[float, ParameterSpec(domain=closed_open(0, oo))]

The upper bound of the multiplicative factor.

1.48
b_low Annotated[float, ParameterSpec(domain=closed_open(0, oo))]

The lower bound of the additive term.

0.0
b_high Annotated[float, ParameterSpec(domain=closed_open(0, oo))]

The upper bound of the additive term.

1.0

Examples:

>>> from skcomplex.core.elements import GroupElement
>>> from skcomplex.models.economics.finance import KestenFirmElement
>>> GroupElement('firms', element=KestenFirmElement, size=10, values={'b_high': 2.0}).size
10
Source code in src/skcomplex/models/economics/finance/_kesten_firm.py
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
def __init__(
    self,
    label: str,
    visual: VisualSpec | None = None,
    *,
    x: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 100.0,
    lambda_low: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.48,
    lambda_high: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.48,
    b_low: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
    b_high: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
) -> None:
    super().__init__(label, visual=visual)
    self.x = x
    self.lambda_low = lambda_low
    self.lambda_high = lambda_high
    self.b_low = b_low
    self.b_high = b_high

get_growth(agent, model) staticmethod

Return the size multiplied by a random factor plus a random term.

Parameters:

Name Type Description Default
agent KestenFirmView

The firm.

required
model ModelView

The model view of the growth.

required

Returns:

Name Type Description
size float

The size after the step.

Source code in src/skcomplex/models/economics/finance/_kesten_firm.py
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
@staticmethod
def get_growth(agent: KestenFirmView, model: ModelView) -> float:
    """Return the size multiplied by a random factor plus a random term.

    Args:
        agent:
            The firm.

        model:
            The model view of the growth.

    Returns:
        size:
            The size after the step.
    """
    factor = float(model.rng.uniform(agent.lambda_low, agent.lambda_high))
    return factor * agent.x + float(model.rng.uniform(agent.b_low, agent.b_high))

NetworkSaverElement(label, visual=None, *, W=1.0, J=1.0, sigma=0.5, degree=1.0)

Bases: Element

Saver of Bouchaud and Mezard, whose wealth earns random returns and flows to and from its exchange partners.

Parameters:

Name Type Description Default
label str

The label of the saver.

required
visual VisualSpec | None

How the element is drawn, or None for default_visual_.

None
W Annotated[object, StateSpec(domain=open(0, oo))]

The wealth.

1.0
J Annotated[object, ParameterSpec(domain=closed_open(0, oo))]

The exchange rate with the partners.

1.0
sigma Annotated[object, ParameterSpec(domain=closed_open(0, oo))]

The volatility of the returns.

0.5
degree Annotated[object, ParameterSpec(domain=open(0, oo))]

The mean number of exchange partners c, which divides the exchange rate, N - 1 for exchange with every other saver.

1.0

Examples:

>>> from skcomplex.models.economics.finance import NetworkSaverElement
>>> NetworkSaverElement('saver', W=2.0).W
2.0
Source code in src/skcomplex/models/economics/finance/_wealth.py
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
def __init__(
    self,
    label: str,
    visual: VisualSpec | None = None,
    *,
    W: Annotated[object, StateSpec(domain=RealSet.open(0, oo))] = 1.0,
    J: Annotated[object, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
    sigma: Annotated[object, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.5,
    degree: Annotated[object, ParameterSpec(domain=RealSet.open(0, oo))] = 1.0,
) -> None:
    super().__init__(label, visual=visual)
    self.W = W
    self.J = J
    self.sigma = sigma
    self.degree = degree

ReflectingFloorInteraction(label, *, target, c=0.9, phase=None)

Bases: MapInteraction

Lower bound on the sizes of a group at a fraction of their mean before the step, x = max(x, c <x_{t-1}>).

It is a map of the agent form that raises every size below the floor to the floor, the mean of the states previous of the group times c.

Parameters:

Name Type Description Default
label str

The label of the interaction.

required
target Annotated[str, RoleSpec()]

The path of the group of firms.

required
c Annotated[float, ParameterSpec(domain=closed_open(0, oo))]

The fraction of the previous mean size below which no size falls.

0.9
phase str | None

The phase of the step the floor acts in, or None for the default phase.

None

Examples:

>>> from skcomplex.models.economics.finance import ReflectingFloorInteraction
>>> ReflectingFloorInteraction('floor', target='firms', c=0.5).check().c
0.5
Source code in src/skcomplex/models/economics/finance/_reflecting_floor.py
43
44
45
46
47
48
49
50
51
52
def __init__(
    self,
    label: str,
    *,
    target: Annotated[str, RoleSpec()],
    c: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.9,
    phase: str | None = None,
) -> None:
    super().__init__(label, target=target, state='x', function=self._get_floored, form='agent', phase=phase)
    self.c = c

StockTraderElement(label, visual=VisualSpec('agents', {'color_by': 'strategy', 'color_range': (0.0, 1.0)}), *, strategy=0.0, money=1000.0, shares=10.0, bid=0.0, ask=0.0, demand=0.0, supply=0.0, wealth=0.0, gain=0.0, fundamental=100.0, reversion=0.2, extrapolation=1.0, noise=0.02)

Bases: Element

Trader who forecasts the price as a fundamentalist or a chartist, quotes a limit order, and imitates.

A fundamentalist expects the price p to revert to the fundamental value F, p + reversion (F - p), and a chartist expects the last change to go on, p + extrapolation (p - p'), both with a relative noise. The trader bids its expectation for one share when it is above the price and it can pay, and asks its expectation for one share when it is below and it holds one. After trading, an ImitationInteraction of rule 'better' lets a trader whose wealth grew less than that of a random neighbour copy the neighbour's strategy.

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': 'strategy', 'color_range': (0.0, 1.0)})
strategy Annotated[float, StateSpec(domain=closed_open(0, oo))]

0 for a fundamentalist, 1 for a chartist.

0.0
money Annotated[float, StateSpec()]

The money of the trader.

1000.0
shares Annotated[float, StateSpec()]

The shares of the trader.

10.0
bid Annotated[float, StateSpec(domain=closed_open(0, oo))]

The limit price of a buy order.

0.0
ask Annotated[float, StateSpec(domain=closed_open(0, oo))]

The limit price of a sell order.

0.0
demand Annotated[float, StateSpec(domain=closed_open(0, oo))]

The shares wanted in the step.

0.0
supply Annotated[float, StateSpec(domain=closed_open(0, oo))]

The shares offered in the step.

0.0
wealth Annotated[float, StateSpec()]

The money plus the shares at the last price.

0.0
gain Annotated[float, StateSpec()]

The change of the wealth in the last step.

0.0
fundamental Annotated[float, ParameterSpec(domain=open(0, oo))]

The fundamental value of the share.

100.0
reversion Annotated[float, ParameterSpec(domain=closed_open(0, oo))]

The share of the gap to the fundamental value a fundamentalist expects to close.

0.2
extrapolation Annotated[float, ParameterSpec(domain=closed_open(0, oo))]

The share of the last change of the price a chartist expects to repeat.

1.0
noise Annotated[float, ParameterSpec(domain=closed_open(0, oo))]

The standard deviation of the relative noise of a forecast.

0.02

Examples:

>>> from skcomplex.core.elements import GroupElement
>>> from skcomplex.models.economics.finance import StockTraderElement
>>> GroupElement('traders', element=StockTraderElement, size=4, values={'strategy': 1.0}).size
4
Source code in src/skcomplex/models/economics/finance/_stock_trader.py
 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
def __init__(
    self,
    label: str,
    visual: VisualSpec | None = VisualSpec('agents', {'color_by': 'strategy', 'color_range': (0.0, 1.0)}),
    *,
    strategy: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
    money: Annotated[float, StateSpec()] = 1000.0,
    shares: Annotated[float, StateSpec()] = 10.0,
    bid: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
    ask: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
    demand: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
    supply: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
    wealth: Annotated[float, StateSpec()] = 0.0,
    gain: Annotated[float, StateSpec()] = 0.0,
    fundamental: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 100.0,
    reversion: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.2,
    extrapolation: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
    noise: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.02,
) -> None:
    super().__init__(label, visual=visual)
    self.strategy = strategy
    self.money = money
    self.shares = shares
    self.bid = bid
    self.ask = ask
    self.demand = demand
    self.supply = supply
    self.wealth = wealth
    self.gain = gain
    self.fundamental = fundamental
    self.reversion = reversion
    self.extrapolation = extrapolation
    self.noise = noise

get_gain(agent, model) staticmethod

Return the change of the wealth valued at the last price, zero before the first valuation.

Parameters:

Name Type Description Default
agent StockTraderView

The trader.

required
model ModelView

The model view of the accounting.

required

Returns:

Name Type Description
gain float

The change of the wealth in the step.

Source code in src/skcomplex/models/economics/finance/_stock_trader.py
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
@staticmethod
def get_gain(agent: StockTraderView, model: ModelView) -> float:
    """Return the change of the wealth valued at the last price, zero before the first valuation.

    Args:
        agent:
            The trader.

        model:
            The model view of the accounting.

    Returns:
        gain:
            The change of the wealth in the step.
    """
    return StockTraderElement.get_wealth(agent, model) - agent.wealth if agent.wealth else 0.0

get_wealth(agent, model) staticmethod

Return the money plus the shares valued at the last price.

Parameters:

Name Type Description Default
agent StockTraderView

The trader.

required
model ModelView

The model view of the accounting.

required

Returns:

Name Type Description
wealth float

The wealth of the trader.

Source code in src/skcomplex/models/economics/finance/_stock_trader.py
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
@staticmethod
def get_wealth(agent: StockTraderView, model: ModelView) -> float:
    """Return the money plus the shares valued at the last price.

    Args:
        agent:
            The trader.

        model:
            The model view of the accounting.

    Returns:
        wealth:
            The wealth of the trader.
    """
    price, _ = StockTraderElement._prices(agent, model)
    return agent.money + agent.shares * price

quote(agent, model)

Forecast the price and quote a limit order for one share.

Parameters:

Name Type Description Default
agent StockTraderView

The trader.

required
model ModelView

The model.

required
Source code in src/skcomplex/models/economics/finance/_stock_trader.py
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
def quote(self, agent: StockTraderView, model: ModelView) -> None:
    """Forecast the price and quote a limit order for one share.

    Args:
        agent:
            The trader.

        model:
            The model.
    """
    price, previous = self._prices(agent, model)
    if agent.strategy:
        expected = heuristics.trend([previous, price], strength=model.parameter('extrapolation'))
    else:
        reversion = 1 - model.parameter('reversion')
        expected = heuristics.fundamental([price], fundamental=model.parameter('fundamental'), reversion=reversion)
    expected = max(LOWEST_PRICE, expected * (1 + model.parameter('noise') * model.rng.normal()))
    buying = expected > price
    new_bid, new_ask = (expected, 0.0) if buying else (0.0, expected)
    agent.bid = new_bid
    agent.ask = new_ask
    agent.demand = float(buying and agent.money >= expected)
    agent.supply = float(not buying and agent.shares >= 1)

arbitrage_traders(n_sites=200, *, length=20.0, D=0.5, A_x=0.0, E=0.5, A_t=0.0)

Return Schwichtenberg's diffusion of a good by random-walk traders, minimally coupled to exchange rates.

The density of the good follows (d_t - A_t) rho = D (d_x - A_x(t))^2 rho on a periodic line, with the exchange rate connection A_x(t) = A_x + E t. The arbitrage opportunity is the field strength F_tx = E. Without it, the traders diffuse the good symmetrically, and with it, they drift at the velocity 2 D A_x(t).

Parameters:

Name Type Description Default
n_sites int

The number of sites of the line.

200
length float

The length of the line.

20.0
D float

The diffusion constant of the traders.

0.5
A_x float

The initial spatial component of the exchange rate connection.

0.0
E float

The arbitrage field strength, the growth rate of the spatial component.

0.5
A_t float

The temporal component of the exchange rate connection.

0.0

Returns:

Name Type Description
system System

The model, with the density of the good market.rho.

Examples:

>>> from skcomplex.models.economics.finance import arbitrage_traders
>>> arbitrage_traders(50).elements[0].values['E']
0.5
Source code in src/skcomplex/models/economics/finance/_arbitrage.py
 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
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
@register_model(
    field='economics/finance',
    tags=('arbitrage', 'diffusion', 'gauge field', 'pde'),
    references=('Schwichtenberg 2020',),
    defaults={'until': 2.0, 'dt': 0.1},
)
def arbitrage_traders(
    n_sites: int = 200,
    *,
    length: float = 20.0,
    D: float = 0.5,
    A_x: float = 0.0,
    E: float = 0.5,
    A_t: float = 0.0,
) -> System:
    """Return Schwichtenberg's diffusion of a good by random-walk traders, minimally coupled to exchange rates.

    The density of the good follows `(d_t - A_t) rho = D (d_x - A_x(t))^2 rho` on a periodic line, with the exchange
    rate connection `A_x(t) = A_x + E t`. The arbitrage opportunity is the field strength `F_tx = E`. Without it, the
    traders diffuse the good symmetrically, and with it, they drift at the velocity `2 D A_x(t)`.

    Args:
        n_sites:
            The number of sites of the line.

        length:
            The length of the line.

        D:
            The diffusion constant of the traders.

        A_x:
            The initial spatial component of the exchange rate connection.

        E:
            The arbitrage field strength, the growth rate of the spatial component.

        A_t:
            The temporal component of the exchange rate connection.

    Returns:
        system:
            The model, with the density of the good `market.rho`.

    Examples:
        >>> from skcomplex.models.economics.finance import arbitrage_traders
        >>> arbitrage_traders(50).elements[0].values['E']
        0.5
    """
    spacing = length / n_sites
    positions = (np.arange(n_sites) + 0.5) * spacing - length / 2
    rho = np.exp(-(positions**2)) / np.sqrt(np.pi)
    values = {'rho': rho, 'D': D, 'A_x': A_x, 'E': E, 'A_t': A_t}
    line = LatticeSpace(n_sites, spacing=spacing, boundary='periodic')
    market = FieldElement('market', element=ArbitrageDensityElement, space=line, values=values)
    transport = RateInteraction('transport', target='market', state='rho', rate=_covariant_diffusion)
    return System(elements=[market], interactions=[transport])

bouchaud_mezard_wealth(n_agents=20, *, graph=None, J=1.0, sigma=0.5, wealth=1.0)

Return Bouchaud and Mezard's wealth dynamics, random multiplicative returns with exchange between neighbours.

Agent i follows the Stratonovich equation dW_i = sigma W_i dB_i + (J / c) sum_j (W_j - W_i) dt over its neighbours j on a network of mean degree c. Without a graph, the agents are the members of a group and every agent exchanges with every other, c = N - 1, the complete graph at a cost linear in the number of agents. A given graph keeps the agents on a field over its edges. Exchange conserves the total wealth, and on the complete graph the relative wealth W_i / <W> approaches a stationary distribution with the Pareto tail P(W_i / <W> > w) ~ w^-mu of exponent mu = 1 + 2 J / sigma^2.

Parameters:

Name Type Description Default
n_agents int

The number of agents, used when no graph is given.

20
graph object

The NetworkX graph of the exchanges, or None for exchange with every other of n_agents agents.

None
J float

The exchange rate between agents.

1.0
sigma float

The volatility of the returns.

0.5
wealth float | Sequence[float]

The initial wealth of every agent, or one value per agent.

1.0

Returns:

Name Type Description
system System

The model, with the wealth agents.W and the mean wealth mean_wealth.

Examples:

>>> from skcomplex.models.economics.finance import bouchaud_mezard_wealth
>>> bouchaud_mezard_wealth(10).elements[0].values['J']
1.0
Source code in src/skcomplex/models/economics/finance/_wealth.py
 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
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
@register_model(
    field='economics/finance',
    tags=('wealth', 'sde', 'network', 'pareto'),
    references=('Yakovenko and Rosser 2009', 'Bouchaud and Mezard 2000'),
    defaults={'until': 1.0, 'dt': 0.1},
    observables=lambda system: [ObservableSpec('mean_wealth', group='agents', variable='W')],
)
def bouchaud_mezard_wealth(
    n_agents: int = 20,
    *,
    graph: object = None,
    J: float = 1.0,
    sigma: float = 0.5,
    wealth: float | Sequence[float] = 1.0,
) -> System:
    """Return Bouchaud and Mezard's wealth dynamics, random multiplicative returns with exchange between neighbours.

    Agent `i` follows the Stratonovich equation `dW_i = sigma W_i dB_i + (J / c) sum_j (W_j - W_i) dt` over its
    neighbours `j` on a network of mean degree `c`. Without a graph, the agents are the members of a group and every
    agent exchanges with every other, `c = N - 1`, the complete graph at a cost linear in the number of agents. A given
    graph keeps the agents on a field over its edges. Exchange conserves the total wealth, and on the complete graph
    the relative wealth `W_i / <W>` approaches a stationary distribution with the Pareto tail
    `P(W_i / <W> > w) ~ w^-mu` of exponent `mu = 1 + 2 J / sigma^2`.

    Args:
        n_agents:
            The number of agents, used when no graph is given.

        graph:
            The NetworkX graph of the exchanges, or `None` for exchange with every other of `n_agents` agents.

        J:
            The exchange rate between agents.

        sigma:
            The volatility of the returns.

        wealth:
            The initial wealth of every agent, or one value per agent.

    Returns:
        system:
            The model, with the wealth `agents.W` and the mean wealth `mean_wealth`.

    Examples:
        >>> from skcomplex.models.economics.finance import bouchaud_mezard_wealth
        >>> bouchaud_mezard_wealth(10).elements[0].values['J']
        1.0
    """
    returns = RateInteraction('returns', target='agents', state='W', noise=_returns, convention='stratonovich')
    if graph is None:
        values = {'W': wealth, 'J': J, 'sigma': sigma, 'degree': n_agents - 1}
        agents: FieldElement | GroupElement = GroupElement(
            'agents',
            element=NetworkSaverElement,
            size=n_agents,
            values=values,
        )
        mode = 'group'
    else:
        degree = 2 * nx.number_of_edges(graph) / nx.number_of_nodes(graph)
        values = {'W': wealth, 'J': J, 'sigma': sigma, 'degree': degree}
        agents = FieldElement('agents', element=NetworkSaverElement, space=NetworkSpace(graph), values=values)
        mode = 'edges'
    exchange = CouplingInteraction('exchange', target='agents', read='W', write='W', function=_exchange, mode=mode)
    return System(elements=[agents], interactions=[exchange, returns])

correlated_geometric_brownian_motion(prices=(100.0, 100.0), *, mu=(0.05, 0.08), sigma=(0.2, 0.3), rho=0.7)

Return two assets whose prices follow geometric Brownian motions driven by correlated Wiener processes.

A WienerProcessSpec of dimension 2 with the covariance [[1, rho], [rho, 1]] drives the two prices, so their log returns over a short time have the correlation rho.

Parameters:

Name Type Description Default
prices tuple[float, float]

The initial prices of the two assets.

(100.0, 100.0)
mu tuple[float, float]

The drifts of the two assets.

(0.05, 0.08)
sigma tuple[float, float]

The volatilities of the two assets.

(0.2, 0.3)
rho float

The correlation of the two Wiener processes, from -1 to 1.

0.7

Returns:

Name Type Description
system System

The model, with the assets first and second, whose motions share the Wiener processes market.

Examples:

>>> from skcomplex.models.economics.finance import correlated_geometric_brownian_motion
>>> correlated_geometric_brownian_motion(rho=0.5).interactions[1].process.covariance
[[1.0, 0.5], [0.5, 1.0]]
Source code in src/skcomplex/models/economics/finance/_correlated_assets.py
14
15
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
@register_model(
    field='economics/finance',
    tags=('asset', 'sde', 'correlated noise'),
    references=('Black and Scholes 1973', 'Merton 1973'),
    defaults={'until': 1.0, 'dt': 0.01, 'solver': EulerMaruyamaSolver(dt=1e-3)},
)
def correlated_geometric_brownian_motion(
    prices: tuple[float, float] = (100.0, 100.0),
    *,
    mu: tuple[float, float] = (0.05, 0.08),
    sigma: tuple[float, float] = (0.2, 0.3),
    rho: float = 0.7,
) -> System:
    """Return two assets whose prices follow geometric Brownian motions driven by correlated Wiener processes.

    A `WienerProcessSpec` of dimension 2 with the covariance `[[1, rho], [rho, 1]]` drives the two prices, so their log
    returns over a short time have the correlation `rho`.

    Args:
        prices:
            The initial prices of the two assets.

        mu:
            The drifts of the two assets.

        sigma:
            The volatilities of the two assets.

        rho:
            The correlation of the two Wiener processes, from -1 to 1.

    Returns:
        system:
            The model, with the assets `first` and `second`, whose motions share the Wiener processes `market`.

    Examples:
        >>> from skcomplex.models.economics.finance import correlated_geometric_brownian_motion
        >>> correlated_geometric_brownian_motion(rho=0.5).interactions[1].process.covariance
        [[1.0, 0.5], [0.5, 1.0]]
    """
    labels = ('first', 'second')
    market = WienerProcessSpec('market', dimension=2, covariance=[[1.0, rho], [rho, 1.0]])
    return System(
        elements=[AssetElement(label, price=price) for label, price in zip(labels, prices, strict=True)],
        interactions=[
            GeometricBrownianMotionInteraction(
                f'{label}_motion',
                target=label,
                process=market,
                index=index,
                mu=drift,
                sigma=volatility,
            )
            for index, (label, drift, volatility) in enumerate(zip(labels, mu, sigma, strict=True))
        ],
    )

debtrank(liabilities=None, capital=None, *, distressed=(0,), psi=1.0, n_banks=10, density=0.3, random_state=0)

Return the DebtRank cascade of distress from a set of banks through their interbank loans.

liabilities[i][j] is the loan from bank j to bank i, which j loses when i defaults, and the impact of i on j is min(1, liabilities[i][j] / capital[j]). The banks in distressed start with distress psi. The DebtRank is the final value at risk minus the initial one.

Parameters:

Name Type Description Default
liabilities Sequence[Sequence[float]] | None

The interbank liabilities, or None for a random network of n_banks banks.

None
capital Sequence[float] | None

The capital of every bank, or None for random capitals.

None
distressed Sequence[int]

The indices of the banks distressed at the start.

(0,)
psi float

The initial distress of those banks, where 1 is default.

1.0
n_banks int

The number of banks of a random network.

10
density float

The probability of a loan between two banks of a random network.

0.3
random_state int | Generator | None

The seed or generator of a random network.

0

Returns:

Name Type Description
system System

The model, with the group banks and the distress weighted by economic value value_at_risk.

Examples:

>>> from skcomplex.models.economics.finance import debtrank
>>> debtrank([[0.0, 5.0], [0.0, 0.0]], [10.0, 10.0]).elements[0].check().prepare().size_
2
Source code in src/skcomplex/models/economics/finance/_contagion.py
 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
@register_model(
    field='economics/finance',
    tags=('systemic risk', 'network', 'contagion', 'agents'),
    references=('Thurner et al. 2018', 'Battiston et al. 2012'),
    defaults={'until': 10},
    observables=lambda system: [ObservableSpec('value_at_risk', _value_at_risk, group='banks', reduce='sum')],
)
def debtrank(
    liabilities: Sequence[Sequence[float]] | None = None,
    capital: Sequence[float] | None = None,
    *,
    distressed: Sequence[int] = (0,),
    psi: float = 1.0,
    n_banks: int = 10,
    density: float = 0.3,
    random_state: int | np.random.Generator | None = 0,
) -> System:
    """Return the DebtRank cascade of distress from a set of banks through their interbank loans.

    `liabilities[i][j]` is the loan from bank `j` to bank `i`, which `j` loses when `i` defaults, and the impact of `i`
    on `j` is `min(1, liabilities[i][j] / capital[j])`. The banks in `distressed` start with distress `psi`. The
    DebtRank is the final value at risk minus the initial one.

    Args:
        liabilities:
            The interbank liabilities, or `None` for a random network of `n_banks` banks.

        capital:
            The capital of every bank, or `None` for random capitals.

        distressed:
            The indices of the banks distressed at the start.

        psi:
            The initial distress of those banks, where 1 is default.

        n_banks:
            The number of banks of a random network.

        density:
            The probability of a loan between two banks of a random network.

        random_state:
            The seed or generator of a random network.

    Returns:
        system:
            The model, with the group `banks` and the distress weighted by economic value `value_at_risk`.

    Examples:
        >>> from skcomplex.models.economics.finance import debtrank
        >>> debtrank([[0.0, 5.0], [0.0, 0.0]], [10.0, 10.0]).elements[0].check().prepare().size_
        2
    """
    if liabilities is None or capital is None:
        matrix, capitals = _random_exposures(n_banks, density, np.random.default_rng(random_state))
    else:
        matrix, capitals = np.asarray(liabilities, dtype=float), np.asarray(capital, dtype=float)
    size = len(matrix)
    h = np.zeros(size)
    h[list(distressed)] = psi
    status = np.where(h > 0, DISTRESSED, UNDISTRESSED).astype(float)
    banks = GroupElement(
        'banks',
        element=InterbankElement,
        size=size,
        values={'h': h, 'status': status, 'value': _economic_values(matrix)},
    )
    return System(
        elements=[banks],
        interactions=[DebtRankCascadeInteraction('cascade', banks='banks', impact=_impact(matrix, capitals).tolist())],
    )

geometric_brownian_motion(price=100.0, *, mu=0.05, sigma=0.2)

Return an asset whose price follows geometric Brownian motion.

Parameters:

Name Type Description Default
price float

The initial price.

100.0
mu float

The drift.

0.05
sigma float

The volatility.

0.2

Returns:

Name Type Description
system System

The model, with the asset asset.

Examples:

>>> from skcomplex.models.economics.finance import geometric_brownian_motion
>>> geometric_brownian_motion(sigma=0.3).interactions[0].sigma
0.3
Source code in src/skcomplex/models/economics/finance/_assets.py
13
14
15
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
@register_model(
    field='economics/finance',
    tags=('asset', 'sde'),
    references=('Black and Scholes 1973',),
    defaults={'until': 1.0, 'dt': 0.01},
)
def geometric_brownian_motion(price: float = 100.0, *, mu: float = 0.05, sigma: float = 0.2) -> System:
    """Return an asset whose price follows geometric Brownian motion.

    Args:
        price:
            The initial price.

        mu:
            The drift.

        sigma:
            The volatility.

    Returns:
        system:
            The model, with the asset `asset`.

    Examples:
        >>> from skcomplex.models.economics.finance import geometric_brownian_motion
        >>> geometric_brownian_motion(sigma=0.3).interactions[0].sigma
        0.3
    """
    return System(
        elements=[AssetElement('asset', price=price)],
        interactions=[GeometricBrownianMotionInteraction('gbm', target='asset', mu=mu, sigma=sigma)],
    )

gibrat_process(n_firms=200, *, x=100.0, mu=1.02, sigma=0.05)

Return firms that grow by Gibrat's law of proportionate effect, which makes their sizes lognormal.

Parameters:

Name Type Description Default
n_firms int

The number of firms.

200
x float

The initial size of every firm.

100.0
mu float

The mean of the growth factor.

1.02
sigma float

The standard deviation of the growth factor.

0.05

Returns:

Name Type Description
system System

The model, with the group firms, the mean size mean_size, and the largest size max_size.

Examples:

>>> from skcomplex.models.economics.finance import gibrat_process
>>> gibrat_process(10).elements[0].check().prepare().size_
10
Source code in src/skcomplex/models/economics/finance/_growth_processes.py
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
@register_model(
    field='economics/finance',
    tags=('multiplicative process', 'lognormal', 'agents'),
    references=('Caiani et al. 2016', 'Thurner et al. 2018', 'Gibrat 1931'),
    defaults={'until': 100},
    observables=lambda system: _size_observables(),
)
def gibrat_process(
    n_firms: int = 200,
    *,
    x: float = 100.0,
    mu: float = 1.02,
    sigma: float = 0.05,
) -> System:
    """Return firms that grow by Gibrat's law of proportionate effect, which makes their sizes lognormal.

    Args:
        n_firms:
            The number of firms.

        x:
            The initial size of every firm.

        mu:
            The mean of the growth factor.

        sigma:
            The standard deviation of the growth factor.

    Returns:
        system:
            The model, with the group `firms`, the mean size `mean_size`, and the largest size `max_size`.

    Examples:
        >>> from skcomplex.models.economics.finance import gibrat_process
        >>> gibrat_process(10).elements[0].check().prepare().size_
        10
    """
    firms = GroupElement(
        'firms',
        element=GibratFirmElement,
        size=n_firms,
        values={'x': x, 'previous': x, 'mu': mu, 'sigma': sigma},
    )
    return System(elements=[firms], interactions=_get_gibrat_growth(), time=DiscreteTime())

kesten_process(n_firms=200, *, x=100.0, lambda_low=0.48, lambda_high=1.48, b_low=0.0, b_high=1.0)

Return firms whose sizes follow Kesten's process, a random multiplicative factor plus a random positive term.

When the factor shrinks sizes on average, the additive term reinjects them, and the stationary distribution has a power-law tail.

Parameters:

Name Type Description Default
n_firms int

The number of firms.

200
x float

The initial size of every firm.

100.0
lambda_low float

The lower bound of the multiplicative factor.

0.48
lambda_high float

The upper bound of the multiplicative factor.

1.48
b_low float

The lower bound of the additive term.

0.0
b_high float

The upper bound of the additive term.

1.0

Returns:

Name Type Description
system System

The model, with the group firms, the mean size mean_size, and the largest size max_size.

Examples:

>>> from skcomplex.models.economics.finance import kesten_process
>>> kesten_process(10).elements[0].check().prepare().size_
10
Source code in src/skcomplex/models/economics/finance/_growth_processes.py
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
@register_model(
    field='economics/finance',
    tags=('multiplicative process', 'power law', 'agents'),
    references=('Caiani et al. 2016', 'Thurner et al. 2018', 'Kesten 1973'),
    defaults={'until': 100},
    observables=lambda system: _size_observables(),
)
def kesten_process(
    n_firms: int = 200,
    *,
    x: float = 100.0,
    lambda_low: float = 0.48,
    lambda_high: float = 1.48,
    b_low: float = 0.0,
    b_high: float = 1.0,
) -> System:
    """Return firms whose sizes follow Kesten's process, a random multiplicative factor plus a random positive term.

    When the factor shrinks sizes on average, the additive term reinjects them, and the stationary distribution has a
    power-law tail.

    Args:
        n_firms:
            The number of firms.

        x:
            The initial size of every firm.

        lambda_low:
            The lower bound of the multiplicative factor.

        lambda_high:
            The upper bound of the multiplicative factor.

        b_low:
            The lower bound of the additive term.

        b_high:
            The upper bound of the additive term.

    Returns:
        system:
            The model, with the group `firms`, the mean size `mean_size`, and the largest size `max_size`.

    Examples:
        >>> from skcomplex.models.economics.finance import kesten_process
        >>> kesten_process(10).elements[0].check().prepare().size_
        10
    """
    firms = GroupElement(
        'firms',
        element=KestenFirmElement,
        size=n_firms,
        values={'x': x, 'lambda_low': lambda_low, 'lambda_high': lambda_high, 'b_low': b_low, 'b_high': b_high},
    )
    growth = MapInteraction('growth', target='firms', state='x', function=KestenFirmElement.get_growth, form='agent')
    return System(elements=[firms], interactions=[growth], time=DiscreteTime())

multiplicative_process(n_firms=200, *, x=100.0, mu=1.02, sigma=0.05, c=0.9)

Return Levy and Solomon's multiplicative process, whose floor at a fraction of the mean makes a power-law tail.

Every size is multiplied by a normal growth factor and then raised to c times the mean size of the step before if it fell below it. With c = 0 the process is Gibrat's.

Parameters:

Name Type Description Default
n_firms int

The number of firms.

200
x float

The initial size of every firm.

100.0
mu float

The mean of the growth factor.

1.02
sigma float

The standard deviation of the growth factor.

0.05
c float

The fraction of the previous mean size below which no size falls.

0.9

Returns:

Name Type Description
system System

The model, with the group firms, the mean size mean_size, and the largest size max_size.

Examples:

>>> from skcomplex.models.economics.finance import multiplicative_process
>>> multiplicative_process(c=0.5).interactions[-1].c
0.5
Source code in src/skcomplex/models/economics/finance/_growth_processes.py
 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
124
125
126
127
128
129
130
131
132
133
134
135
136
@register_model(
    field='economics/finance',
    tags=('multiplicative process', 'power law', 'agents'),
    references=('Caiani et al. 2016', 'Thurner et al. 2018', 'Levy and Solomon 1996'),
    defaults={'until': 100},
    observables=lambda system: _size_observables(),
)
def multiplicative_process(
    n_firms: int = 200,
    *,
    x: float = 100.0,
    mu: float = 1.02,
    sigma: float = 0.05,
    c: float = 0.9,
) -> System:
    """Return Levy and Solomon's multiplicative process, whose floor at a fraction of the mean makes a power-law tail.

    Every size is multiplied by a normal growth factor and then raised to `c` times the mean size of the step before
    if it fell below it. With `c = 0` the process is Gibrat's.

    Args:
        n_firms:
            The number of firms.

        x:
            The initial size of every firm.

        mu:
            The mean of the growth factor.

        sigma:
            The standard deviation of the growth factor.

        c:
            The fraction of the previous mean size below which no size falls.

    Returns:
        system:
            The model, with the group `firms`, the mean size `mean_size`, and the largest size `max_size`.

    Examples:
        >>> from skcomplex.models.economics.finance import multiplicative_process
        >>> multiplicative_process(c=0.5).interactions[-1].c
        0.5
    """
    firms = GroupElement(
        'firms',
        element=GibratFirmElement,
        size=n_firms,
        values={'x': x, 'previous': x, 'mu': mu, 'sigma': sigma},
    )
    return System(
        elements=[firms],
        interactions=[*_get_gibrat_growth(), ReflectingFloorInteraction('floor', target='firms', c=c)],
        time=DiscreteTime(),
    )

order_book(width=20, height=20, *, chartists=0.3, fundamental=100.0, reversion=0.2, extrapolation=1.0, noise=0.02, imitation=0.1, random_state=0)

Return a stock market of fundamentalists and chartists on a grid who trade in a limit order book and imitate.

Every step every trader forecasts the price, as a fundamentalist who expects it to revert to the fundamental value or as a chartist who expects the last change to go on, and quotes a limit order for one share at its forecast. The order book matches the orders with price-time priority, and every trade moves money and a share between the two traders. Then traders copy the strategy of a neighbour whose wealth grew more. Fundamentalists hold the price near the fundamental value, chartists amplify its swings, and imitation lets the share of chartists wander, so the returns come in bursts with fat tails (Caiani et al. 2016 ch. 3).

Parameters:

Name Type Description Default
width int

The number of columns of the grid, with one trader per cell.

20
height int

The number of rows of the grid.

20
chartists float

The initial share of chartists.

0.3
fundamental float

The fundamental value of the share.

100.0
reversion float

The share of the gap to the fundamental value a fundamentalist expects to close.

0.2
extrapolation float

The share of the last change of the price a chartist expects to repeat.

1.0
noise float

The standard deviation of the relative noise of a forecast.

0.02
imitation float

The probability that a trader looks at a neighbour in a step.

0.1
random_state int | Generator | None

The seed or generator of the initial strategies.

0

Returns:

Name Type Description
system System

The model, with the group traders and the book book, whose reporters are book.price and book.volume, and the imitation imitation of the strategies of more successful neighbours.

Examples:

>>> from skcomplex.models.economics.finance import order_book
>>> order_book(5, 5).elements[0].check().prepare().size_
25
Source code in src/skcomplex/models/economics/finance/_order_book.py
 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
124
125
126
127
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
@register_model(
    field='economics/finance',
    tags=('stock market', 'agents', 'order book', 'imitation', 'fundamentalists', 'chartists'),
    references=('Caiani et al. 2016 ch. 3', 'Lux and Marchesi 1999'),
    defaults={'until': 200, 'invariants': {f'total_{name}': 1e-6 for name in HOLDINGS}},
    observables=_order_book_observables,
)
def order_book(
    width: int = 20,
    height: int = 20,
    *,
    chartists: float = 0.3,
    fundamental: float = 100.0,
    reversion: float = 0.2,
    extrapolation: float = 1.0,
    noise: float = 0.02,
    imitation: float = 0.1,
    random_state: int | np.random.Generator | None = 0,
) -> System:
    """Return a stock market of fundamentalists and chartists on a grid who trade in a limit order book and imitate.

    Every step every trader forecasts the price, as a fundamentalist who expects it to revert to the fundamental value
    or as a chartist who expects the last change to go on, and quotes a limit order for one share at its forecast. The
    order book matches the orders with price-time priority, and every trade moves money and a share between the two
    traders. Then traders copy the strategy of a neighbour whose wealth grew more. Fundamentalists hold the price near
    the fundamental value, chartists amplify its swings, and imitation lets the share of chartists wander, so the
    returns come in bursts with fat tails (Caiani et al. 2016 ch. 3).

    Args:
        width:
            The number of columns of the grid, with one trader per cell.

        height:
            The number of rows of the grid.

        chartists:
            The initial share of chartists.

        fundamental:
            The fundamental value of the share.

        reversion:
            The share of the gap to the fundamental value a fundamentalist expects to close.

        extrapolation:
            The share of the last change of the price a chartist expects to repeat.

        noise:
            The standard deviation of the relative noise of a forecast.

        imitation:
            The probability that a trader looks at a neighbour in a step.

        random_state:
            The seed or generator of the initial strategies.

    Returns:
        system:
            The model, with the group `traders` and the book `book`, whose reporters are `book.price` and
            `book.volume`, and the imitation `imitation` of the strategies of more successful neighbours.

    Examples:
        >>> from skcomplex.models.economics.finance import order_book
        >>> order_book(5, 5).elements[0].check().prepare().size_
        25
    """
    rng = np.random.default_rng(random_state)
    cells = [(x, y) for x in range(width) for y in range(height)]
    traders = GroupElement(
        'traders',
        element=StockTraderElement,
        size=len(cells),
        values={
            'strategy': (rng.random(len(cells)) < chartists).astype(float),
            'money': 10 * fundamental,
            'shares': 10.0,
            'wealth': 20 * fundamental,
            'fundamental': fundamental,
            'reversion': reversion,
            'extrapolation': extrapolation,
            'noise': noise,
        },
        location=cells,
    )
    accounts = [
        MapInteraction(
            f'{name}_account',
            target='traders',
            state=name,
            function=function,
            form='agent',
            phase='account',
        )
        for name, function in (('gain', StockTraderElement.get_gain), ('wealth', StockTraderElement.get_wealth))
    ]
    return System(
        elements=[traders],
        interactions=[
            BookInteraction(
                'book',
                requesters='traders',
                providers='traders',
                rule=ContinuousMatchingSpec(),
                opening=fundamental,
            ),
            *accounts,
            ImitationInteraction(
                'imitation',
                target='traders',
                state='strategy',
                payoff='gain',
                rule='better',
                form='agent',
                probability=imitation,
                phase='imitation',
            ),
        ],
        space=LatticeSpace((width, height), boundary='periodic', neighborhood='von_neumann'),
        time=DiscreteTime(
            schedule=PhasesScheduleSpec(
                [
                    PhaseSpec('quote'),
                    PhaseSpec('trade'),
                    PhaseSpec('account'),
                    PhaseSpec('imitation', activation='simultaneous'),
                ],
            ),
        ),
    )