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.systems.economics.finance import ArbitrageDensityElement
>>> ArbitrageDensityElement('market', E=1.0).E
1.0

Source code in src/skcomplex/systems/economics/finance/_arbitrage.py

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.systems.economics.finance import AssetElement
>>> AssetElement('stock', price=100.0).price
100.0

Source code in src/skcomplex/systems/economics/common/_asset.py

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

BalanceSheetElement(label, visual=VisualSpec('agents', {'color_by': 'status'}), *, external=0.0, equity=0.0, payment=0.0, status=SOLVENT)

Bases: Element

Bank of an interbank network, with its external assets, its equity, its payments, and its state of default.

The bank is a passive agent that the contagion of the system updates, a solvency cascade through its equity or a clearing of payments through its payment.

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': 'status'})
external	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The external assets.	0.0
equity	Annotated[float, StateSpec()]	The equity, the assets minus the liabilities.	0.0
payment	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The payment the bank makes on its interbank liabilities.	0.0
status	Annotated[float, StateSpec(domain=NN)]	0 for solvent, 1 for defaulting in this step, and 2 for defaulted.	SOLVENT

Examples:

>>> from skcomplex.core.components.elements import GroupElement
>>> from skcomplex.systems.economics.finance import BalanceSheetElement
>>> GroupElement('banks', element=BalanceSheetElement, size=3, values={'equity': 4.0}).size
3

Source code in src/skcomplex/systems/economics/common/_balance_sheet.py

def __init__(
    self,
    label: str,
    visual: VisualSpec | None = VisualSpec('agents', {'color_by': 'status'}),
    *,
    external: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
    equity: Annotated[float, StateSpec()] = 0.0,
    payment: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
    status: Annotated[float, StateSpec(domain=NN)] = SOLVENT,
) -> None:
    super().__init__(label, visual=visual)
    self.external = external
    self.equity = equity
    self.payment = payment
    self.status = status

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[SystemView]

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, SystemView], None] | None	A function of every trade and the system view, run after the transfers, or None.	None
opening	float	The price before the first trade.	100.0

Examples:

>>> from skcomplex.systems.economics.finance import BookInteraction
>>> BookInteraction('book', requesters='traders', providers='traders').reports
('price', 'volume')

Source code in src/skcomplex/systems/economics/finance/_order_book.py

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, SystemView], 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.systems.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/systems/economics/finance/_contagion.py

def __init__(
    self,
    label: str,
    *,
    banks: Annotated[str, RoleSpec()],
    impact: Sequence[Sequence[float]] = (),
) -> None:
    super().__init__(label)
    self.banks = banks
    self.impact = impact

spread(system)

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

Parameters:

Name	Type	Description	Default
system	SystemView	The system.	required

Source code in src/skcomplex/systems/economics/finance/_contagion.py

def spread(self, system: SystemView) -> None:
    """Advance the distress and the states of every bank by one step.

    Args:
        system:
            The system.
    """
    banks = sorted(system.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)

DefaultCascadeInteraction(label, *, rule=None, phase=None, banks, liabilities=(), threshold=0.0, recovery=0.0)

Bases: FeedbackInteraction[SystemView]

Synchronous cascade of defaults, in which the creditors of a defaulting bank write off their claims on it.

Every step the banks that defaulted in the last step pass their losses on: every creditor j of a defaulting bank i loses (1 - recovery) L_ij of its equity, and a solvent bank whose equity falls below the threshold defaults.

Parameters:

Name	Type	Description	Default
label	str	The label of the cascade.	required
rule	None	Unused, since the cascade overrides act.	None
phase	str | None	The phase the cascade acts in, or None for the default phase.	None
banks	Annotated[str, RoleSpec()]	The path of the group of banks.	required
liabilities	Sequence[Sequence[float]]	The interbank liabilities, liabilities[i][j] owed by bank i to bank j.	()
threshold	Annotated[float, ParameterSpec()]	The equity below which a bank defaults.	0.0
recovery	Annotated[float, ParameterSpec(domain=closed(0, 1))]	The share of a claim on a defaulted bank that its creditor recovers.	0.0

Examples:

>>> from skcomplex.systems.economics.finance import DefaultCascadeInteraction
>>> DefaultCascadeInteraction('cascade', banks='banks').reports
('defaults', 'losses')

Source code in src/skcomplex/systems/economics/finance/_solvency.py

def __init__(
    self,
    label: str,
    *,
    rule: None = None,
    phase: str | None = None,
    banks: Annotated[str, RoleSpec()],
    liabilities: Sequence[Sequence[float]] = (),
    threshold: Annotated[float, ParameterSpec()] = 0.0,
    recovery: Annotated[float, ParameterSpec(domain=RealSet.closed(0, 1))] = 0.0,
) -> None:
    super().__init__(label, rule=rule, phase=phase)
    self.banks = banks
    self.liabilities = liabilities
    self.threshold = threshold
    self.recovery = recovery

act(system)

Pass on the losses of the defaulting banks and default the banks below the threshold.

Parameters:

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

Source code in src/skcomplex/systems/economics/finance/_solvency.py

def act(self, system: SystemView) -> None:
    """Pass on the losses of the defaulting banks and default the banks below the threshold.

    Args:
        system:
            The system view of the cascade.
    """
    banks = sorted(system.agents('banks', BalanceSheetView), key=lambda bank: bank.id)
    status = np.array([bank.status for bank in banks])
    equity = np.array([bank.equity for bank in banks])
    defaulting = status == DEFAULTING
    losses = (1 - system.parameter('recovery')) * np.asarray(self.liabilities, dtype=float).T @ defaulting
    equity = equity - losses
    hit = (status == SOLVENT) & (equity < system.parameter('threshold'))
    status = np.where(defaulting, DEFAULTED, np.where(hit, DEFAULTING, status))
    for bank, value, code in zip(banks, equity, status, strict=True):
        bank.equity, bank.status = float(value), float(code)
    system.report('defaults', float(np.sum(status != SOLVENT)))
    system.report('losses', float(losses.sum()))

ExchangeLinkElement(label, visual=None, *, A=0.0, E=0.0)

Bases: Element

Link of a lattice with an exchange-rate potential A and its rate of change E.

Parameters:

Name	Type	Description	Default
label	str	The label of the link or the group on the links.	required
visual	VisualSpec | None	How the link is drawn, or None for default_visual_.	None
A	Annotated[float, StateSpec()]	The initial exchange-rate potential.	0.0
E	Annotated[float, StateSpec()]	The initial value of F_0i of Schwichtenberg (2020 Eq. 2.22), the rate of change of the potential.	0.0

Attributes:

Name	Type	Description
default_visual_		The visual of the link when it sets none.

Examples:

>>> from skcomplex.systems.economics.finance import ExchangeLinkElement
>>> ExchangeLinkElement('link', A=0.5).A
0.5

Source code in src/skcomplex/systems/economics/finance/_gauge_arbitrage.py

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

FireSaleInteraction(label, *, rule=None, phase=None, banks, assets, holdings=(), liabilities=(), alpha=1.0, leverage=20.0, selling='threshold')

Bases: FeedbackInteraction[SystemView]

Banks that sell common assets on losses, and the market impact that lowers the prices of what they sell.

Every step the banks value their holdings at the current prices. Under the rule 'threshold' a bank whose losses exhaust its equity defaults and liquidates its whole portfolio, and otherwise stays passive. Under 'leverage' a bank sells the same share of every holding to bring its assets back to leverage times its equity, and liquidates everything when its equity is gone. The proceeds become cash, and the price of every asset falls to p0 exp(-alpha x), where x is the share of the asset liquidated so far and p0 its price after the shock.

Parameters:

Name	Type	Description	Default
label	str	The label of the fire sales.	required
rule	None	Unused, since the fire sales override act.	None
phase	str | None	The phase the fire sales act in, or None for the default phase.	None
banks	Annotated[str, RoleSpec()]	The path of the group of banks.	required
assets	Annotated[str, RoleSpec()]	The path of the group of assets.	required
holdings	tuple[tuple[float, ...], ...]	The initial holdings, holdings[i][a] the units of asset a that bank i holds.	()
liabilities	tuple[float, ...]	The liabilities of every bank.	()
alpha	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The market impact.	1.0
leverage	Annotated[float, ParameterSpec(domain=open(1, oo))]	The target ratio of assets to equity of the rule 'leverage'.	20.0
selling	str	The rule of the banks, 'threshold' or 'leverage'.	'threshold'

Raises:

Type	Description
ValidationError	The rule is unknown, raised by prepare.

Examples:

>>> from skcomplex.systems.economics.finance import FireSaleInteraction
>>> FireSaleInteraction('sales', banks='banks', assets='assets').reports
('defaults', 'sold')

Source code in src/skcomplex/systems/economics/finance/_fire_sales.py

def __init__(
    self,
    label: str,
    *,
    rule: None = None,
    phase: str | None = None,
    banks: Annotated[str, RoleSpec()],
    assets: Annotated[str, RoleSpec()],
    holdings: tuple[tuple[float, ...], ...] = (),
    liabilities: tuple[float, ...] = (),
    alpha: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
    leverage: Annotated[float, ParameterSpec(domain=RealSet.open(1, oo))] = 20.0,
    selling: str = 'threshold',
) -> None:
    super().__init__(label, rule=rule, phase=phase)
    self.banks = banks
    self.assets = assets
    self.holdings = holdings
    self.liabilities = liabilities
    self.alpha = alpha
    self.leverage = leverage
    self.selling = selling

act(system)

Value the portfolios, sell, and move the prices by the market impact.

Parameters:

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

Source code in src/skcomplex/systems/economics/finance/_fire_sales.py

def act(self, system: SystemView) -> None:
    """Value the portfolios, sell, and move the prices by the market impact.

    Args:
        system:
            The system view of the fire sales.
    """
    banks = sorted(system.agents('banks', BalanceSheetView), key=lambda bank: bank.id)
    assets = sorted(system.agents('assets', MarketAssetView), key=lambda asset: asset.id)
    initial = np.asarray(self.holdings, dtype=float)
    holdings = np.asarray(system.storage.get('holdings', initial))
    prices = np.array([asset.price for asset in assets])
    base = np.asarray(system.storage.get('base', prices))
    liquidated = np.array([asset.liquidated for asset in assets])
    cash = np.array([bank.external for bank in banks])
    status = np.array([bank.status for bank in banks])
    values = holdings @ prices + cash
    equity = values - np.asarray(self.liabilities, dtype=float)
    failed = equity <= 0
    if self.selling == 'leverage':
        excess = values - system.parameter('leverage') * equity
        excess = np.where(excess > 1e-9 * values, excess, 0.0)
        risky = holdings @ prices
        share = np.divide(excess, risky, out=np.zeros_like(excess), where=risky > 0)
        share = np.where(failed, 1.0, np.minimum(share, 1.0))
    else:
        share = np.where(failed & (status == SOLVENT), 1.0, 0.0)
    sold = holdings * share[:, None]
    cash = cash + sold @ prices
    holdings = holdings - sold
    totals = initial.sum(axis=0)
    liquidated = np.minimum(
        1.0,
        liquidated + np.divide(sold.sum(axis=0), totals, out=np.zeros_like(totals), where=totals > 0),
    )
    prices = base * np.exp(-system.parameter('alpha') * liquidated)
    status = np.where(failed, DEFAULTED, status)
    equity = holdings @ prices + cash - np.asarray(self.liabilities, dtype=float)
    system.storage.update(holdings=holdings, base=base)
    for bank, value, money, code in zip(banks, equity, cash, status, strict=True):
        bank.equity, bank.external, bank.status = float(value), float(money), float(code)
    for asset, price, share_sold in zip(assets, prices, liquidated, strict=True):
        asset.price, asset.liquidated = float(price), float(share_sold)
    system.report('defaults', float(np.sum(status == DEFAULTED)))
    system.report('sold', float((sold @ prices).sum()))

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.systems.economics.finance import GeometricBrownianMotionInteraction
>>> from skcomplex.core.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).index
1

Source code in src/skcomplex/systems/economics/common/_geometric_brownian_motion.py

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.components.elements import GroupElement
>>> from skcomplex.systems.economics.finance import GibratFirmElement
>>> GroupElement('firms', element=GibratFirmElement, size=10, values={'mu': 1.0}).size
10

Source code in src/skcomplex/systems/economics/finance/_growth_processes.py

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, system) staticmethod

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

Parameters:

Name	Type	Description	Default
agent	GibratFirmView	The firm.	required
system	SystemView	The system view of the growth.	required

Returns:

Name	Type	Description
size	float	The size after the step.

Source code in src/skcomplex/systems/economics/finance/_growth_processes.py

@staticmethod
def get_growth(agent: GibratFirmView, system: SystemView) -> float:
    """Return the size multiplied by a growth factor drawn from a normal distribution.

    Args:
        agent:
            The firm.

        system:
            The system view of the growth.

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

GoodSiteElement(label, visual=None, *, rho=1.0)

Bases: Element

Site of a lattice with a density rho of a good.

Parameters:

Name	Type	Description	Default
label	str	The label of the site or the group on the sites.	required
visual	VisualSpec | None	How the site is drawn, or None for default_visual_.	None
rho	Annotated[float, StateSpec()]	The initial density of the good.	1.0

Attributes:

Name	Type	Description
default_visual_		The visual of the site when it sets none.

Examples:

>>> from skcomplex.systems.economics.finance import GoodSiteElement
>>> GoodSiteElement('site', rho=2.0).rho
2.0

Source code in src/skcomplex/systems/economics/finance/_gauge_arbitrage.py

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

GuruMarketInteraction(label, *, rule=None, phase=None, traders, tau=100.0, beta_step=1.5, opening=10.0)

Bases: FeedbackInteraction[SystemView]

Order-driven market of noise traders who imitate the expectation of a guru, on a network rewired every day.

Every step is a trading day. First every trader picks a random trader j and moves its link from the trader k it follows to j with probability 1 / (1 + exp(-beta (f_j - f_k))), where the fitness f is the wealth relative to the richest trader (Eqs. 3.4 and 3.5 of Caiani et al. 2016). The guru is the trader of the largest in-degree of the day before, and the intensity of choice beta grows by beta_step every day the guru survives and returns to 1 when it is replaced. Every trader draws its expected return sigma_i eps (Eq. 3.3) and adopts the one of the trader it follows, except the guru.

Then tau intraday operations run, each by a trader drawn at random. With the expected price p_ref exp(sqrt(tau) r) (Eq. 3.1), a trader expecting a rise bids U(p (1 - gamma1), p_hat) for a random share of its free cash, and a trader expecting a fall asks U(p_hat, p (1 + gamma2)) for a random share of its free stocks (Eq. 3.2 and Table 3.1). An order that crosses the book is a market order that trades with the best quotes by price and time priority, and what is left rests as a limit order, whose cash or stocks stay tied in the book until it trades or expires at the end of the day. The price is the last traded price, the mid quote when an operation trades nothing, and the previous price when the book is empty.

Parameters:

Name	Type	Description	Default
label	str	The label of the market.	required
rule	None	Unused, since the market overrides act.	None
phase	str | None	The phase the market acts in, or None for the default phase.	None
traders	Annotated[str, RoleSpec()]	The path of the group of traders.	required
tau	Annotated[float, ParameterSpec(domain=open(0, oo))]	The intraday operations of a day.	100.0
beta_step	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The growth of the intensity of choice every day the guru survives.	1.5
opening	Annotated[float, ParameterSpec(domain=open(0, oo))]	The price before the first day.	10.0

Examples:

>>> from skcomplex.systems.economics.finance import GuruMarketInteraction
>>> GuruMarketInteraction('market', traders='traders').reports[:2]
('price', 'log_return')

Source code in src/skcomplex/systems/economics/finance/_guru_order_book.py

def __init__(
    self,
    label: str,
    *,
    rule: None = None,
    phase: str | None = None,
    traders: Annotated[str, RoleSpec()],
    tau: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 100.0,
    beta_step: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.5,
    opening: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 10.0,
) -> None:
    super().__init__(label, rule=rule, phase=phase)
    self.traders = traders
    self.tau = tau
    self.beta_step = beta_step
    self.opening = opening

act(system)

Rewire the network, form the expectations, and run the trading day.

Parameters:

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

Source code in src/skcomplex/systems/economics/finance/_guru_order_book.py

def act(self, system: SystemView) -> None:
    """Rewire the network, form the expectations, and run the trading day.

    Args:
        system:
            The system view of the market.
    """
    traders = sorted(system.agents('traders', GuruTraderView), key=lambda trader: trader.id)
    rng, store = system.rng, system.storage
    price = self._stored(system, 'price', system.parameter('opening'))
    beta = self._stored(system, 'beta', 1.0)
    cash = np.array([trader.cash for trader in traders])
    stocks = np.array([trader.stocks for trader in traders])
    follows = np.array([int(trader.follows) for trader in traders])
    sigma = np.array([trader.sigma for trader in traders])
    guru = int(np.argmax(np.bincount(follows, minlength=len(traders))))
    beta = beta + system.parameter('beta_step') if guru == self._stored(system, 'guru', -1.0) else 1.0
    wealth = cash + price * stocks
    follows = self._rewire(follows, wealth / wealth.max(), beta, rng)
    own = sigma * rng.normal(size=len(traders))
    expectation = own[follows]
    expectation[guru] = own[guru]
    holdings = {
        'cash': cash,
        'stocks': stocks,
        'tied_cash': np.zeros_like(cash),
        'tied_stocks': np.zeros_like(cash),
    }
    closing, volume = self._trade_day(system, holdings, expectation, price)
    followers = np.bincount(follows.astype(int), minlength=len(traders))
    store.update(price=closing, beta=beta, guru=guru)
    for index, trader in enumerate(traders):
        trader.cash, trader.stocks = float(cash[index]), float(stocks[index])
        trader.wealth = float(cash[index] + closing * stocks[index])
        trader.expectation, trader.follows = float(expectation[index]), float(follows[index])
        trader.followers = float(followers[index])
    system.report('price', closing)
    system.report('log_return', math.log(closing / price))
    system.report('volume', volume)
    system.report('guru', float(guru))
    system.report('beta', beta)
    system.report('guru_followers', float(followers.max()))

GuruTraderElement(label, visual=VisualSpec('agents', {'color_by': 'followers'}), *, cash=1000.0, stocks=100.0, wealth=2000.0, expectation=0.0, follows=0, sigma=0.01, followers=0)

Bases: Element

Noise trader of an order-driven market, who follows the expectation of the trader it is linked to.

The trader holds cash and stocks, has one outgoing link to the trader whose expected return it adopts, and expects returns of its own volatility sigma. The market of the system rewires its link and runs its orders.

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': 'followers'})
cash	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The cash of the trader, including the cash held by its limit orders.	1000.0
stocks	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The stocks of the trader, including the stocks held by its limit orders.	100.0
wealth	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The cash plus the stocks at the price of the day.	2000.0
expectation	Annotated[float, StateSpec()]	The expected return of the day after imitation.	0.0
follows	Annotated[float, StateSpec(domain=NN)]	The index of the trader whose expectation the trader adopts.	0
sigma	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The volatility of the expected returns of the trader.	0.01
followers	Annotated[float, StateSpec(domain=NN)]	The number of traders that follow the trader, its in-degree.	0

Examples:

>>> from skcomplex.core.components.elements import GroupElement
>>> from skcomplex.systems.economics.finance import GuruTraderElement
>>> GroupElement('traders', element=GuruTraderElement, size=3).size
3

Source code in src/skcomplex/systems/economics/finance/_guru_order_book.py

def __init__(
    self,
    label: str,
    visual: VisualSpec | None = VisualSpec('agents', {'color_by': 'followers'}),
    *,
    cash: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 1000.0,
    stocks: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 100.0,
    wealth: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 2000.0,
    expectation: Annotated[float, StateSpec()] = 0.0,
    follows: Annotated[float, StateSpec(domain=NN)] = 0,
    sigma: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.01,
    followers: Annotated[float, StateSpec(domain=NN)] = 0,
) -> None:
    super().__init__(label, visual=visual)
    self.cash = cash
    self.stocks = stocks
    self.wealth = wealth
    self.expectation = expectation
    self.follows = follows
    self.sigma = sigma
    self.followers = followers

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.components.elements import GroupElement
>>> from skcomplex.systems.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/systems/economics/finance/_contagion.py

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.components.elements import GroupElement
>>> from skcomplex.systems.economics.finance import KestenFirmElement
>>> GroupElement('firms', element=KestenFirmElement, size=10, values={'b_high': 2.0}).size
10

Source code in src/skcomplex/systems/economics/finance/_growth_processes.py

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, system) staticmethod

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

Parameters:

Name	Type	Description	Default
agent	KestenFirmView	The firm.	required
system	SystemView	The system view of the growth.	required

Returns:

Name	Type	Description
size	float	The size after the step.

Source code in src/skcomplex/systems/economics/finance/_growth_processes.py

@staticmethod
def get_growth(agent: KestenFirmView, system: SystemView) -> float:
    """Return the size multiplied by a random factor plus a random term.

    Args:
        agent:
            The firm.

        system:
            The system view of the growth.

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

MarketAssetElement(label, visual=None, *, price=1.0, liquidated=0.0)

Bases: Element

Asset held by banks, whose price falls with the share of it liquidated, p = exp(-alpha 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
price	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The price of the asset, 1 before any sale.	1.0
liquidated	Annotated[float, StateSpec(domain=closed(0, 1))]	The share of the asset sold so far.	0.0

Examples:

>>> from skcomplex.core.components.elements import GroupElement
>>> from skcomplex.systems.economics.finance import MarketAssetElement
>>> GroupElement('assets', element=MarketAssetElement, size=4).size
4

Source code in src/skcomplex/systems/economics/finance/_fire_sales.py

def __init__(
    self,
    label: str,
    visual: VisualSpec | None = None,
    *,
    price: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
    liquidated: Annotated[float, StateSpec(domain=RealSet.closed(0, 1))] = 0.0,
) -> None:
    super().__init__(label, visual=visual)
    self.price = price
    self.liquidated = liquidated

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.systems.economics.finance import NetworkSaverElement
>>> NetworkSaverElement('saver', W=2.0).W
2.0

Source code in src/skcomplex/systems/economics/finance/_wealth.py

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

PaymentClearingInteraction(label, *, rule=None, phase=None, banks, liabilities=())

Bases: FeedbackInteraction[SystemView]

Clearing of interbank payments by the iteration p <- min(p_bar, e + Pi^T p) of Eisenberg and Noe.

Every step every bank pays the smaller of what it owes, p_bar_i = sum_j L_ij, and what it has, its external assets plus the payments it receives, which its debtors make in proportion to their liabilities, Pi_ij = L_ij / p_bar_i. From full payments the iteration falls to the greatest clearing vector, and a bank that pays less than it owes defaults.

Parameters:

Name	Type	Description	Default
label	str	The label of the clearing.	required
rule	None	Unused, since the clearing overrides act.	None
phase	str | None	The phase the clearing acts in, or None for the default phase.	None
banks	Annotated[str, RoleSpec()]	The path of the group of banks.	required
liabilities	Sequence[Sequence[float]]	The interbank liabilities, liabilities[i][j] owed by bank i to bank j.	()

Examples:

>>> from skcomplex.systems.economics.finance import PaymentClearingInteraction
>>> PaymentClearingInteraction('clearing', banks='banks').reports
('shortfall', 'defaults')

Source code in src/skcomplex/systems/economics/finance/_solvency.py

def __init__(
    self,
    label: str,
    *,
    rule: None = None,
    phase: str | None = None,
    banks: Annotated[str, RoleSpec()],
    liabilities: Sequence[Sequence[float]] = (),
) -> None:
    super().__init__(label, rule=rule, phase=phase)
    self.banks = banks
    self.liabilities = liabilities

act(system)

Make one round of payments and record the shortfall and the defaults.

Parameters:

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

Source code in src/skcomplex/systems/economics/finance/_solvency.py

def act(self, system: SystemView) -> None:
    """Make one round of payments and record the shortfall and the defaults.

    Args:
        system:
            The system view of the clearing.
    """
    banks = sorted(system.agents('banks', BalanceSheetView), key=lambda bank: bank.id)
    matrix = np.asarray(self.liabilities, dtype=float)
    owed = matrix.sum(axis=1)
    shares = np.divide(matrix, owed[:, None], out=np.zeros_like(matrix), where=owed[:, None] > 0)
    payments = np.array([bank.payment for bank in banks])
    external = np.array([bank.external for bank in banks])
    payments = np.minimum(owed, external + shares.T @ payments)
    equity = external + shares.T @ payments - payments
    defaulted = payments < owed * (1 - 1e-12)
    for bank, paid, value, flag in zip(banks, payments, equity, defaulted, strict=True):
        bank.payment, bank.equity, bank.status = float(paid), float(value), float(DEFAULTED * flag)
    system.report('shortfall', float(np.sum(owed - payments)))
    system.report('defaults', float(defaulted.sum()))

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.systems.economics.finance import ReflectingFloorInteraction
>>> ReflectingFloorInteraction('floor', target='firms', c=0.5).c
0.5

Source code in src/skcomplex/systems/economics/finance/_growth_processes.py

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.components.elements import GroupElement
>>> from skcomplex.systems.economics.finance import StockTraderElement
>>> GroupElement('traders', element=StockTraderElement, size=4, values={'strategy': 1.0}).size
4

Source code in src/skcomplex/systems/economics/finance/_order_book.py

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, system) 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
system	SystemView	The system view of the accounting.	required

Returns:

Name	Type	Description
gain	float	The change of the wealth in the step.

Source code in src/skcomplex/systems/economics/finance/_order_book.py

@staticmethod
def get_gain(agent: StockTraderView, system: SystemView) -> float:
    """Return the change of the wealth valued at the last price, zero before the first valuation.

    Args:
        agent:
            The trader.

        system:
            The system view of the accounting.

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

get_wealth(agent, system) staticmethod

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

Parameters:

Name	Type	Description	Default
agent	StockTraderView	The trader.	required
system	SystemView	The system view of the accounting.	required

Returns:

Name	Type	Description
wealth	float	The wealth of the trader.

Source code in src/skcomplex/systems/economics/finance/_order_book.py

@staticmethod
def get_wealth(agent: StockTraderView, system: SystemView) -> float:
    """Return the money plus the shares valued at the last price.

    Args:
        agent:
            The trader.

        system:
            The system view of the accounting.

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

quote(agent, system)

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

Parameters:

Name	Type	Description	Default
agent	StockTraderView	The trader.	required
system	SystemView	The system.	required

Source code in src/skcomplex/systems/economics/finance/_order_book.py

def quote(self, agent: StockTraderView, system: SystemView) -> None:
    """Forecast the price and quote a limit order for one share.

    Args:
        agent:
            The trader.

        system:
            The system.
    """
    price, previous = self._prices(agent, system)
    if agent.strategy:
        expected = heuristics.trend([previous, price], strength=system.parameter('extrapolation'))
    else:
        reversion = 1 - system.parameter('reversion')
        expected = heuristics.fundamental([price], fundamental=system.parameter('fundamental'), reversion=reversion)
    expected = max(LOWEST_PRICE, expected * (1 + system.parameter('noise') * system.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)

SwitchingExchangeElement(label, visual=None, *, x=0.1, k=50, n_traders=100.0, r=0.01, v=0.5, g=1.0, sigma=0.01, epsilon=0.002, delta=0.01)

Bases: Element

Foreign exchange market whose rate deviation follows the share of fundamentalists of Kirman's recruitment chain.

The deviation x = s - s* of the exchange rate from its fundamental value follows Eq. 1.36 of Caiani et al. (2016), (1 + r) x_t = [1 - v n_t + g (1 - n_t)] x_t-1 - g (1 - n_t) x_t-2 + sigma eps_t, with a standard normal shock eps_t of the supply of foreign exchange. Of the N traders, k are fundamentalists, n = k / N. Every step one trader meets another at random, after Kirman (1993): a chartist turns fundamentalist with probability (1 - k / N) (epsilon + (1 - delta) k / (N - 1)) and a fundamentalist turns chartist with probability (k / N) (epsilon + (1 - delta) (N - k) / (N - 1)), by a spontaneous change epsilon or by recruitment.

Parameters:

Name	Type	Description	Default
label	str	The label of the market.	required
visual	VisualSpec | None	How the element is drawn, or None for default_visual_.	None
x	Annotated[float, StateSpec(lags=1)]	The deviation of the exchange rate from its fundamental value.	0.1
k	Annotated[float, StateSpec(domain=NN)]	The number of fundamentalists.	50
n_traders	Annotated[float, ParameterSpec(domain=open(1, oo))]	The number of traders N.	100.0
r	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The interest rate of the domestic currency.	0.01
v	Annotated[float, ParameterSpec(domain=closed(0, 1))]	The share of the deviation that fundamentalists expect to close.	0.5
g	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The share of the last change that chartists expect to repeat.	1.0
sigma	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The standard deviation of the shocks of the deviation.	0.01
epsilon	Annotated[float, ParameterSpec(domain=closed(0, 1))]	The probability of a spontaneous change of strategy.	0.002
delta	Annotated[float, ParameterSpec(domain=closed(0, 1))]	One minus the probability that a trader recruits the trader it meets.	0.01

Examples:

>>> from skcomplex.systems.economics.finance import SwitchingExchangeElement
>>> SwitchingExchangeElement('market', g=1.2).g
1.2

Source code in src/skcomplex/systems/economics/finance/_switching_exchange.py

def __init__(
    self,
    label: str,
    visual: VisualSpec | None = None,
    *,
    x: Annotated[float, StateSpec(lags=1)] = 0.1,
    k: Annotated[float, StateSpec(domain=NN)] = 50,
    n_traders: Annotated[float, ParameterSpec(domain=RealSet.open(1, oo))] = 100.0,
    r: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.01,
    v: Annotated[float, ParameterSpec(domain=RealSet.closed(0, 1))] = 0.5,
    g: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
    sigma: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.01,
    epsilon: Annotated[float, ParameterSpec(domain=RealSet.closed(0, 1))] = 0.002,
    delta: Annotated[float, ParameterSpec(domain=RealSet.closed(0, 1))] = 0.01,
) -> None:
    super().__init__(label, visual=visual)
    self.x = x
    self.k = k
    self.n_traders = n_traders
    self.r = r
    self.v = v
    self.g = g
    self.sigma = sigma
    self.epsilon = epsilon
    self.delta = delta

get_share(s) staticmethod

Return the share of fundamentalists n = k / N.

Parameters:

Name	Type	Description	Default
s	SimpleNamespace	The namespace of the market.	required

Returns:

Name	Type	Description
share	Expression	The share of fundamentalists.

Source code in src/skcomplex/systems/economics/finance/_switching_exchange.py

@staticmethod
def get_share(s: SimpleNamespace) -> Expression:
    """Return the share of fundamentalists `n = k / N`.

    Args:
        s:
            The namespace of the market.

    Returns:
        share:
            The share of fundamentalists.
    """
    return s.k / s.n_traders

TraderCurrentInteraction(label, *, target, links, D=0.5, q=0.2)

Bases: NeighborInteraction

Trader current J = -D (rho_t - rho_s - q A (rho_s + rho_t) / 2) along the links, down the covariant gradient.

The covariant derivative is D_mu = d_mu - A_mu of Schwichtenberg (2020 Eq. 2.55), with the charge q of its Eq. 4.24, and the current is minus the diffusion constant times this derivative of the good, as in Eq. 2.56. The current moves the good from the source of a link to its target (footnote 18 of §2.1), and it is the source of the field of the link with the sign of the Ampere-Maxwell law of Eq. 2.39, E' = ... - J, so the Gauss law holds as the good moves. These signs give the potential a mass, A'' = ... - D q rho A, for a positive density and charge, so the fields stay bounded.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction, unique among the parts of its system.	required
target	Annotated[str, RoleSpec()]	The path of the group on the sites.	required
links	Annotated[str, RoleSpec()]	The path of the group on the edges.	required
D	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The diffusivity of the good.	0.5
q	Annotated[float, ParameterSpec()]	The charge of the good.	0.2

Examples:

>>> from skcomplex.systems.economics.finance import TraderCurrentInteraction
>>> TraderCurrentInteraction('current', target='sites', links='links', q=0.5).q
0.5

Source code in src/skcomplex/systems/economics/finance/_gauge_arbitrage.py

def __init__(
    self,
    label: str,
    *,
    target: Annotated[str, RoleSpec()],
    links: Annotated[str, RoleSpec()],
    D: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.5,
    q: Annotated[float, ParameterSpec()] = 0.2,
) -> None:
    super().__init__(label)
    self.target = target
    self.links = links
    self.D = D
    self.q = q

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 system, with the density of the good market.rho.

Examples:

>>> from skcomplex.systems.economics.finance import arbitrage_traders
>>> arbitrage_traders(50).elements[0].values['E']
0.5

Source code in src/skcomplex/systems/economics/finance/_arbitrage.py

@register_system(
    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 system, with the density of the good `market.rho`.

    Examples:
        >>> from skcomplex.systems.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 system, with the wealth agents.W and the mean wealth mean_wealth.

Examples:

>>> from skcomplex.systems.economics.finance import bouchaud_mezard_wealth
>>> bouchaud_mezard_wealth(10).elements[0].values['J']
1.0

Source code in src/skcomplex/systems/economics/finance/_wealth.py

@register_system(
    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 system, with the wealth `agents.W` and the mean wealth `mean_wealth`.

    Examples:
        >>> from skcomplex.systems.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 system, with the assets first and second, whose motions share the Wiener processes market.

Examples:

>>> from skcomplex.systems.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/systems/economics/finance/_correlated_assets.py

@register_system(
    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 system, with the assets `first` and `second`, whose motions share the Wiener processes `market`.

    Examples:
        >>> from skcomplex.systems.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 system, with the group banks and the distress weighted by economic value value_at_risk.

Examples:

>>> from skcomplex.systems.economics.finance import debtrank
>>> debtrank([[0.0, 5.0], [0.0, 0.0]], [10.0, 10.0]).elements[0].prepare().size_
2

Source code in src/skcomplex/systems/economics/finance/_contagion.py

@register_system(
    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 system, with the group `banks` and the distress weighted by economic value `value_at_risk`.

    Examples:
        >>> from skcomplex.systems.economics.finance import debtrank
        >>> debtrank([[0.0, 5.0], [0.0, 0.0]], [10.0, 10.0]).elements[0].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())],
    )

default_cascade(liabilities=None, equity=None, *, shocked=(0,), shock=1.0, threshold=0.0, recovery=0.0, n_banks=250, degree=4.0, interbank_share=0.2, capital_ratio=0.04, random_state=0)

Return a cascade of defaults through interbank exposures, from a shock to the external assets of some banks.

Banks are linked by contracts, liabilities[i][j] the amount bank i owes bank j. A bank whose equity falls below the threshold h defaults, and its creditors write off their claims on it, losing (1 - R) L_ij of their equity at the next step, which may default them in turn. This is the cascade of Thurner et al. (2018, Section 5.5.3.2), and with h = 0 and the recovery rate R = 0 it is the contagion on default of Gai and Kapadia (2010) as Bardoscia et al. (2021, Section 3.1) review it. The shocked banks lose the share shock of their external assets at the start. Without a given network, every bank has assets of 100, a fifth of them interbank claims split evenly over a random number of debtors of mean degree, and capital of 4 percent, the network of Gai and Kapadia, whose cascades are large in a window of intermediate degrees and rare for sparse or dense networks.

Parameters:

Name	Type	Description	Default
liabilities	Sequence[Sequence[float]] | None	The interbank liabilities, or None for the random network of Gai and Kapadia.	None
equity	Sequence[float] | None	The equity of every bank before the shock, or None for the capital of the random network.	None
shocked	Sequence[int]	The indices of the banks whose external assets the shock hits.	(0,)
shock	float	The share of their external assets the shocked banks lose.	1.0
threshold	float	The equity below which a bank defaults.	0.0
recovery	float	The share of a claim on a defaulted bank that its creditor recovers.	0.0
n_banks	int	The number of banks of the random network.	250
degree	float	The mean number of debtors of a bank of the random network.	4.0
interbank_share	float	The share of the assets of a bank of the random network that are interbank claims.	0.2
capital_ratio	float	The capital of a bank of the random network as a share of its assets.	0.04
random_state	int | Generator | None	The seed or generator of the random network.	0

Returns:

Name	Type	Description
system	System	The system, with the group banks and the cascade cascade, whose reporters are the number of defaulted banks cascade.defaults and the losses written off in the step cascade.losses.

Examples:

>>> from skcomplex.systems.economics.finance import default_cascade
>>> default_cascade([[0.0, 5.0], [0.0, 0.0]], [1.0, 2.0]).elements[0].prepare().size_
2

Source code in src/skcomplex/systems/economics/finance/_solvency.py

@register_system(
    field='economics/finance',
    tags=('systemic risk', 'network', 'contagion', 'default cascade', 'agents'),
    references=('Thurner et al. 2018', 'Bardoscia et al. 2021', 'Gai and Kapadia 2010'),
    defaults={'until': 30},
)
def default_cascade(
    liabilities: Sequence[Sequence[float]] | None = None,
    equity: Sequence[float] | None = None,
    *,
    shocked: Sequence[int] = (0,),
    shock: float = 1.0,
    threshold: float = 0.0,
    recovery: float = 0.0,
    n_banks: int = 250,
    degree: float = 4.0,
    interbank_share: float = 0.2,
    capital_ratio: float = 0.04,
    random_state: int | np.random.Generator | None = 0,
) -> System:
    """Return a cascade of defaults through interbank exposures, from a shock to the external assets of some banks.

    Banks are linked by contracts, `liabilities[i][j]` the amount bank `i` owes bank `j`. A bank whose equity falls
    below the threshold `h` defaults, and its creditors write off their claims on it, losing `(1 - R) L_ij` of their
    equity at the next step, which may default them in turn. This is the cascade of Thurner et al. (2018, Section
    5.5.3.2), and with `h = 0` and the recovery rate `R = 0` it is the contagion on default of Gai and Kapadia (2010) as
    Bardoscia et al. (2021, Section 3.1) review it. The shocked banks lose the share `shock` of their external assets at
    the start. Without a given network, every bank has assets of 100, a fifth of them interbank claims split evenly over
    a random number of debtors of mean `degree`, and capital of 4 percent, the network of Gai and Kapadia, whose
    cascades are large in a window of intermediate degrees and rare for sparse or dense networks.

    Args:
        liabilities:
            The interbank liabilities, or `None` for the random network of Gai and Kapadia.

        equity:
            The equity of every bank before the shock, or `None` for the capital of the random network.

        shocked:
            The indices of the banks whose external assets the shock hits.

        shock:
            The share of their external assets the shocked banks lose.

        threshold:
            The equity below which a bank defaults.

        recovery:
            The share of a claim on a defaulted bank that its creditor recovers.

        n_banks:
            The number of banks of the random network.

        degree:
            The mean number of debtors of a bank of the random network.

        interbank_share:
            The share of the assets of a bank of the random network that are interbank claims.

        capital_ratio:
            The capital of a bank of the random network as a share of its assets.

        random_state:
            The seed or generator of the random network.

    Returns:
        system:
            The system, with the group `banks` and the cascade `cascade`, whose reporters are the number of defaulted
            banks `cascade.defaults` and the losses written off in the step `cascade.losses`.

    Examples:
        >>> from skcomplex.systems.economics.finance import default_cascade
        >>> default_cascade([[0.0, 5.0], [0.0, 0.0]], [1.0, 2.0]).elements[0].prepare().size_
        2
    """
    if liabilities is None or equity is None:
        rng = np.random.default_rng(random_state)
        matrix, external, capital = _gai_kapadia_network(n_banks, degree, interbank_share, capital_ratio, rng)
    else:
        matrix, capital = np.asarray(liabilities, dtype=float), np.asarray(equity, dtype=float)
        external = np.maximum(capital, 0.0)
    size = len(matrix)
    hit = np.zeros(size, dtype=bool)
    hit[list(shocked)] = True
    losses = np.where(hit, shock * external, 0.0)
    after = capital - losses
    status = np.where(after < threshold, DEFAULTING, SOLVENT).astype(float)
    banks = GroupElement(
        'banks',
        element=BalanceSheetElement,
        size=size,
        values={'external': external - losses, 'equity': after, 'status': status},
    )
    cascade = DefaultCascadeInteraction(
        'cascade',
        banks='banks',
        liabilities=matrix.tolist(),
        threshold=threshold,
        recovery=recovery,
    )
    return System(elements=[banks], interactions=[cascade])

eisenberg_noe_clearing(liabilities=None, external=None, *, n_banks=50, density=0.1, random_state=0)

Return the clearing of interbank payments of Eisenberg and Noe, a fixed point of proportional payments.

Bank i owes L_ij to bank j, p_bar_i in all, and holds external assets e_i. A clearing vector pays p_i = min(p_bar_i, e_i + sum_j Pi_ji p_j), every bank paying in full or all it has, its debtors paying in proportion Pi_ij = L_ij / p_bar_i, the liquidity contagion of Bardoscia et al. (2021, Section 3.1). Every step is one round of the iteration from full payments, which falls monotonically to the greatest clearing vector, the solution of the linear programme max sum_i p_i with p <= p_bar and p <= e + Pi^T p. Without a given network the liabilities are random and the external assets are small for a few banks.

Parameters:

Name	Type	Description	Default
liabilities	Sequence[Sequence[float]] | None	The interbank liabilities, or None for a random network.	None
external	Sequence[float] | None	The external assets of every bank, or None for random ones.	None
n_banks	int	The number of banks of a random network.	50
density	float	The probability of a liability between two banks of a random network.	0.1
random_state	int | Generator | None	The seed or generator of a random network.	0

Returns:

Name	Type	Description
system	System	The system, with the group banks of the payments banks.payment, and the clearing clearing, whose reporters are the total unpaid liabilities clearing.shortfall and the number of defaulting banks clearing.defaults.

Examples:

>>> from skcomplex.systems.economics.finance import eisenberg_noe_clearing
>>> eisenberg_noe_clearing([[0.0, 2.0], [1.0, 0.0]], [0.5, 1.0]).elements[0].prepare().size_
2

Source code in src/skcomplex/systems/economics/finance/_solvency.py

@register_system(
    field='economics/finance',
    tags=('systemic risk', 'network', 'clearing', 'payments', 'fixed point', 'agents'),
    references=('Bardoscia et al. 2021', 'Eisenberg and Noe 2001'),
    defaults={'until': 30},
)
def eisenberg_noe_clearing(
    liabilities: Sequence[Sequence[float]] | None = None,
    external: Sequence[float] | None = None,
    *,
    n_banks: int = 50,
    density: float = 0.1,
    random_state: int | np.random.Generator | None = 0,
) -> System:
    """Return the clearing of interbank payments of Eisenberg and Noe, a fixed point of proportional payments.

    Bank `i` owes `L_ij` to bank `j`, `p_bar_i` in all, and holds external assets `e_i`. A clearing vector pays
    `p_i = min(p_bar_i, e_i + sum_j Pi_ji p_j)`, every bank paying in full or all it has, its debtors paying in
    proportion `Pi_ij = L_ij / p_bar_i`, the liquidity contagion of Bardoscia et al. (2021, Section 3.1). Every step is
    one round of the iteration from full payments, which falls monotonically to the greatest clearing vector, the
    solution of the linear programme `max sum_i p_i` with `p <= p_bar` and `p <= e + Pi^T p`. Without a given network
    the liabilities are random and the external assets are small for a few banks.

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

        external:
            The external assets of every bank, or `None` for random ones.

        n_banks:
            The number of banks of a random network.

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

        random_state:
            The seed or generator of a random network.

    Returns:
        system:
            The system, with the group `banks` of the payments `banks.payment`, and the clearing `clearing`, whose
            reporters are the total unpaid liabilities `clearing.shortfall` and the number of defaulting banks
            `clearing.defaults`.

    Examples:
        >>> from skcomplex.systems.economics.finance import eisenberg_noe_clearing
        >>> eisenberg_noe_clearing([[0.0, 2.0], [1.0, 0.0]], [0.5, 1.0]).elements[0].prepare().size_
        2
    """
    if liabilities is None or external is None:
        rng = np.random.default_rng(random_state)
        links = rng.random((n_banks, n_banks)) < density
        np.fill_diagonal(links, val=False)
        matrix = links * rng.uniform(1.0, 10.0, (n_banks, n_banks))
        assets = rng.exponential(5.0, n_banks)
    else:
        matrix, assets = np.asarray(liabilities, dtype=float), np.asarray(external, dtype=float)
    banks = GroupElement(
        'banks',
        element=BalanceSheetElement,
        size=len(matrix),
        values={'external': assets, 'payment': matrix.sum(axis=1)},
    )
    clearing = PaymentClearingInteraction('clearing', banks='banks', liabilities=matrix.tolist())
    return System(elements=[banks], interactions=[clearing])

gauge_arbitrage_market(size=16, *, D=0.5, q=0.2, bump=1.0, seed=0)

Return exchange-rate potentials on links as lattice Maxwell fields, sourced by traders who move a good on sites.

The potentials follow A_t = E and E_t = -codifferential(dA) - J, and the good follows rho_t = -codifferential(J) with the current J = -D (d rho - q A rho) along its covariant gradient, after Schwichtenberg (2020 §2.1 and §2.2.1). Here E is F_0i of the book, so codifferential(E) equals the excess of the good over its mean, as in its Gauss law, Eq. 2.37, with a uniform background of the mean density that keeps the periodic lattice neutral. The system departs from the book in two places. The book writes Eq. 2.39 with no metric, which gives A_tt = +codifferential(dA) - J, where every mode with a curl grows, so the system takes the sign of a wave. And Eq. 2.56 applies the covariant derivative twice, which does not conserve the good, so the system applies it once inside the divergence of the current. The book gives no values of D and q and sets mu_0 by no number, so D, q, and mu_0 = 1 are choices of this system.

Parameters:

Name	Type	Description	Default
size	int	The number of sites along both axes.	16
D	float	The diffusivity of the good.	0.5
q	float	The charge of the good.	0.2
bump	float	The height of the random part of the initial density above a density of 1.	1.0
seed	int	The seed of the initial density.	0

Returns:

Name	Type	Description
system	System	The system.

Examples:

>>> from skcomplex.systems.economics.finance import gauge_arbitrage_market
>>> [element.label for element in gauge_arbitrage_market(4).elements]
['sites', 'links']

Source code in src/skcomplex/systems/economics/finance/_gauge_arbitrage.py

@register_system(
    field='economics/finance',
    tags=('gauge', 'arbitrage', 'exchange rates', 'lattice', 'maxwell', 'cell complex'),
    references=('Schwichtenberg 2020',),
    defaults={'until': 20.0, 'dt': 0.5, 'solver': RK4Solver(dt=0.02)},
    observables=_market_observables,
)
def gauge_arbitrage_market(
    size: int = 16,
    *,
    D: float = 0.5,
    q: float = 0.2,
    bump: float = 1.0,
    seed: int = 0,
) -> System:
    """Return exchange-rate potentials on links as lattice Maxwell fields, sourced by traders who move a good on sites.

    The potentials follow `A_t = E` and `E_t = -codifferential(dA) - J`, and the good follows `rho_t =
    -codifferential(J)` with the current `J = -D (d rho - q A rho)` along its covariant gradient, after Schwichtenberg
    (2020 §2.1 and §2.2.1). Here `E` is `F_0i` of the book, so `codifferential(E)` equals the excess of the good over
    its mean, as in its Gauss law, Eq. 2.37, with a uniform background of the mean density that keeps the periodic
    lattice neutral. The system departs from the book in two places. The book writes Eq. 2.39 with no metric, which
    gives `A_tt = +codifferential(dA) - J`, where every mode with a curl grows, so the system takes the sign of a wave.
    And Eq. 2.56 applies the covariant derivative twice, which does not conserve the good, so the system applies it once
    inside the divergence of the current. The book gives no values of `D` and `q` and sets `mu_0` by no number, so `D`,
    `q`, and `mu_0 = 1` are choices of this system.

    Args:
        size:
            The number of sites along both axes.

        D:
            The diffusivity of the good.

        q:
            The charge of the good.

        bump:
            The height of the random part of the initial density above a density of 1.

        seed:
            The seed of the initial density.

    Returns:
        system:
            The system.

    Examples:
        >>> from skcomplex.systems.economics.finance import gauge_arbitrage_market
        >>> [element.label for element in gauge_arbitrage_market(4).elements]
        ['sites', 'links']
    """
    space = CellComplexSpace(LatticeSpace((size, size))).prepare()
    density = 1.0 + bump * np.random.default_rng(seed).random(size * size)
    field = _initial_field(space.incidences_[0], density - density.mean())
    sites = GroupElement('sites', element=GoodSiteElement, space=space, on='sites', values={'rho': density})
    links = GroupElement(
        'links',
        element=ExchangeLinkElement,
        space=space,
        on='edges',
        values={'E': field},
        visual=VisualSpec('cell', {'state': 'A'}),
    )
    interactions = [
        RateInteraction('potential', target='links', state='A', rate=_potential_rate),
        RateInteraction('field', target='links', state='E', rate=_field_rate),
        TraderCurrentInteraction('current', target='sites', links='links', D=D, q=q),
    ]
    return System(elements=[sites, links], interactions=interactions)

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 system, with the asset asset.

Examples:

>>> from skcomplex.systems.economics.finance import geometric_brownian_motion
>>> geometric_brownian_motion(sigma=0.3).interactions[0].sigma
0.3

Source code in src/skcomplex/systems/economics/finance/_assets.py

@register_system(
    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 system, with the asset `asset`.

    Examples:
        >>> from skcomplex.systems.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 system, with the group firms, the mean size mean_size, and the largest size max_size.

Examples:

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

Source code in src/skcomplex/systems/economics/finance/_growth_processes.py

@register_system(
    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 system, with the group `firms`, the mean size `mean_size`, and the largest size `max_size`.

    Examples:
        >>> from skcomplex.systems.economics.finance import gibrat_process
        >>> gibrat_process(10).elements[0].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())

guru_order_book(n_traders=100, *, tau=100, beta_step=1.5, sigma=0.01, cash=1000.0, stocks=100.0, opening=10.0, random_state=0)

Return the order-driven market of Grilli and Tedeschi, whose noise traders imitate gurus they choose by wealth.

Noise traders expect random returns of their own volatility, but every day each adopts the expectation of the trader it follows on a directed communication network. Every day each trader moves its link to a random trader with a logit probability of the difference of their wealths, whose intensity grows every day the guru, the most followed trader, keeps its place, so followers concentrate on successful traders and their expectations synchronize. The traders then place market and limit orders around a geometric Brownian forecast in an order book with price and time priority, within the day. This is the model of Caiani et al. (2016, Chapter 3, Sections 3.2 and 3.4). Imitation produces coordination periods in which most traders follow one guru, and the returns get fat tails and volatility clusters. The source leaves two choices open: the volatilities are uniform on (0, sigma), and the intensity of choice grows by beta_step every day the guru survives. The cash and the stocks of the traders are conserved, which the default run checks as invariants.

Parameters:

Name	Type	Description	Default
n_traders	int	The number of traders.	100
tau	int	The intraday operations of a day.	100
beta_step	float	The growth of the intensity of choice every day the guru survives.	1.5
sigma	float	The largest volatility of the expected returns of a trader.	0.01
cash	float	The initial cash of every trader.	1000.0
stocks	float	The initial stocks of every trader.	100.0
opening	float	The price before the first day.	10.0
random_state	int | Generator | None	The seed or generator of the volatilities and the initial links.	0

Returns:

Name	Type	Description
system	System	The system, with the group traders and the market market, whose reporters are market.price, market.log_return, market.volume, market.guru, market.beta, and market.guru_followers. The observables of the system are the totals total_cash and total_stocks.

Examples:

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

Source code in src/skcomplex/systems/economics/finance/_guru_order_book.py

@register_system(
    field='economics/finance',
    tags=('stock market', 'agents', 'order book', 'imitation', 'network', 'gurus', 'noise traders'),
    references=('Caiani et al. 2016 §3.2',),
    defaults={'until': 300, 'invariants': {f'total_{name}': 1e-6 for name in HOLDINGS}},
    observables=lambda system: [
        ObservableSpec(f'total_{name}', group='traders', variable=name, reduce='sum') for name in HOLDINGS
    ],
)
def guru_order_book(
    n_traders: int = 100,
    *,
    tau: int = 100,
    beta_step: float = 1.5,
    sigma: float = 0.01,
    cash: float = 1000.0,
    stocks: float = 100.0,
    opening: float = 10.0,
    random_state: int | np.random.Generator | None = 0,
) -> System:
    """Return the order-driven market of Grilli and Tedeschi, whose noise traders imitate gurus they choose by wealth.

    Noise traders expect random returns of their own volatility, but every day each adopts the expectation of the
    trader it follows on a directed communication network. Every day each trader moves its link to a random trader with
    a logit probability of the difference of their wealths, whose intensity grows every day the guru, the most followed
    trader, keeps its place, so followers concentrate on successful traders and their expectations synchronize. The
    traders then place market and limit orders around a geometric Brownian forecast in an order book with price and
    time priority, within the day. This is the model of Caiani et al. (2016, Chapter 3, Sections 3.2 and 3.4). Imitation
    produces coordination periods in which most traders follow one guru, and the returns get fat tails and volatility
    clusters. The source leaves two choices open: the volatilities are uniform on `(0, sigma)`, and the intensity of
    choice grows by `beta_step` every day the guru survives. The cash and the stocks of the traders are conserved,
    which the default run checks as invariants.

    Args:
        n_traders:
            The number of traders.

        tau:
            The intraday operations of a day.

        beta_step:
            The growth of the intensity of choice every day the guru survives.

        sigma:
            The largest volatility of the expected returns of a trader.

        cash:
            The initial cash of every trader.

        stocks:
            The initial stocks of every trader.

        opening:
            The price before the first day.

        random_state:
            The seed or generator of the volatilities and the initial links.

    Returns:
        system:
            The system, with the group `traders` and the market `market`, whose reporters are `market.price`,
            `market.log_return`, `market.volume`, `market.guru`, `market.beta`, and `market.guru_followers`. The
            observables of the system are the totals `total_cash` and `total_stocks`.

    Examples:
        >>> from skcomplex.systems.economics.finance import guru_order_book
        >>> guru_order_book(10).elements[0].prepare().size_
        10
    """
    rng = np.random.default_rng(random_state)
    links = rng.integers(n_traders - 1, size=n_traders)
    links += links >= np.arange(n_traders)
    traders = GroupElement(
        'traders',
        element=GuruTraderElement,
        size=n_traders,
        values={
            'cash': cash,
            'stocks': stocks,
            'wealth': cash + opening * stocks,
            'follows': links.astype(float),
            'sigma': rng.uniform(0.0, sigma, n_traders),
        },
    )
    market = GuruMarketInteraction('market', traders='traders', tau=tau, beta_step=beta_step, opening=opening)
    return System(elements=[traders], interactions=[market])

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 system, with the group firms, the mean size mean_size, and the largest size max_size.

Examples:

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

Source code in src/skcomplex/systems/economics/finance/_growth_processes.py

@register_system(
    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 system, with the group `firms`, the mean size `mean_size`, and the largest size `max_size`.

    Examples:
        >>> from skcomplex.systems.economics.finance import kesten_process
        >>> kesten_process(10).elements[0].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())

kirman_exchange_rate(x=0.1, k=15, *, n_traders=30, r=0.01, v=0.5, g=1.0, sigma=0.01, epsilon=0.005, delta=0.01)

Return Kirman's foreign exchange market, whose traders switch between fundamentalism and chartism like ants.

Fundamentalists expect the exchange rate to revert to its fundamental value, s_t-1 + v (s* - s_t-1), and chartists expect the last change to go on, s_t-1 + g (s_t-1 - s_t-2). With the mean-variance demand of both and the share n_t of fundamentalists, the market clears at the deviation of Eq. 1.36 of Caiani et al. (2016, Section 1.3.2), (1 + r) x_t = [1 - v n_t + g (1 - n_t)] x_t-1 - g (1 - n_t) x_t-2, written there with s_t-2 for x_t-2, a shock of the supply of foreign exchange added. The share follows the recruitment chain of Kirman's ants, which the source refers to and Kirman (1993) gives: when spontaneous changes are rarer than recruitment, epsilon < (1 - delta) / (N - 1), the traders spend long periods mostly of one kind and switch suddenly. A market of fundamentalists is stable with the eigenvalue (1 - v) / (1 + r) (Eq. 1.37), and a market of chartists oscillates, stable only while g <= 1 + r (Eq. 1.38), so the volatility of the rate comes and goes with the chartists. One trader meets another every step, so the defaults take a small market of 30 traders, whose regimes switch within a few thousand steps.

Parameters:

Name	Type	Description	Default
x	float	The initial deviation of the exchange rate from its fundamental value, also its value one step earlier.	0.1
k	int	The initial number of fundamentalists.	15
n_traders	int	The number of traders.	30
r	float	The interest rate of the domestic currency.	0.01
v	float	The share of the deviation that fundamentalists expect to close.	0.5
g	float	The share of the last change that chartists expect to repeat.	1.0
sigma	float	The standard deviation of the shocks of the deviation.	0.01
epsilon	float	The probability of a spontaneous change of strategy.	0.005
delta	float	One minus the probability that a trader recruits the trader it meets.	0.01

Returns:

Name	Type	Description
system	System	The system, with the deviation market.x and the number of fundamentalists market.k. The observable of the system is the share of fundamentalists market.fundamentalists.

Examples:

>>> from skcomplex.systems.economics.finance import kirman_exchange_rate
>>> kirman_exchange_rate(g=1.2).elements[0].g
1.2

Source code in src/skcomplex/systems/economics/finance/_switching_exchange.py

@register_system(
    field='economics/finance',
    tags=('exchange rate', 'fundamentalists', 'chartists', 'herding', 'ants', 'discrete map'),
    references=('Caiani et al. 2016 §1.3.2', 'Kirman 1993'),
    defaults={'until': 20000},
    observables=lambda system: [
        ObservableSpec('fundamentalists', SwitchingExchangeElement.get_share, element='market'),
    ],
)
def kirman_exchange_rate(
    x: float = 0.1,
    k: int = 15,
    *,
    n_traders: int = 30,
    r: float = 0.01,
    v: float = 0.5,
    g: float = 1.0,
    sigma: float = 0.01,
    epsilon: float = 0.005,
    delta: float = 0.01,
) -> System:
    """Return Kirman's foreign exchange market, whose traders switch between fundamentalism and chartism like ants.

    Fundamentalists expect the exchange rate to revert to its fundamental value, `s_t-1 + v (s* - s_t-1)`, and chartists
    expect the last change to go on, `s_t-1 + g (s_t-1 - s_t-2)`. With the mean-variance demand of both and the share
    `n_t` of fundamentalists, the market clears at the deviation of Eq. 1.36 of Caiani et al. (2016, Section 1.3.2),
    `(1 + r) x_t = [1 - v n_t + g (1 - n_t)] x_t-1 - g (1 - n_t) x_t-2`, written there with `s_t-2` for `x_t-2`, a shock
    of the supply of foreign exchange added. The share follows the recruitment chain of Kirman's ants, which the source
    refers to and Kirman (1993) gives: when spontaneous changes are rarer than recruitment, `epsilon < (1 - delta) /
    (N - 1)`, the traders spend long periods mostly of one kind and switch suddenly. A market of fundamentalists is
    stable with the eigenvalue `(1 - v) / (1 + r)` (Eq. 1.37), and a market of chartists oscillates, stable only while
    `g <= 1 + r` (Eq. 1.38), so the volatility of the rate comes and goes with the chartists. One trader meets another
    every step, so the defaults take a small market of 30 traders, whose regimes switch within a few thousand steps.

    Args:
        x:
            The initial deviation of the exchange rate from its fundamental value, also its value one step earlier.

        k:
            The initial number of fundamentalists.

        n_traders:
            The number of traders.

        r:
            The interest rate of the domestic currency.

        v:
            The share of the deviation that fundamentalists expect to close.

        g:
            The share of the last change that chartists expect to repeat.

        sigma:
            The standard deviation of the shocks of the deviation.

        epsilon:
            The probability of a spontaneous change of strategy.

        delta:
            One minus the probability that a trader recruits the trader it meets.

    Returns:
        system:
            The system, with the deviation `market.x` and the number of fundamentalists `market.k`. The observable of
            the system is the share of fundamentalists `market.fundamentalists`.

    Examples:
        >>> from skcomplex.systems.economics.finance import kirman_exchange_rate
        >>> kirman_exchange_rate(g=1.2).elements[0].g
        1.2
    """
    market = SwitchingExchangeElement(
        'market',
        x=x,
        k=k,
        n_traders=n_traders,
        r=r,
        v=v,
        g=g,
        sigma=sigma,
        epsilon=epsilon,
        delta=delta,
    )
    return System(elements=[market])

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 system, with the group firms, the mean size mean_size, and the largest size max_size.

Examples:

>>> from skcomplex.systems.economics.finance import multiplicative_process
>>> multiplicative_process(c=0.5).interactions[-1].c
0.5

Source code in src/skcomplex/systems/economics/finance/_growth_processes.py

@register_system(
    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 system, with the group `firms`, the mean size `mean_size`, and the largest size `max_size`.

    Examples:
        >>> from skcomplex.systems.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.

The two strategies are the fundamentalists and chartists of Lux and Marchesi (1999), and imitating a wealthier neighbour on the grid stands in for their switching between strategies. The order book is the order-driven market of Caiani et al. (2016, section 3.2.1). The traders of Caiani's chapter 3, noise traders who follow gurus of an evolving network, are a different model.

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 system, 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.systems.economics.finance import order_book
>>> order_book(5, 5).elements[0].prepare().size_
25

Source code in src/skcomplex/systems/economics/finance/_order_book.py

@register_system(
    field='economics/finance',
    tags=('stock market', 'agents', 'order book', 'imitation', 'fundamentalists', 'chartists'),
    references=('Lux and Marchesi 1999', 'Caiani et al. 2016 §3.2.1'),
    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.

    The two strategies are the fundamentalists and chartists of Lux and Marchesi (1999), and imitating a wealthier
    neighbour on the grid stands in for their switching between strategies. The order book is the order-driven market
    of Caiani et al. (2016, section 3.2.1). The traders of Caiani's chapter 3, noise traders who follow gurus of an
    evolving network, are a different model.

    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 system, 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.systems.economics.finance import order_book
        >>> order_book(5, 5).elements[0].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'),
                ],
            ),
        ),
    )

overlapping_portfolio_fire_sales(n_banks=100, n_assets=100, *, diversification=10, leverage=20.0, alpha=1.0, shocked=0, shock=0.2, selling='threshold', random_state=0)

Return banks with overlapping portfolios whose fire sales of common assets spread their losses to one another.

Every bank invests equal amounts in diversification assets drawn at random, a bipartite network of banks and assets, with assets leverage times its equity. A shock lowers the price of one asset, and banks react to their losses: under the linear threshold rule a bank stays passive until its losses exhaust its equity and then liquidates its whole portfolio, and under leverage targeting it sells a share of every holding to restore its leverage. Sales lower prices through the market impact p = p0 exp(-alpha x) of the share x of an asset liquidated, which hits every bank holding the asset, so losses travel between banks that share no contract, the indirect contagion of Bardoscia et al. (2021, Section 3.2) after Caccioli et al. (2014). Larger impact and leverage make the cascades larger, and leverage targeting sells more for small shocks.

Parameters:

Name	Type	Description	Default
n_banks	int	The number of banks.	100
n_assets	int	The number of assets.	100
diversification	int	The number of assets in every portfolio.	10
leverage	float	The ratio of assets to equity of every bank, and the target of leverage targeting.	20.0
alpha	float	The market impact.	1.0
shocked	int	The index of the asset the shock hits.	0
shock	float	The share of its price the shocked asset loses.	0.2
selling	str	The rule of the banks, 'threshold' or 'leverage'.	'threshold'
random_state	int | Generator | None	The seed or generator of the portfolios.	0

Returns:

Name	Type	Description
system	System	The system, with the groups banks, of the equity banks.equity and the cash banks.external, and assets, of the prices assets.price, and the fire sales sales, whose reporters are the number of defaulted banks sales.defaults and the value sold in the step sales.sold.

Examples:

>>> from skcomplex.systems.economics.finance import overlapping_portfolio_fire_sales
>>> [group.prepare().size_ for group in overlapping_portfolio_fire_sales(10, 5, diversification=2).elements]
[10, 5]

Source code in src/skcomplex/systems/economics/finance/_fire_sales.py

@register_system(
    field='economics/finance',
    tags=('systemic risk', 'fire sales', 'overlapping portfolios', 'bipartite network', 'market impact', 'agents'),
    references=('Bardoscia et al. 2021', 'Caccioli et al. 2014'),
    defaults={'until': 30},
)
def overlapping_portfolio_fire_sales(
    n_banks: int = 100,
    n_assets: int = 100,
    *,
    diversification: int = 10,
    leverage: float = 20.0,
    alpha: float = 1.0,
    shocked: int = 0,
    shock: float = 0.2,
    selling: str = 'threshold',
    random_state: int | np.random.Generator | None = 0,
) -> System:
    """Return banks with overlapping portfolios whose fire sales of common assets spread their losses to one another.

    Every bank invests equal amounts in `diversification` assets drawn at random, a bipartite network of banks and
    assets, with assets `leverage` times its equity. A shock lowers the price of one asset, and banks react to their
    losses: under the linear threshold rule a bank stays passive until its losses exhaust its equity and then
    liquidates its whole portfolio, and under leverage targeting it sells a share of every holding to restore its
    leverage. Sales lower prices through the market impact `p = p0 exp(-alpha x)` of the share `x` of an asset
    liquidated, which hits every bank holding the asset, so losses travel between banks that share no contract, the
    indirect contagion of Bardoscia et al. (2021, Section 3.2) after Caccioli et al. (2014). Larger impact and leverage
    make the cascades larger, and leverage targeting sells more for small shocks.

    Args:
        n_banks:
            The number of banks.

        n_assets:
            The number of assets.

        diversification:
            The number of assets in every portfolio.

        leverage:
            The ratio of assets to equity of every bank, and the target of leverage targeting.

        alpha:
            The market impact.

        shocked:
            The index of the asset the shock hits.

        shock:
            The share of its price the shocked asset loses.

        selling:
            The rule of the banks, `'threshold'` or `'leverage'`.

        random_state:
            The seed or generator of the portfolios.

    Returns:
        system:
            The system, with the groups `banks`, of the equity `banks.equity` and the cash `banks.external`, and
            `assets`, of the prices `assets.price`, and the fire sales `sales`, whose reporters are the number of
            defaulted banks `sales.defaults` and the value sold in the step `sales.sold`.

    Examples:
        >>> from skcomplex.systems.economics.finance import overlapping_portfolio_fire_sales
        >>> [group.prepare().size_ for group in overlapping_portfolio_fire_sales(10, 5, diversification=2).elements]
        [10, 5]
    """
    rng = np.random.default_rng(random_state)
    holdings = np.zeros((n_banks, n_assets))
    for bank in range(n_banks):
        holdings[bank, rng.choice(n_assets, diversification, replace=False)] = 1.0 / diversification
    assets_value = holdings.sum(axis=1)
    liabilities = assets_value * (1 - 1 / leverage)
    prices = np.ones(n_assets)
    prices[shocked] = 1 - shock
    equity = holdings @ prices - liabilities
    banks = GroupElement('banks', element=BalanceSheetElement, size=n_banks, values={'equity': equity})
    assets = GroupElement('assets', element=MarketAssetElement, size=n_assets, values={'price': prices})
    sales = FireSaleInteraction(
        'sales',
        banks='banks',
        assets='assets',
        holdings=tuple(map(tuple, holdings.tolist())),
        liabilities=tuple(liabilities.tolist()),
        alpha=alpha,
        leverage=leverage,
        selling=selling,
    )
    return System(elements=[banks, assets], interactions=[sales])