Markets

Markets: prices, competition, retail, inventories, queues, and barter of Cobb-Douglas traders on trade networks.

AdaptiveAuctionInteraction(label, *, rule=None, phase=None, buyers, sellers, step=1.0, unlimited=False)

Bases: FeedbackInteraction[SystemView]

Double auction in which every buyer meets a random seller every iteration and both adapt their quotes.

Every iteration the buyers that still want units, in random order, meet a random seller that still holds units. When the bid reaches the ask, one unit trades at the ask, and afterwards the buyer lowers its bid and the seller raises its ask, to push the price their way. When it does not, the buyer raises its bid and the seller lowers its ask, to create trade, each change a uniform draw between 0 and step. Bids never exceed the values and asks never fall below the costs. With unlimited units, traders are resupplied and keep consuming, so trade goes on forever.

Parameters:

Name	Type	Description	Default
label	str	The label of the auction.	required
rule	None	Unused, since the auction overrides act.	None
phase	str | None	The phase the auction acts in, or None for the default phase.	None
buyers	Annotated[str, RoleSpec()]	The path of the group of buyers.	required
sellers	Annotated[str, RoleSpec()]	The path of the group of sellers.	required
step	Annotated[float, ParameterSpec(domain=open(0, oo))]	The change of a quote after a meeting.	1.0
unlimited	bool	Whether the units of the traders never run out.	False

Examples:

>>> from skcomplex.systems.economics.markets import AdaptiveAuctionInteraction
>>> AdaptiveAuctionInteraction('auction', buyers='buyers', sellers='sellers').reports
('price', 'volume', 'mean_bid', 'mean_ask')

Source code in src/skcomplex/systems/economics/markets/_adaptive_auction.py

def __init__(
    self,
    label: str,
    *,
    rule: None = None,
    phase: str | None = None,
    buyers: Annotated[str, RoleSpec()],
    sellers: Annotated[str, RoleSpec()],
    step: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 1.0,
    unlimited: bool = False,
) -> None:
    super().__init__(label, rule=rule, phase=phase)
    self.buyers = buyers
    self.sellers = sellers
    self.step = step
    self.unlimited = unlimited

act(system)

Run the meetings of one iteration.

Parameters:

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

Source code in src/skcomplex/systems/economics/markets/_adaptive_auction.py

def act(self, system: SystemView) -> None:
    """Run the meetings of one iteration.

    Args:
        system:
            The system view of the auction.
    """
    rng, step = system.rng, system.parameter('step')
    buyers = sorted(system.agents('buyers', AdaptiveTraderView), key=lambda trader: trader.id)
    sellers = sorted(system.agents('sellers', AdaptiveTraderView), key=lambda trader: trader.id)
    prices = []
    for buyer in (buyers[index] for index in rng.permutation(len(buyers))):
        stocked = [seller for seller in sellers if seller.units > 0]
        if buyer.units <= 0 or not stocked:
            continue
        seller = stocked[int(rng.integers(len(stocked)))]
        bid_step, ask_step = step * rng.random(), step * rng.random()
        if buyer.quote >= seller.quote:
            prices.append(seller.quote)
            if not self.unlimited:
                buyer.units, seller.units = buyer.units - 1, seller.units - 1
            buyer.quote = max(buyer.quote - bid_step, 0.0)
            seller.quote += ask_step
        else:
            buyer.quote = min(buyer.quote + bid_step, buyer.value)
            seller.quote = max(seller.quote - ask_step, seller.value)
    last = system.history('price', 1)
    price = prices[-1] if prices else (float(last[-1]) if len(last) else float('nan'))
    system.report('price', price)
    system.report('volume', float(len(prices)))
    system.report('mean_bid', float(np.mean([buyer.quote for buyer in buyers])))
    system.report('mean_ask', float(np.mean([seller.quote for seller in sellers])))

AdaptiveTraderElement(label, visual=VisualSpec('agents', {'color_by': 'quote'}), *, value=0.0, quote=0.0, units=5.0)

Bases: Element

Buyer or seller of a double auction, with its value or cost, its current bid or ask, and its units left.

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': 'quote'})
value	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The value of a unit to a buyer, or its cost to a seller.	0.0
quote	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The current bid of a buyer or ask of a seller.	0.0
units	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The units a buyer still wants or a seller still holds.	5.0

Examples:

>>> from skcomplex.core.components.elements import GroupElement
>>> from skcomplex.systems.economics.markets import AdaptiveTraderElement
>>> GroupElement('buyers', element=AdaptiveTraderElement, size=3, values={'value': 120.0}).size
3

Source code in src/skcomplex/systems/economics/markets/_adaptive_auction.py

def __init__(
    self,
    label: str,
    visual: VisualSpec | None = VisualSpec('agents', {'color_by': 'quote'}),
    *,
    value: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
    quote: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
    units: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 5.0,
) -> None:
    super().__init__(label, visual=visual)
    self.value = value
    self.quote = quote
    self.units = units

AuctionLedgerInteraction(label, *, rule=None, phase=None)

Bases: FeedbackInteraction[SystemView]

Ledger that closes every trading period: it measures its surplus and gives every trader a new unit.

The realised surplus is the value of the units bought minus the cost of the units sold, and the allocative efficiency is its ratio to the maximum surplus of the supply and demand schedules.

Parameters:

Name	Type	Description	Default
label	str	The label of the ledger.	required
phase	str | None	The phase the ledger acts in.	None

Examples:

>>> from skcomplex.systems.economics.markets import AuctionLedgerInteraction
>>> AuctionLedgerInteraction('ledger', phase='close').reports
('surplus', 'efficiency')

Source code in src/skcomplex/core/components/interactions/_feedback.py

def __init__(
    self,
    label: str,
    *,
    rule: Callable[[SystemT], None] | None = None,
    phase: str | None = None,
) -> None:
    super().__init__(label)
    self.rule = rule
    self.phase = phase

act(system)

Report the surplus and the efficiency of the trading period and open the next one.

Parameters:

Name	Type	Description	Default
system	SystemView	The system view scoped to the ledger.	required

Source code in src/skcomplex/systems/economics/markets/_double_auction.py

def act(self, system: SystemView) -> None:
    """Report the surplus and the efficiency of the trading period and open the next one.

    Args:
        system:
            The system view scoped to the ledger.
    """
    buyers = system.agents('buyers', ZeroIntelligenceBuyerView)
    sellers = system.agents('sellers', ZeroIntelligenceSellerView)
    values = np.array([buyer.value for buyer in buyers])
    costs = np.array([seller.value for seller in sellers])
    bought = np.array([buyer.demand <= 0 for buyer in buyers])
    sold = np.array([seller.supply <= 0 for seller in sellers])
    surplus = float(values[bought].sum() - costs[sold].sum())
    system.report('surplus', surplus)
    system.report('efficiency', surplus / maximum_surplus(values, costs))
    for buyer in buyers:
        buyer.demand = 1.0
    for seller in sellers:
        seller.supply = 1.0

BarterTraderElement(label, visual=None, *, good1=10.0, good2=10.0, mrs=1.0, links=0.0, partners=math.inf)

Bases: Element

Trader with a bundle of two goods and the Cobb-Douglas utility good1 good2, who records its rate.

Its marginal rate of substitution good2 / good1 is the price of the first good, in units of the second, at which it is indifferent to trade. Every step it records the rate of its current bundle.

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
good1	Annotated[float, StateSpec(domain=open(0, oo))]	The holding of the first good.	10.0
good2	Annotated[float, StateSpec(domain=open(0, oo))]	The holding of the second good.	10.0
mrs	Annotated[float, StateSpec(domain=open(0, oo))]	The marginal rate of substitution of the bundle.	1.0
links	Annotated[float, StateSpec(domain=NN)]	The number of trading partners the trader can search.	0.0
partners	Annotated[float, StateSpec()]	The number of partners that can match with the trader in a step, without limit.	inf

Examples:

>>> from skcomplex.core.components.elements import GroupElement
>>> from skcomplex.systems.economics.markets import BarterTraderElement
>>> GroupElement('traders', element=BarterTraderElement, size=4).size
4

Source code in src/skcomplex/systems/economics/markets/_barter_networks.py

def __init__(
    self,
    label: str,
    visual: VisualSpec | None = None,
    *,
    good1: Annotated[float, StateSpec(domain=RealSet.open(0, oo))] = 10.0,
    good2: Annotated[float, StateSpec(domain=RealSet.open(0, oo))] = 10.0,
    mrs: Annotated[float, StateSpec(domain=RealSet.open(0, oo))] = 1.0,
    links: Annotated[float, StateSpec(domain=NN)] = 0.0,
    partners: Annotated[float, StateSpec()] = math.inf,
) -> None:
    super().__init__(label, visual=visual)
    self.good1 = good1
    self.good2 = good2
    self.mrs = mrs
    self.links = links
    self.partners = partners

get_rate(agent, system) staticmethod

Return the marginal rate of substitution of the bundle, good2 / good1.

Parameters:

Name	Type	Description	Default
agent	BarterTraderView	The trader.	required
system	SystemView	The system view of the appraisal.	required

Returns:

Name	Type	Description
mrs	float	The marginal rate of substitution.

Source code in src/skcomplex/systems/economics/markets/_barter_networks.py

@staticmethod
def get_rate(agent: BarterTraderView, system: SystemView) -> float:
    """Return the marginal rate of substitution of the bundle, `good2 / good1`.

    Args:
        agent:
            The trader.

        system:
            The system view of the appraisal.

    Returns:
        mrs:
            The marginal rate of substitution.
    """
    return agent.good2 / agent.good1

get_rate_gap(trader, other) staticmethod

Return the gap between the logarithms of the rates of substitution of a trader and a trading partner.

The gap is minus infinity for a trader that is not a partner, so a trader takes a partner whenever it has one.

Parameters:

Name	Type	Description	Default
trader	object	The trader that searches.	required
other	object	Another trader.	required

Returns:

Name	Type	Description
gap	float	The absolute difference of the logarithms of the two rates.

Source code in src/skcomplex/systems/economics/markets/_barter_networks.py

@staticmethod
def get_rate_gap(trader: object, other: object) -> float:
    """Return the gap between the logarithms of the rates of substitution of a trader and a trading partner.

    The gap is minus infinity for a trader that is not a partner, so a trader takes a partner whenever it has one.

    Args:
        trader:
            The trader that searches.

        other:
            Another trader.

    Returns:
        gap:
            The absolute difference of the logarithms of the two rates.
    """
    if not BarterTraderElement.is_partner(trader, other):
        return -math.inf
    if not isinstance(trader, BarterTraderView) or not isinstance(other, BarterTraderView):
        return -math.inf
    return abs(math.log(other.good2 / other.good1) - math.log(trader.good2 / trader.good1))

is_partner(trader, other) staticmethod

Return whether another trader is a trading partner of a trader, a neighbour on the trade network.

Parameters:

Name	Type	Description	Default
trader	object	The trader that searches.	required
other	object	Another trader.	required

Returns:

Name	Type	Description
partner	bool	Whether the other trader is in a node next to the trader's node.

Source code in src/skcomplex/systems/economics/markets/_barter_networks.py

@staticmethod
def is_partner(trader: object, other: object) -> bool:
    """Return whether another trader is a trading partner of a trader, a neighbour on the trade network.

    Args:
        trader:
            The trader that searches.

        other:
            Another trader.

    Returns:
        partner:
            Whether the other trader is in a node next to the trader's node.
    """
    if not isinstance(trader, BarterTraderView) or not isinstance(other, BarterTraderView):
        return False
    return other.id != trader.id and other.location in trader.neighbor_cells()

BartererElement(label, visual=VisualSpec('agents', {'color_by': 'chocolate_weight', 'color_range': (0.0, 1.0)}), *, chocolate=5.0, cigarettes=5.0, chocolate_weight=0.5, cigarettes_weight=0.5)

Bases: Element

Barterer who holds chocolate and cigarettes and values them with Cobb-Douglas weights.

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': 'chocolate_weight', 'color_range': (0.0, 1.0)})
chocolate	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The holding of chocolate.	5.0
cigarettes	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The holding of cigarettes.	5.0
chocolate_weight	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The Cobb-Douglas weight of chocolate.	0.5
cigarettes_weight	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The Cobb-Douglas weight of cigarettes.	0.5

Examples:

>>> from skcomplex.systems.economics.markets import BartererElement
>>> GroupElement('barterers', element=BartererElement, size=2, values={'chocolate': [8.0, 2.0]}).size
2

Source code in src/skcomplex/systems/economics/markets/_barter.py

def __init__(
    self,
    label: str,
    visual: VisualSpec | None = VisualSpec('agents', {'color_by': 'chocolate_weight', 'color_range': (0.0, 1.0)}),
    *,
    chocolate: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 5.0,
    cigarettes: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 5.0,
    chocolate_weight: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.5,
    cigarettes_weight: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.5,
) -> None:
    super().__init__(label, visual=visual)
    self.chocolate = chocolate
    self.cigarettes = cigarettes
    self.chocolate_weight = chocolate_weight
    self.cigarettes_weight = cigarettes_weight

BilateralBarterInteraction(label, *, requesters, providers, rule=RandomSearchSpec(), limit='limit', quantity='quantity', allowance='allowance', transfers=(), phase=None, on_match=None, units=100)

Bases: MatchingInteraction[SystemView]

Barter in which every trader trades with the neighbour it matched, the one whose rate differs the most.

A greedy matching pairs every trader, in random order, with the neighbour whose marginal rate of substitution differs the most in logarithm, and a neighbour takes any number of partners. The score of a trader that is not a neighbour is minus infinity, and a trader without neighbours trades with no one. Each pair trades single units of the first good at the price (g2 + g2') / (g1 + g1') of their pooled bundles, which lies between their rates, while the trade raises the utility of both. Besides the reporters of a matching, the reporters trades and searches count the units traded and the neighbours searched in a step.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction.	required
requesters	Annotated[str, RoleSpec()]	The path of the group of traders that search for a partner.	required
providers	Annotated[str, RoleSpec()]	The path of the group of traders that can be found, the same group.	required
rule	MatchingRule	The spec of the algorithm that pairs the traders, such as PartnerSpec('local', clearing='provider').	RandomSearchSpec()
limit	str | tuple[str, str]	The state with the limit of both sides, missing for traders, which then accept any partner.	'limit'
quantity	str | tuple[str, str]	The states of the partners a requester wants and a provider takes, one and without limit for traders.	'quantity'
allowance	str	The state of a requester with the most value it can give, missing for traders.	'allowance'
transfers	Sequence[tuple[str, int]]	The states every match moves, none for traders, whose trades move the goods.	()
phase	str | None	The phase the barter runs in, or None for the default phase.	None
on_match	Callable[[MatchSpec, SystemView], None] | None	A function of every match and the system view, run after the trade, or None.	None
units	Annotated[int, ParameterSpec(domain=NN)]	The most units one pair trades in a step.	100

Examples:

>>> from skcomplex.systems.economics.markets import BilateralBarterInteraction
>>> BilateralBarterInteraction('barter', requesters='traders', providers='traders').units
100

Source code in src/skcomplex/systems/economics/markets/_barter_networks.py

def __init__(
    self,
    label: str,
    *,
    requesters: Annotated[str, RoleSpec()],
    providers: Annotated[str, RoleSpec()],
    rule: MatchingRule = RandomSearchSpec(),
    limit: str | tuple[str, str] = 'limit',
    quantity: str | tuple[str, str] = 'quantity',
    allowance: str = 'allowance',
    transfers: Sequence[tuple[str, int]] = (),
    phase: str | None = None,
    on_match: Callable[[MatchSpec, SystemView], None] | None = None,
    units: Annotated[int, ParameterSpec(domain=NN)] = 100,
) -> 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.units = units

match(system)

Pair and trade, and report the units traded and the neighbours searched besides the matching reporters.

Parameters:

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

Source code in src/skcomplex/systems/economics/markets/_barter_networks.py

def match(self, system: SystemView) -> None:
    """Pair and trade, and report the units traded and the neighbours searched besides the matching reporters.

    Args:
        system:
            The system view of the interaction.
    """
    system.storage['trades'] = 0.0
    super().match(system)
    system.report('trades', self._get_trades(system))
    traders = system.agents('requesters', BarterTraderView)
    system.report('searches', float(sum(trader.links for trader in traders)))

BreweryInteraction(label, *, factory)

Bases: TransferInteraction

Brewery of a supply chain, which brews the orders of the factory over its production delay.

The brewing is a transfer from the brewery into the beer in production at the factory and on to its inventory.

Parameters:

Name	Type	Description	Default
label	str	The label of the brewery.	required
factory	Annotated[str, RoleSpec()]	The path of the factory.	required

Examples:

>>> from skcomplex.systems.economics.markets import BreweryInteraction
>>> BreweryInteraction('brewery', factory='factory').factory
'factory'

Source code in src/skcomplex/systems/economics/markets/_supply_chains.py

def __init__(
    self,
    label: str,
    *,
    factory: Annotated[str, RoleSpec()],
) -> None:
    super().__init__(
        label,
        source=factory,
        target=factory,
        state=Source('brewery'),
        into='in_transit',
        amount=lambda s: BreweryInteraction._orders(s.factory),
        flow='starts',
        legs=[('in_transit', 'inventory', lambda s: BreweryInteraction._completions(s.factory), 'completions')],
    )
    self.factory = factory

BudgetHouseholdElement(label, visual=None, *, demand=0.0, expenditure=0.0, utility=0.0, price=1.0, budget=100.0, alpha=0.1, elasticity=-1.0, change=0.1, initial_price=1.0, kind='cobb-douglas')

Bases: Element

Household with a budget that buys a good at a price that rises by a share every step.

With the demand 'cobb-douglas' the household maximizes A^alpha B^(1 - alpha) against a composite good B of price 1, so it buys A = alpha m / p and spends the share alpha of its budget m on the good. With the demand 'elasticity' it spends the budget share alpha at the initial price p0 and its demand changes with its own-price elasticity, A = (alpha m / p0) (p / p0)^elasticity.

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
demand	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The demand for the good.	0.0
expenditure	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The expenditure on the good.	0.0
utility	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The Cobb-Douglas utility of the good and the composite good bought.	0.0
price	Annotated[float, StateSpec(domain=open(0, oo))]	The price of the good.	1.0
budget	Annotated[float, StateSpec(domain=open(0, oo))]	The budget of the household.	100.0
alpha	Annotated[float, StateSpec(domain=closed(0, 1))]	The budget share of the good, the exponent of the Cobb-Douglas utility.	0.1
elasticity	Annotated[float, StateSpec()]	The own-price elasticity of the demand.	-1.0
change	Annotated[float, ParameterSpec(domain=open(-1, oo))]	The relative rise of the price every step.	0.1
initial_price	Annotated[float, ParameterSpec(domain=open(0, oo))]	The price at the start.	1.0
kind	str	The demand, 'cobb-douglas' or 'elasticity'.	'cobb-douglas'

Raises:

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

Examples:

>>> from skcomplex.core.components.elements import GroupElement
>>> from skcomplex.systems.economics.markets import BudgetHouseholdElement
>>> GroupElement('households', element=BudgetHouseholdElement, size=2, values={'budget': [50.0, 150.0]}).size
2

Source code in src/skcomplex/systems/economics/markets/_household_demand.py

def __init__(
    self,
    label: str,
    visual: VisualSpec | None = None,
    *,
    demand: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
    expenditure: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
    utility: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
    price: Annotated[float, StateSpec(domain=RealSet.open(0, oo))] = 1.0,
    budget: Annotated[float, StateSpec(domain=RealSet.open(0, oo))] = 100.0,
    alpha: Annotated[float, StateSpec(domain=RealSet.closed(0, 1))] = 0.1,
    elasticity: Annotated[float, StateSpec()] = -1.0,
    change: Annotated[float, ParameterSpec(domain=RealSet.open(-1, oo))] = 0.1,
    initial_price: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 1.0,
    kind: str = 'cobb-douglas',
) -> None:
    super().__init__(label, visual=visual)
    self.demand = demand
    self.expenditure = expenditure
    self.utility = utility
    self.price = price
    self.budget = budget
    self.alpha = alpha
    self.elasticity = elasticity
    self.change = change
    self.initial_price = initial_price
    self.kind = kind

get_demand(s, price)

Return the demand of the household at a price.

Parameters:

Name	Type	Description	Default
s	SimpleNamespace	The namespace of the household.	required
price	Expression	The price.	required

Returns:

Name	Type	Description
demand	Expression	The demand for the good.

Source code in src/skcomplex/systems/economics/markets/_household_demand.py

def get_demand(self, s: SimpleNamespace, price: Expression) -> Expression:
    """Return the demand of the household at a price.

    Args:
        s:
            The namespace of the household.

        price:
            The price.

    Returns:
        demand:
            The demand for the good.
    """
    if self.kind == 'cobb-douglas':
        return s.alpha * s.budget / price
    return s.alpha * s.budget / s.initial_price * (price / s.initial_price) ** s.elasticity

CarDealershipElement(label, visual=None, *, inventory=200.0, demand=20.0, demand_step=2.0, step_time=25.0, coverage=10.0, perception_delay=5.0, response_delay=3.0, delivery_delay=5.0)

Bases: Element

Car dealership whose inventory deliveries fill and sales drain, ordered to close the gap to a desired coverage.

Sales are the demand, stepping up at step_time, limited by the inventory. The dealer perceives sales through a first-order smoothing, desires coverage days of perceived sales, and orders the perceived sales plus the gap to the desired inventory over the response delay, never negative. Orders arrive through a third-order delay. The dealership holds the inventory and the parameters, and a transfer within it of the system moves the cars.

Parameters:

Name	Type	Description	Default
label	str	The label of the dealership.	required
visual	VisualSpec | None	How the element is drawn, or None for default_visual_.	None
inventory	Annotated[float, StateSpec(domain=closed_open(0, oo), quantity=cars)]	The initial number of cars on the lot.	200.0
demand	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The customer demand in cars per day.	20.0
demand_step	Annotated[float, ParameterSpec()]	The permanent change of the demand.	2.0
step_time	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The day of the change of the demand.	25.0
coverage	Annotated[float, ParameterSpec(domain=open(0, oo))]	The days of perceived sales the dealer wants on the lot.	10.0
perception_delay	Annotated[float, ParameterSpec(domain=open(0, oo))]	The time over which the dealer averages sales.	5.0
response_delay	Annotated[float, ParameterSpec(domain=open(0, oo))]	The time over which the dealer closes the inventory gap.	3.0
delivery_delay	Annotated[float, ParameterSpec(domain=open(0, oo))]	The average time between an order and its delivery.	5.0

Examples:

>>> from skcomplex.systems.economics.markets import CarDealershipElement
>>> CarDealershipElement('dealer', inventory=150.0).inventory
150.0

Source code in src/skcomplex/systems/economics/markets/_inventory.py

def __init__(
    self,
    label: str,
    visual: VisualSpec | None = None,
    *,
    inventory: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo), quantity='cars')] = 200.0,
    demand: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 20.0,
    demand_step: Annotated[float, ParameterSpec()] = 2.0,
    step_time: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 25.0,
    coverage: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 10.0,
    perception_delay: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 5.0,
    response_delay: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 3.0,
    delivery_delay: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 5.0,
) -> None:
    super().__init__(label, visual=visual)
    self.inventory = inventory
    self.demand = demand
    self.demand_step = demand_step
    self.step_time = step_time
    self.coverage = coverage
    self.perception_delay = perception_delay
    self.response_delay = response_delay
    self.delivery_delay = delivery_delay

CarElement(label, visual=None, *, stage=WASH, work=525.0, crew=0)

Bases: Element

Car at the wash, waiting for or receiving a wash or a valet.

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
stage	Annotated[float, StateSpec(domain=NN)]	0 for the wash and 1 for the valet.	WASH
work	Annotated[float, StateSpec()]	The person-seconds of work left at the current stage.	525.0
crew	Annotated[float, StateSpec(domain=NN)]	The number of workers on the car in the current second.	0

Examples:

>>> from skcomplex.core.components.elements import GroupElement
>>> from skcomplex.systems.economics.markets import CarElement
>>> GroupElement('cars', element=CarElement, size=1).size
1

Source code in src/skcomplex/systems/economics/markets/_queues.py

def __init__(
    self,
    label: str,
    visual: VisualSpec | None = None,
    *,
    stage: Annotated[float, StateSpec(domain=NN)] = WASH,
    work: Annotated[float, StateSpec()] = 525.0,
    crew: Annotated[float, StateSpec(domain=NN)] = 0,
) -> None:
    super().__init__(label, visual=visual)
    self.stage = stage
    self.work = work
    self.crew = crew

CarWashOperationInteraction(label, *, cars, till, p_pass=0.33, p_enter=0.005, p_valet=0.5, wash_work=525.0, valet_work=1750.0, wash_price=4.0, valet_price=10.0, balk=3.0)

Bases: Interaction

Arrival of passing cars, completion of washes and valets, and payment, once every second.

A passing car turns in with a probability, unless the wash queue holds balk cars or more. A washed car stays for a valet with a probability, and pays the wash price or the valet price when it leaves.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction.	required
cars	Annotated[str, RoleSpec()]	The path of the group of cars.	required
till	Annotated[str, RoleSpec()]	The path of the group that holds the till.	required
p_pass	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The probability that a car passes in a second.	0.33
p_enter	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The probability that a passing car turns in.	0.005
p_valet	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The probability that a washed car stays for a valet.	0.5
wash_work	Annotated[float, ParameterSpec(domain=open(0, oo))]	The person-seconds of a wash.	525.0
valet_work	Annotated[float, ParameterSpec(domain=open(0, oo))]	The person-seconds of a valet.	1750.0
wash_price	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The price of a wash.	4.0
valet_price	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The price of a wash with a valet.	10.0
balk	Annotated[float, ParameterSpec(domain=open(0, oo))]	The length of the wash queue at which drivers no longer turn in.	3.0

Examples:

>>> from skcomplex.systems.economics.markets import CarWashOperationInteraction
>>> CarWashOperationInteraction('wash', cars='cars', till='till', p_enter=0.01).p_enter
0.01

Source code in src/skcomplex/systems/economics/markets/_queues.py

def __init__(
    self,
    label: str,
    *,
    cars: Annotated[str, RoleSpec()],
    till: Annotated[str, RoleSpec()],
    p_pass: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.33,
    p_enter: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.005,
    p_valet: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.5,
    wash_work: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 525.0,
    valet_work: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 1750.0,
    wash_price: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 4.0,
    valet_price: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 10.0,
    balk: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 3.0,
) -> None:
    super().__init__(label)
    self.cars = cars
    self.till = till
    self.p_pass = p_pass
    self.p_enter = p_enter
    self.p_valet = p_valet
    self.wash_work = wash_work
    self.valet_work = valet_work
    self.wash_price = wash_price
    self.valet_price = valet_price
    self.balk = balk

operate(system)

Finish the cars whose work is done, reset the crews, and let a passing car turn in or drive on.

Parameters:

Name	Type	Description	Default
system	SystemView	The system.	required

Source code in src/skcomplex/systems/economics/markets/_queues.py

def operate(self, system: SystemView) -> None:
    """Finish the cars whose work is done, reset the crews, and let a passing car turn in or drive on.

    Args:
        system:
            The system.
    """
    values = {name: float(system.parameter(name)) for name in _PARAMETERS}
    till = system.agents('till', TillView)[0]
    for car in system.agents('cars', CarView):
        car.crew = 0
        if car.work <= 0:
            self._finish(car, till, system, values)
    if system.rng.random() >= values['p_pass'] * values['p_enter']:
        return
    if sum(car.stage == WASH for car in system.agents('cars', CarView)) >= values['balk']:
        till.turned_away += 1
        return
    system.create('cars', stage=WASH, work=values['wash_work'])

CobwebInteraction(label, *, target, a=10.0, b=2.0, c=1.0, d=1.0)

Bases: MapInteraction

Cobweb system, where supply reacts to the previous price and demand clears the market.

Demand is a - b * price and supply is c + d * previous price, so the next price is (a - c - d * price) / b, a map of the state price of the market.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction.	required
target	Annotated[str, RoleSpec()]	The path of the market.	required
a	Annotated[float, ParameterSpec()]	The demand intercept.	10.0
b	Annotated[float, ParameterSpec(domain=open(0, oo))]	The demand slope.	2.0
c	Annotated[float, ParameterSpec()]	The supply intercept.	1.0
d	Annotated[float, ParameterSpec(domain=open(0, oo))]	The supply slope.	1.0

Examples:

>>> from skcomplex.systems.economics.markets import CobwebInteraction
>>> CobwebInteraction('cobweb', target='wheat', a=10, b=2, c=1, d=1).b
2

Source code in src/skcomplex/systems/economics/markets/_cobweb.py

def __init__(
    self,
    label: str,
    *,
    target: Annotated[str, RoleSpec()],
    a: Annotated[float, ParameterSpec()] = 10.0,
    b: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 2.0,
    c: Annotated[float, ParameterSpec()] = 1.0,
    d: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 1.0,
) -> None:
    super().__init__(label, target=target, state='price', function=self._get_price)
    self.a = a
    self.b = b
    self.c = c
    self.d = d

CommodityMarketElement(label, visual=None, *, inventory=25.0, capacity=100.0, demand=100.0, demand_step=10.0, step_time=2.0, desired_coverage=0.25, reference_price=1.0, supply_elasticity=0.1, demand_elasticity=0.25, expectation_time=1.0, construction_delay=3.0, lifetime=15.0)

Bases: Element

Commodity market whose inventory production fills and consumption drains, with capacity built after a delay.

The price is a falling table function of the coverage of the demand by the inventory relative to the desired coverage. Production is the capacity times the relative price to the supply elasticity, consumption the demand times the relative price to minus the demand elasticity. Capacity is ordered at the rate of retirements times a rising table function of the smoothed price, completed through a third-order delay, and retired over its lifetime. The market holds the stocks and the parameters, and transfers within it of the system move the commodity and the capacity.

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
inventory	Annotated[float, StateSpec(domain=closed_open(0, oo), quantity=commodity)]	The initial inventory of the commodity.	25.0
capacity	Annotated[float, StateSpec(domain=closed_open(0, oo), quantity=capacity)]	The initial production capacity per year.	100.0
demand	Annotated[float, ParameterSpec(domain=open(0, oo))]	The demand per year at the reference price.	100.0
demand_step	Annotated[float, ParameterSpec()]	The permanent change of the demand.	10.0
step_time	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The year of the change of the demand.	2.0
desired_coverage	Annotated[float, ParameterSpec(domain=open(0, oo))]	The years of demand producers want the inventory to hold.	0.25
reference_price	Annotated[float, ParameterSpec(domain=open(0, oo))]	The price at the desired coverage.	1.0
supply_elasticity	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The elasticity of the utilization of capacity with respect to the price.	0.1
demand_elasticity	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The elasticity of consumption with respect to the price, as a positive number.	0.25
expectation_time	Annotated[float, ParameterSpec(domain=open(0, oo))]	The time over which producers smooth the price into their expected price.	1.0
construction_delay	Annotated[float, ParameterSpec(domain=open(0, oo))]	The average time between an order of capacity and its completion.	3.0
lifetime	Annotated[float, ParameterSpec(domain=open(0, oo))]	The average lifetime of capacity.	15.0

Examples:

>>> from skcomplex.systems.economics.markets import CommodityMarketElement
>>> CommodityMarketElement('market', lifetime=20.0).lifetime
20.0

Source code in src/skcomplex/systems/economics/markets/_commodities.py

def __init__(
    self,
    label: str,
    visual: VisualSpec | None = None,
    *,
    inventory: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo), quantity='commodity')] = 25.0,
    capacity: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo), quantity='capacity')] = 100.0,
    demand: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 100.0,
    demand_step: Annotated[float, ParameterSpec()] = 10.0,
    step_time: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 2.0,
    desired_coverage: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 0.25,
    reference_price: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 1.0,
    supply_elasticity: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.1,
    demand_elasticity: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.25,
    expectation_time: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 1.0,
    construction_delay: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 3.0,
    lifetime: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 15.0,
) -> None:
    super().__init__(label, visual=visual)
    self.inventory = inventory
    self.capacity = capacity
    self.demand = demand
    self.demand_step = demand_step
    self.step_time = step_time
    self.desired_coverage = desired_coverage
    self.reference_price = reference_price
    self.supply_elasticity = supply_elasticity
    self.demand_elasticity = demand_elasticity
    self.expectation_time = expectation_time
    self.construction_delay = construction_delay
    self.lifetime = lifetime

ConsumerElement(label, visual=VisualSpec('agents', {'color_by': 'fair', 'color_range': (0.0, 1.0)}), *, fair=0.0, ethics=0.5, budget=2.0, premium=0.3, ethics_weight=2.0, price_weight=5.0, social_weight=2.0)

Bases: Element

Consumer who buys fair-trade or regular food, with an ethical concern, a budget, and the weights of its utility.

The utility of fair food is ethics_weight * ethics - price_weight * premium / budget + social_weight * share, where share is the share of neighbours who bought fair food in the last step, and the utility of regular food is zero. A consumer whose budget does not cover the price 1 + premium of fair food buys regular food.

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': 'fair', 'color_range': (0.0, 1.0)})
fair	Annotated[float, StateSpec(domain=FiniteEnumeratedSet([0, 1]))]	One when the consumer bought fair food in the last step, zero otherwise.	0.0
ethics	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The ethical concern of the consumer, between zero and one.	0.5
budget	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The money the consumer can spend on one unit of food, regular food costing one.	2.0
premium	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The extra price of fair food over regular food.	0.3
ethics_weight	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The weight of ethical concern in the utility of fair food.	2.0
price_weight	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The weight of the premium relative to the budget.	5.0
social_weight	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The weight of the share of neighbours who buy fair food.	2.0

Examples:

>>> from skcomplex.systems.economics.markets import ConsumerElement
>>> GroupElement('consumers', element=ConsumerElement, size=4, values={'ethics': 0.5}).size
4

Source code in src/skcomplex/systems/economics/markets/_fair_food.py

def __init__(
    self,
    label: str,
    visual: VisualSpec | None = VisualSpec('agents', {'color_by': 'fair', 'color_range': (0.0, 1.0)}),
    *,
    fair: Annotated[float, StateSpec(domain=FiniteEnumeratedSet([0, 1]))] = 0.0,
    ethics: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.5,
    budget: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 2.0,
    premium: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.3,
    ethics_weight: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 2.0,
    price_weight: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 5.0,
    social_weight: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 2.0,
) -> None:
    super().__init__(label, visual=visual)
    self.fair = fair
    self.ethics = ethics
    self.budget = budget
    self.premium = premium
    self.ethics_weight = ethics_weight
    self.price_weight = price_weight
    self.social_weight = social_weight

CournotAdjustmentInteraction(label, *, target, a=1500.0, b=10.0, error=0.0)

Bases: MapInteraction

Simultaneous best responses of two firms to their beliefs about the rival's last output.

Under the inverse demand P = a - b Q, each firm sets q_i = (a - c_i) / (2 b) - belief_j / 2, where the belief is the rival's last output scaled by a factor drawn uniformly from [1 - error, 1 + error]. It is a synchronous map of the agent form of the outputs, so both firms respond to the outputs of the step before.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction.	required
target	Annotated[str, RoleSpec()]	The path of the group of the two firms.	required
a	Annotated[float, ParameterSpec(domain=open(0, oo))]	The intercept of the inverse demand.	1500.0
b	Annotated[float, ParameterSpec(domain=open(0, oo))]	The slope of the inverse demand.	10.0
error	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The largest relative error of a firm's belief about the rival's output.	0.0

Examples:

>>> from skcomplex.systems.economics.markets import CournotAdjustmentInteraction
>>> CournotAdjustmentInteraction('market', target='firms', error=0.1).error
0.1

Source code in src/skcomplex/systems/economics/markets/_duopoly.py

def __init__(
    self,
    label: str,
    *,
    target: Annotated[str, RoleSpec()],
    a: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 1500.0,
    b: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 10.0,
    error: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
) -> None:
    super().__init__(label, target=target, state='output', function=self._get_response, form='agent')
    self.a = a
    self.b = b
    self.error = error

CournotBestResponseElement(label, *, q_i=0.0, q_j=0.0, demand=10.0, c_i=1.0, c_j=2.0, lambda_i=0.0, lambda_j=0.0)

Bases: Element

Quantities of two firms at the best responses of a Cournot game, where each firm holds a share of the other.

The conditions are alpha - 2 q_i - q_j = c_i / (1 + lambda_i) and its mirror (Forrest 2018 Eq. 21.5), whose solution is q_i = (alpha - 2 c_i / (1 + lambda_i) + c_j / (1 + lambda_j)) / 3 (Eq. 21.6).

Parameters:

Name	Type	Description	Default
label	str	The label of the element.	required
q_i	Annotated[float, StateSpec()]	The initial guess of the quantity of the first firm, which the equations replace.	0.0
q_j	Annotated[float, StateSpec()]	The initial guess of the quantity of the second firm, which the equations replace.	0.0
demand	Annotated[float, ParameterSpec()]	The intercept alpha of the inverse demand.	10.0
c_i	Annotated[float, ParameterSpec()]	The unit cost of the first firm.	1.0
c_j	Annotated[float, ParameterSpec()]	The unit cost of the second firm.	2.0
lambda_i	Annotated[float, ParameterSpec()]	The cross holding of the first firm.	0.0
lambda_j	Annotated[float, ParameterSpec()]	The cross holding of the second firm.	0.0

Examples:

>>> from skcomplex.systems.economics.markets import CournotBestResponseElement
>>> CournotBestResponseElement('duopoly', demand=8.0).demand
8.0

Source code in src/skcomplex/systems/economics/markets/_oligopoly.py

def __init__(
    self,
    label: str,
    *,
    q_i: Annotated[float, StateSpec()] = 0.0,
    q_j: Annotated[float, StateSpec()] = 0.0,
    demand: Annotated[float, ParameterSpec()] = 10.0,
    c_i: Annotated[float, ParameterSpec()] = 1.0,
    c_j: Annotated[float, ParameterSpec()] = 2.0,
    lambda_i: Annotated[float, ParameterSpec()] = 0.0,
    lambda_j: Annotated[float, ParameterSpec()] = 0.0,
) -> None:
    super().__init__(label)
    self.q_i = q_i
    self.q_j = q_j
    self.demand = demand
    self.c_i = c_i
    self.c_j = c_j
    self.lambda_i = lambda_i
    self.lambda_j = lambda_j

CustomerElement(label, visual=None, *, s1=0.5, s2=0.5, shop=0.0, freshness=1.0, T=0.1)

Bases: Element

Customer who remembers how satisfied it was with the products of each of two shops.

The customer visits the shop k of larger satisfaction S_k more often, with the logit utility S_k / T, and holds the freshness of the product it bought last.

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
s1	Annotated[float, StateSpec()]	The satisfaction with the first shop.	0.5
s2	Annotated[float, StateSpec()]	The satisfaction with the second shop.	0.5
shop	Annotated[float, StateSpec(domain=FiniteEnumeratedSet([0, 1]))]	The index of the shop the customer visits, 0 or 1.	0.0
freshness	Annotated[float, StateSpec(domain=closed(0, 1))]	The freshness of the product the customer bought last.	1.0
T	Annotated[float, ParameterSpec(domain=open(0, oo))]	The temperature of the choice, the randomness of the customer.	0.1

Examples:

>>> from skcomplex.core.components.elements import GroupElement
>>> from skcomplex.systems.economics.markets import CustomerElement
>>> GroupElement('customers', element=CustomerElement, size=10, values={'s1': 0.5, 's2': 0.5}).size
10

Source code in src/skcomplex/systems/economics/markets/_perishables.py

def __init__(
    self,
    label: str,
    visual: VisualSpec | None = None,
    *,
    s1: Annotated[float, StateSpec()] = 0.5,
    s2: Annotated[float, StateSpec()] = 0.5,
    shop: Annotated[float, StateSpec(domain=FiniteEnumeratedSet([0, 1]))] = 0.0,
    freshness: Annotated[float, StateSpec(domain=RealSet.closed(0, 1))] = 1.0,
    T: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 0.1,
) -> None:
    super().__init__(label, visual=visual)
    self.s1 = s1
    self.s2 = s2
    self.shop = shop
    self.freshness = freshness
    self.T = T

destination(agent, system) staticmethod

Return the node of the shop the customer chose.

Parameters:

Name	Type	Description	Default
agent	AgentViewProtocol	The customer.	required
system	SystemViewProtocol	The system view of the movement.	required

Returns:

Name	Type	Description
node	int | None	The node of the shop, or None to stay for an agent that is not a customer.

Source code in src/skcomplex/systems/economics/markets/_perishables.py

@staticmethod
def destination(agent: AgentViewProtocol, system: SystemViewProtocol) -> int | None:
    """Return the node of the shop the customer chose.

    Args:
        agent:
            The customer.

        system:
            The system view of the movement.

    Returns:
        node:
            The node of the shop, or `None` to stay for an agent that is not a customer.
    """
    return int(agent.shop) if isinstance(agent, CustomerView) else None

payoff(agent, system, option) staticmethod

Return the freshness of the product the customer bought last, the payoff of the shop it visited.

Parameters:

Name	Type	Description	Default
agent	CustomerView	The customer.	required
system	SystemView	The system view of the reinforcement.	required
option	int	The index of the shop the customer visited.	required

Returns:

Name	Type	Description
freshness	float	The freshness of the product.

Source code in src/skcomplex/systems/economics/markets/_perishables.py

@staticmethod
def payoff(agent: CustomerView, system: SystemView, option: int) -> float:
    """Return the freshness of the product the customer bought last, the payoff of the shop it visited.

    Args:
        agent:
            The customer.

        system:
            The system view of the reinforcement.

        option:
            The index of the shop the customer visited.

    Returns:
        freshness:
            The freshness of the product.
    """
    return agent.freshness

preference(agent) staticmethod

Return p_1 - p_2 = tanh((S_1 - S_2) / 2T), how much more likely the customer visits the first shop.

Parameters:

Name	Type	Description	Default
agent	CustomerView	The customer.	required

Returns:

Name	Type	Description
preference	float	The difference of the probabilities of visiting the first and the second shop.

Source code in src/skcomplex/systems/economics/markets/_perishables.py

@staticmethod
def preference(agent: CustomerView) -> float:
    """Return `p_1 - p_2 = tanh((S_1 - S_2) / 2T)`, how much more likely the customer visits the first shop.

    Args:
        agent:
            The customer.

    Returns:
        preference:
            The difference of the probabilities of visiting the first and the second shop.
    """
    return float(np.tanh((agent.s1 - agent.s2) / (2 * agent.T)))

utility(agent, system, option) staticmethod

Return the utility S_k / T of visiting shop k.

Parameters:

Name	Type	Description	Default
agent	CustomerView	The customer.	required
system	SystemView	The system view of the choice.	required
option	float	The index of the shop.	required

Returns:

Name	Type	Description
utility	float	The satisfaction with the shop divided by the temperature.

Source code in src/skcomplex/systems/economics/markets/_perishables.py

@staticmethod
def utility(agent: CustomerView, system: SystemView, option: float) -> float:
    """Return the utility `S_k / T` of visiting shop `k`.

    Args:
        agent:
            The customer.

        system:
            The system view of the choice.

        option:
            The index of the shop.

    Returns:
        utility:
            The satisfaction with the shop divided by the temperature.
    """
    return (agent.s1 if option == 0 else agent.s2) / agent.T

DuopolistElement(label, visual=None, *, output=50.0, cost=0.0)

Bases: Element

Firm of a Cournot duopoly, described by its output and its constant marginal cost.

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
output	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The initial output.	50.0
cost	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The marginal cost.	0.0

Examples:

>>> from skcomplex.core.components.elements import GroupElement
>>> from skcomplex.systems.economics.markets import DuopolistElement
>>> GroupElement('firms', element=DuopolistElement, size=2, values={'output': [100.0, 100.0]}).size
2

Source code in src/skcomplex/systems/economics/markets/_duopoly.py

def __init__(
    self,
    label: str,
    visual: VisualSpec | None = None,
    *,
    output: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 50.0,
    cost: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
) -> None:
    super().__init__(label, visual=visual)
    self.output = output
    self.cost = cost

EchelonElement(label, visual=None, *, inventory=12.0, in_transit=8.0, backlog=4.0, expected=4.0, supply_line=16.0, desired=13.44, theta=0.36, alpha=0.26, beta=0.34, dispatch_time=1.0, mail_delay=2.0, shipping_delay=2.0)

Bases: Element

Echelon of a supply chain, which ships its backlog from its inventory and orders by anchoring and adjustment.

It ships its backlog over the dispatch time, limited by its inventory. Its expected demand adapts to incoming orders at the rate theta, and it orders its expected demand plus alpha times the gap between its desired stock and its effective inventory, the inventory less the backlog, less beta times its supply line, never negative. Its orders reach its supplier after the mail delay, and the beer its supplier ships arrives after the shipping delay.

Parameters:

Name	Type	Description	Default
label	str	The label of the echelon.	required
visual	VisualSpec | None	How the element is drawn, or None for default_visual_.	None
inventory	Annotated[float, StateSpec(domain=closed_open(0, oo), quantity=beer)]	The initial inventory, in cases.	12.0
in_transit	Annotated[float, StateSpec(domain=closed_open(0, oo), quantity=beer)]	The initial beer on its way to the echelon.	8.0
backlog	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The initial orders waiting to ship.	4.0
expected	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The initial expected demand.	4.0
supply_line	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The initial beer ordered and not received.	16.0
desired	Annotated[float, ParameterSpec()]	The desired stock S'.	13.44
theta	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The rate at which the expected demand adapts to incoming orders.	0.36
alpha	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The fraction of the stock gap ordered per week.	0.26
beta	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The weight of the supply line in the stock gap.	0.34
dispatch_time	Annotated[float, ParameterSpec(domain=open(0, oo))]	The time over which the echelon ships its backlog.	1.0
mail_delay	Annotated[float, ParameterSpec(domain=open(0, oo))]	The average time an order of the echelon takes to reach its supplier.	2.0
shipping_delay	Annotated[float, ParameterSpec(domain=open(0, oo))]	The average time a shipment takes to reach the echelon from its supplier.	2.0

Examples:

>>> from skcomplex.systems.economics.markets import EchelonElement
>>> EchelonElement('retailer', inventory=10.0).inventory
10.0

Source code in src/skcomplex/systems/economics/markets/_supply_chains.py

def __init__(
    self,
    label: str,
    visual: VisualSpec | None = None,
    *,
    inventory: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo), quantity='beer')] = 12.0,
    in_transit: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo), quantity='beer')] = 8.0,
    backlog: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 4.0,
    expected: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 4.0,
    supply_line: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 16.0,
    desired: Annotated[float, ParameterSpec()] = 13.44,
    theta: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.36,
    alpha: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.26,
    beta: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.34,
    dispatch_time: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 1.0,
    mail_delay: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 2.0,
    shipping_delay: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 2.0,
) -> None:
    super().__init__(label, visual=visual)
    self.inventory = inventory
    self.in_transit = in_transit
    self.backlog = backlog
    self.expected = expected
    self.supply_line = supply_line
    self.desired = desired
    self.theta = theta
    self.alpha = alpha
    self.beta = beta
    self.dispatch_time = dispatch_time
    self.mail_delay = mail_delay
    self.shipping_delay = shipping_delay

EmployerElement(label, visual=None, *, bid=0.0, demand=0.0, posted=-1.0, vacancy_rate=0.5, low_wage=0.5, high_wage=1.0)

Bases: Element

Firm that posts a vacancy with a random wage offer in some steps, filled by at most one worker.

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
bid	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The wage offered for the vacancy.	0.0
demand	Annotated[float, StateSpec(domain=NN)]	The number of open vacancies, one or zero.	0.0
posted	Annotated[float, StateSpec()]	The step in which the vacancy was posted.	-1.0
vacancy_rate	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The probability of posting a vacancy in a step.	0.5
low_wage	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The lowest wage offer.	0.5
high_wage	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The highest wage offer.	1.0

Examples:

>>> from skcomplex.core.components.elements import GroupElement
>>> from skcomplex.systems.economics.markets import EmployerElement
>>> GroupElement('firms', element=EmployerElement, size=5).size
5

Source code in src/skcomplex/systems/economics/markets/_labour.py

def __init__(
    self,
    label: str,
    visual: VisualSpec | None = None,
    *,
    bid: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
    demand: Annotated[float, StateSpec(domain=NN)] = 0.0,
    posted: Annotated[float, StateSpec()] = -1.0,
    vacancy_rate: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.5,
    low_wage: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.5,
    high_wage: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
) -> None:
    super().__init__(label, visual=visual)
    self.bid = bid
    self.demand = demand
    self.posted = posted
    self.vacancy_rate = vacancy_rate
    self.low_wage = low_wage
    self.high_wage = high_wage

get_posting() staticmethod

Return the states a firm sets in the phase of offers, in their order.

The firm withdraws its last vacancy and posts a new one at the vacancy rate, with a uniform wage offer and the step of posting.

Returns:

Name	Type	Description
posting	tuple[tuple[str, Callable[[EmployerView, SystemView], float]], ...]	The name of every state and the function of its value, of the firm and the system view.

Source code in src/skcomplex/systems/economics/markets/_labour.py

@staticmethod
def get_posting() -> tuple[tuple[str, Callable[[EmployerView, SystemView], float]], ...]:
    """Return the states a firm sets in the phase of offers, in their order.

    The firm withdraws its last vacancy and posts a new one at the vacancy rate, with a uniform wage offer and the
    step of posting.

    Returns:
        posting:
            The name of every state and the function of its value, of the firm and the system view.
    """
    return (
        ('demand', lambda agent, system: float(system.rng.random() < agent.vacancy_rate)),
        (
            'bid',
            lambda agent, system: (
                float(system.rng.uniform(agent.low_wage, agent.high_wage)) if agent.demand else agent.bid
            ),
        ),
        ('posted', lambda agent, system: system.time if agent.demand else agent.posted),
    )

ExchangeInteraction(label, *, source, target, goods=GOODS, pairing='pairs', quantity=0.1, phase='trade')

Bases: TransferInteraction

Bilateral barter between paired traders, a transfer whose trades repeat while both traders gain.

Every step the transfer pairs the traders, and every pair barters small quantities of every pair of goods at the geometric mean of their marginal rates of substitution while both gain. A trader holds every good in the state of its name and its Cobb-Douglas weight in the state <good>_weight. Every trade moves goods between the two traders, so the total of every good stays the same, which the default run checks as invariants. The exchange reports the mean price of the second good in units of the first, the standard deviation dispersion of its logarithm over the trades of the step, and the number of trades volume.

Parameters:

Name	Type	Description	Default
label	str	The label of the exchange.	required
source	str	The path of the group of traders.	required
target	str	The path of the group of traders, the source again.	required
goods	tuple[str, ...]	The names of the traded goods.	GOODS
pairing	str	How the transfer pairs the traders, such as 'neighbor-pairs' for disjoint pairs of neighbours or 'pairs' for disjoint random pairs.	'pairs'
quantity	float	The quantity of the bought good in every trade.	0.1
phase	str	The phase the trades run in.	'trade'

Examples:

>>> from skcomplex.systems.economics.markets import ExchangeInteraction
>>> ExchangeInteraction('exchange', source='traders', target='traders').reports
('price', 'dispersion', 'volume')

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

def __init__(
    self,
    label: str,
    *,
    source: str,
    target: str,
    goods: tuple[str, ...] = GOODS,
    pairing: str = 'pairs',
    quantity: float = 0.1,
    phase: str = 'trade',
) -> None:
    super().__init__(
        label,
        source=source,
        target=target,
        state=goods[0],
        amount=quantity,
        pairing=pairing,
        form='agent',
        phase=phase,
    )
    self.goods = goods
    self.quantity = quantity

report_trades(system)

Report the mean price of the second good in the first, its dispersion, and the trades of the step.

Parameters:

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

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

def report_trades(self, system: SystemViewProtocol) -> None:
    """Report the mean price of the second good in the first, its dispersion, and the trades of the step.

    Args:
        system:
            The system view of the exchange.
    """
    trades = system.storage.pop('trades', [])
    trades = trades if isinstance(trades, list) else []
    logs = [math.log(price) for bought, paid, price in trades if (bought, paid) == (0, 1)]
    system.report('price', math.exp(float(np.mean(logs))) if logs else math.nan)
    system.report('dispersion', float(np.std(logs)) if len(logs) > 1 else math.nan)
    system.report('volume', len(trades))

transfer(source, target, system)

Barter between two traders while both gain, and keep the trades of the step for its report.

Parameters:

Name	Type	Description	Default
source	AgentViewProtocol	The first trader.	required
target	AgentViewProtocol	The second trader.	required
system	SystemViewProtocol	The system view of the exchange.	required

Returns:

Name	Type	Description
traded	bool	Whether the traders made a trade.

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

def transfer(self, source: AgentViewProtocol, target: AgentViewProtocol, system: SystemViewProtocol) -> bool:
    """Barter between two traders while both gain, and keep the trades of the step for its report.

    Args:
        source:
            The first trader.

        target:
            The second trader.

        system:
            The system view of the exchange.

    Returns:
        traded:
            Whether the traders made a trade.
    """
    weights = [f'{good}_weight' for good in self.goods]
    trades = trade_goods(source, target, self.goods, weights, quantity=self.quantity)
    logs = system.storage.setdefault('trades', [])
    if isinstance(logs, list):
        logs.extend(trades)
    return bool(trades)

FactoryElement(label, visual=None, *, inventory=12.0, in_transit=8.0, backlog=4.0, expected=4.0, supply_line=16.0, desired=13.44, theta=0.36, alpha=0.26, beta=0.34, dispatch_time=1.0, mail_delay=2.0, shipping_delay=2.0, production_delay=3.0)

Bases: EchelonElement

Factory of a supply chain, an echelon whose orders a brewery brews over the production delay.

Parameters:

Name	Type	Description	Default
label	str	The label of the factory.	required
visual	VisualSpec | None	How the element is drawn, or None for default_visual_.	None
inventory	Annotated[float, StateSpec(domain=closed_open(0, oo), quantity=beer)]	The initial inventory, in cases.	12.0
in_transit	Annotated[float, StateSpec(domain=closed_open(0, oo), quantity=beer)]	The initial beer on its way to the echelon.	8.0
backlog	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The initial orders waiting to ship.	4.0
expected	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The initial expected demand.	4.0
supply_line	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The initial beer ordered and not received.	16.0
desired	Annotated[float, ParameterSpec()]	The desired stock S'.	13.44
theta	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The rate at which the expected demand adapts to incoming orders.	0.36
alpha	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The fraction of the stock gap ordered per week.	0.26
beta	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The weight of the supply line in the stock gap.	0.34
dispatch_time	Annotated[float, ParameterSpec(domain=open(0, oo))]	The time over which the echelon ships its backlog.	1.0
mail_delay	Annotated[float, ParameterSpec(domain=open(0, oo))]	The average time an order of the echelon takes to reach its supplier.	2.0
shipping_delay	Annotated[float, ParameterSpec(domain=open(0, oo))]	The average time a shipment takes to reach the echelon from its supplier.	2.0
production_delay	Annotated[float, ParameterSpec(domain=open(0, oo))]	The average time the brewery takes to brew an order, through a third-order delay.	3.0

Examples:

>>> from skcomplex.systems.economics.markets import FactoryElement
>>> FactoryElement('factory').production_delay
3.0

Source code in src/skcomplex/systems/economics/markets/_supply_chains.py

def __init__(
    self,
    label: str,
    visual: VisualSpec | None = None,
    *,
    inventory: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo), quantity='beer')] = 12.0,
    in_transit: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo), quantity='beer')] = 8.0,
    backlog: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 4.0,
    expected: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 4.0,
    supply_line: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 16.0,
    desired: Annotated[float, ParameterSpec()] = 13.44,
    theta: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.36,
    alpha: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.26,
    beta: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.34,
    dispatch_time: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 1.0,
    mail_delay: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 2.0,
    shipping_delay: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 2.0,
    production_delay: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 3.0,
) -> None:
    super().__init__(
        label,
        visual=visual,
        inventory=inventory,
        in_transit=in_transit,
        backlog=backlog,
        expected=expected,
        supply_line=supply_line,
        desired=desired,
        theta=theta,
        alpha=alpha,
        beta=beta,
        dispatch_time=dispatch_time,
        mail_delay=mail_delay,
        shipping_delay=shipping_delay,
    )
    self.production_delay = production_delay

FairFoodMarketInteraction(label, *, rule=None, phase=None, shops, farms, pressure=0.001)

Bases: FeedbackInteraction[SystemView]

Tomato market and the Fair Food Program, which buyers and farms join by the four rules of Chesney.

Every period, in the order of the code of Chesney (2021, Section 7.6), every buyer outside the programme joins it at level 1 with the probability pressure (Rule 1). Every buyer at level 1, in random order, moves to level 2 when the member farms grow at least what the buyers at level 2 need plus its own need (Rule 2). The buyers then fill their needs in random order from the farms in random order, any farm at level 0, member farms first at level 1, and only member farms at level 2 (Rule 3). Finally every farm outside the programme joins it when it grows more than the buyers outside the programme need (Rule 4). Levels only rise.

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
shops	Annotated[str, RoleSpec()]	The path of the group of buyers.	required
farms	Annotated[str, RoleSpec()]	The path of the group of farms.	required
pressure	Annotated[float, ParameterSpec(domain=closed(0, 1))]	The probability per period that a buyer outside the programme joins it at level 1.	0.001

Examples:

>>> from skcomplex.systems.economics.markets import FairFoodMarketInteraction
>>> FairFoodMarketInteraction('market', shops='shops', farms='farms').reports[:2]
('committed_shops', 'member_shops')

Source code in src/skcomplex/systems/economics/markets/_fair_food_program.py

def __init__(
    self,
    label: str,
    *,
    rule: None = None,
    phase: str | None = None,
    shops: Annotated[str, RoleSpec()],
    farms: Annotated[str, RoleSpec()],
    pressure: Annotated[float, ParameterSpec(domain=RealSet.closed(0, 1))] = 0.001,
) -> None:
    super().__init__(label, rule=rule, phase=phase)
    self.shops = shops
    self.farms = farms
    self.pressure = pressure

act(system)

Run the four rules of one period.

Parameters:

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

Source code in src/skcomplex/systems/economics/markets/_fair_food_program.py

def act(self, system: SystemView) -> None:
    """Run the four rules of one period.

    Args:
        system:
            The system view of the market.
    """
    rng = system.rng
    shops = sorted(system.agents('shops', ProgramMemberView), key=lambda shop: shop.id)
    farms = sorted(system.agents('farms', ProgramMemberView), key=lambda farm: farm.id)
    need = np.array([shop.acres for shop in shops])
    shop_level = np.array([shop.level for shop in shops])
    acres = np.array([farm.acres for farm in farms])
    farm_level = np.array([farm.level for farm in farms])
    joining = (shop_level == OUTSIDE) & (rng.random(len(shops)) < system.parameter('pressure'))
    shop_level = np.where(joining, float(COMMITTED), shop_level)
    for shop in rng.permutation(len(shops)):
        member_acres = acres[farm_level == 1].sum()
        if shop_level[shop] == COMMITTED and member_acres >= need[shop_level == MEMBER].sum() + need[shop]:
            shop_level[shop] = float(MEMBER)
    left, bought = acres.copy(), np.zeros(len(shops))
    for shop in rng.permutation(len(shops)):
        order = rng.permutation(len(farms))
        members, others = order[farm_level[order] == 1], order[farm_level[order] == 0]
        if shop_level[shop] == OUTSIDE:
            bought[shop] = self._buy(need[shop], order, left)
        elif shop_level[shop] == COMMITTED:
            bought[shop] = self._buy(need[shop], members, left)
            bought[shop] += self._buy(need[shop] - bought[shop], others, left)
        else:
            bought[shop] = self._buy(need[shop], members, left)
    for farm in range(len(farms)):
        if farm_level[farm] == 0 and acres[farm] > need[shop_level == OUTSIDE].sum():
            farm_level[farm] = 1.0
    for view, level, amount in zip(shops, shop_level, bought, strict=True):
        view.level, view.traded = float(level), float(amount)
    for grower, level, rest, grown in zip(farms, farm_level, left, acres, strict=True):
        grower.level, grower.traded = float(level), float(grown - rest)
    system.report('committed_shops', float(np.sum(shop_level == COMMITTED)))
    system.report('member_shops', float(np.sum(shop_level == MEMBER)))
    system.report('member_farms', float(np.sum(farm_level == 1)))
    system.report('unfulfilled', float((need - bought).sum()))
    system.report('unsold', float(left.sum()))

ForagerElement(label, visual=VisualSpec('agents', {'color_by': 'sugar'}), *, sugar=10.0, vision=1.0, metabolism=1.0)

Bases: Element

Agent that moves to the richest free cell in sight, harvests its sugar, burns its metabolism, and may starve.

In its own activation, a forager sees vision cells along each of the four lattice directions and moves to the free cell with the most sugar, the nearest among equals, or stays when no cell in sight beats its own, by a MovementInteraction of the best step with a vision. Transfers then harvest the sugar of its cell and burn its metabolism, and a LifeCycleInteraction removes it when its wealth falls below zero.

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': 'sugar'})
sugar	Annotated[float, StateSpec()]	The wealth of the forager.	10.0
vision	Annotated[float, StateSpec(domain=NN)]	The number of cells the forager sees in every direction.	1.0
metabolism	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The sugar the forager burns in a step.	1.0

Examples:

>>> from skcomplex.core.components.elements import GroupElement
>>> from skcomplex.systems.economics.markets import ForagerElement
>>> GroupElement('foragers', element=ForagerElement, size=3, values={'vision': 2.0}).size
3

Source code in src/skcomplex/systems/economics/markets/_sugarscape.py

def __init__(
    self,
    label: str,
    visual: VisualSpec | None = VisualSpec('agents', {'color_by': 'sugar'}),
    *,
    sugar: Annotated[float, StateSpec()] = 10.0,
    vision: Annotated[float, StateSpec(domain=NN)] = 1.0,
    metabolism: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
) -> None:
    super().__init__(label, visual=visual)
    self.sugar = sugar
    self.vision = vision
    self.metabolism = metabolism

GoodsTraderElement(label, visual=VisualSpec('agents', {'color_by': 'sugar_weight', 'color_range': (0.0, 1.0)}), *, sugar=10.0, spice=10.0, salt=0.0, sugar_weight=0.5, spice_weight=0.5, salt_weight=0.0)

Bases: Element

Trader who holds goods and values them with Cobb-Douglas weights.

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': 'sugar_weight', 'color_range': (0.0, 1.0)})
sugar	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The holding of sugar.	10.0
spice	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The holding of spice.	10.0
salt	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The holding of salt.	0.0
sugar_weight	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The Cobb-Douglas weight of sugar.	0.5
spice_weight	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The Cobb-Douglas weight of spice.	0.5
salt_weight	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The Cobb-Douglas weight of salt, zero when salt is not traded.	0.0

Examples:

>>> from skcomplex.core.components.elements import GroupElement
>>> from skcomplex.systems.economics.markets import GoodsTraderElement
>>> GroupElement('traders', element=GoodsTraderElement, size=4, values={'sugar': 5.0}).size
4

Source code in src/skcomplex/systems/economics/markets/_goods_traders.py

def __init__(
    self,
    label: str,
    visual: VisualSpec | None = VisualSpec('agents', {'color_by': 'sugar_weight', 'color_range': (0.0, 1.0)}),
    *,
    sugar: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 10.0,
    spice: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 10.0,
    salt: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
    sugar_weight: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.5,
    spice_weight: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.5,
    salt_weight: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
) -> None:
    super().__init__(label, visual=visual)
    self.sugar = sugar
    self.spice = spice
    self.salt = salt
    self.sugar_weight = sugar_weight
    self.spice_weight = spice_weight
    self.salt_weight = salt_weight

HeuristicPricingInteraction(label, *, rule=None, phase=None, traders, rate=1.1, strength=1.2, cost=1.0, risk=1.0, noise=0.01, initial=0.1)

Bases: FeedbackInteraction[SystemView]

Market maker that clears the risky asset at the price where the mean forecast of the traders is met.

The deviation x of the price from its fundamental value clears at R x_t = sum_h n_h f_h + eps_t, where n_h is the share of traders using heuristic h. The fundamentalists forecast the fundamental, f = 0, and the trend chasers extrapolate the last deviation, f = g x_{t-1}, both as heuristics.fundamental with a reversion of 0 and g. The realised fitness of a heuristic is the profit of its last forecast, (x_t - R x_{t-1}) (f_h(t - 1) - R x_{t-1}) / (a sigma^2), minus its cost.

Parameters:

Name	Type	Description	Default
label	str	The label of the institution.	required
rule	Callable[[SystemView], None] | None	Unused, since the institution overrides act.	None
phase	str | None	The phase the institution clears in.	None
traders	Annotated[str, RoleSpec()]	The path of the group of traders.	required
rate	Annotated[float, ParameterSpec(domain=open(0, oo))]	The gross return R of the riskless asset.	1.1
strength	Annotated[float, ParameterSpec()]	The extrapolation g of the trend chasers.	1.2
cost	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The cost of the fundamentalist heuristic.	1.0
risk	Annotated[float, ParameterSpec(domain=open(0, oo))]	The risk aversion times the variance of returns, a sigma^2.	1.0
noise	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The standard deviation of the noise of the price.	0.01
initial	Annotated[float, ParameterSpec()]	The initial deviation of the price from its fundamental value.	0.1

Examples:

>>> from skcomplex.systems.economics.markets import HeuristicPricingInteraction
>>> HeuristicPricingInteraction('pricing', traders='traders').reports
('deviation', 'trend_share', 'fitness_fundamentalist', 'fitness_trend')

Source code in src/skcomplex/systems/economics/markets/_brock_hommes.py

def __init__(
    self,
    label: str,
    *,
    rule: Callable[[SystemView], None] | None = None,
    phase: str | None = None,
    traders: Annotated[str, RoleSpec()],
    rate: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 1.1,
    strength: Annotated[float, ParameterSpec()] = 1.2,
    cost: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
    risk: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 1.0,
    noise: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.01,
    initial: Annotated[float, ParameterSpec()] = 0.1,
) -> None:
    super().__init__(label, rule=rule, phase=phase)
    self.traders = traders
    self.rate = rate
    self.strength = strength
    self.cost = cost
    self.risk = risk
    self.noise = noise
    self.initial = initial

act(system)

Clear the price from the forecasts of the traders' heuristics and report the fitnesses.

Parameters:

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

Source code in src/skcomplex/systems/economics/markets/_brock_hommes.py

def act(self, system: SystemView) -> None:
    """Clear the price from the forecasts of the traders' heuristics and report the fitnesses.

    Args:
        system:
            The system view of the institution.
    """
    rate, strength = system.parameter('rate'), system.parameter('strength')
    state = system.storage.get('market')
    if not isinstance(state, _MarketStateSpec):
        state = _MarketStateSpec((system.parameter('initial'),))
    history = list(state.history)
    forecasts = np.array(
        [heuristics.fundamental(history), heuristics.fundamental(history, fundamental=0.0, reversion=strength)],
    )
    traders = system.agents('traders', HeuristicTraderView)
    share = float(np.mean([trader.predictor for trader in traders])) if traders else 0.0
    price = float(forecasts @ [1 - share, share] + system.rng.normal(0.0, system.parameter('noise'))) / rate
    previous = state.forecasts
    excess = price - rate * history[-1]
    if previous is None:
        fitness = np.zeros(len(PREDICTORS))
    else:
        fitness = excess * (previous - rate * history[-1]) / system.parameter('risk')
    fitness -= [system.parameter('cost'), 0.0]
    system.storage['market'] = _MarketStateSpec((history[-1], price), forecasts)
    system.report('deviation', price)
    system.report('trend_share', share)
    for name, value in zip(PREDICTORS, fitness, strict=True):
        system.report(f'fitness_{name}', value)

HeuristicTraderElement(label, visual=None, *, predictor=0.0, fundamentalist_fitness=0.0, trend_fitness=0.0)

Bases: Element

Trader who holds the index of the forecasting heuristic in use, 0 for the fundamentalist and 1 for the trend.

The trader scores every heuristic by its last realised fitness, fundamentalist_fitness and trend_fitness, which a reinforcement of rate 1 sets every step and a logit choice reads.

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
predictor	Annotated[float, StateSpec(domain=NN)]	The index of the heuristic in use.	0.0
fundamentalist_fitness	Annotated[float, StateSpec()]	The last realised fitness of the fundamentalist heuristic.	0.0
trend_fitness	Annotated[float, StateSpec()]	The last realised fitness of the trend heuristic.	0.0

Examples:

>>> from skcomplex.core.components.elements import GroupElement
>>> from skcomplex.systems.economics.markets import HeuristicTraderElement
>>> GroupElement('traders', element=HeuristicTraderElement, size=10, values={'predictor': 1.0}).size
10

Source code in src/skcomplex/systems/economics/markets/_brock_hommes.py

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

get_fitness(agent, system, option) staticmethod

Return the score of a heuristic of a trader, its last realised fitness.

Parameters:

Name	Type	Description	Default
agent	HeuristicTraderView	The trader.	required
system	SystemView	The system view of the choice.	required
option	float	The index of the heuristic.	required

Returns:

Name	Type	Description
fitness	float	The last realised fitness of the heuristic.

Source code in src/skcomplex/systems/economics/markets/_brock_hommes.py

@staticmethod
def get_fitness(agent: HeuristicTraderView, system: SystemView, option: float) -> float:
    """Return the score of a heuristic of a trader, its last realised fitness.

    Args:
        agent:
            The trader.

        system:
            The system view of the choice.

        option:
            The index of the heuristic.

    Returns:
        fitness:
            The last realised fitness of the heuristic.
    """
    return float(getattr(agent, f'{PREDICTORS[int(option)]}_fitness'))

get_realised_fitness(agent, system, index) staticmethod

Return the fitness of a heuristic that the market maker reported last, zero before the first report.

Parameters:

Name	Type	Description	Default
agent	HeuristicTraderView	The trader.	required
system	SystemView	The system view of the reinforcement.	required
index	int	The index of the heuristic.	required

Returns:

Name	Type	Description
fitness	float	The last reported fitness of the heuristic.

Source code in src/skcomplex/systems/economics/markets/_brock_hommes.py

@staticmethod
def get_realised_fitness(agent: HeuristicTraderView, system: SystemView, index: int) -> float:
    """Return the fitness of a heuristic that the market maker reported last, zero before the first report.

    Args:
        agent:
            The trader.

        system:
            The system view of the reinforcement.

        index:
            The index of the heuristic.

    Returns:
        fitness:
            The last reported fitness of the heuristic.
    """
    reported = system.history(f'{PRICING}.fitness_{PREDICTORS[index]}', 1)
    return float(reported[-1]) if len(reported) else 0.0

JobseekerElement(label, visual=None, *, ask=1.0, supply=1.0, employer=-1.0, wage=0.0, duration=0.0, hired=-1.0, offer=-1.0, separation=0.05, reservation=1.0, decline=0.1)

Bases: Element

Worker who loses its job at the separation rate and, while unemployed, lowers its reservation wage.

A worker who loses its job searches from the next step. An unemployed worker asks the reservation wage w_0 (1 - decline)^d, where d is the number of steps of search, and is hired by the matching market.

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
ask	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The reservation wage.	1.0
supply	Annotated[float, StateSpec(domain=NN)]	One while the worker searches in the current step, zero otherwise.	1.0
employer	Annotated[float, StateSpec(reference=firms)]	The id of the employer, or -1 for an unemployed worker.	-1.0
wage	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The wage of the current job.	0.0
duration	Annotated[float, StateSpec(domain=NN)]	The number of steps of the current spell of search.	0.0
hired	Annotated[float, StateSpec()]	The step of the last hire.	-1.0
offer	Annotated[float, StateSpec()]	The step in which the offer of the last hire was posted.	-1.0
separation	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The probability that a job is destroyed in a step.	0.05
reservation	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The reservation wage at the start of a search.	1.0
decline	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The relative fall of the reservation wage per step of search.	0.1

Examples:

>>> from skcomplex.core.components.elements import GroupElement
>>> from skcomplex.systems.economics.markets import JobseekerElement
>>> GroupElement('workers', element=JobseekerElement, size=5).size
5

Source code in src/skcomplex/systems/economics/markets/_labour.py

def __init__(
    self,
    label: str,
    visual: VisualSpec | None = None,
    *,
    ask: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
    supply: Annotated[float, StateSpec(domain=NN)] = 1.0,
    employer: Annotated[float, StateSpec(reference='firms')] = -1.0,
    wage: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
    duration: Annotated[float, StateSpec(domain=NN)] = 0.0,
    hired: Annotated[float, StateSpec()] = -1.0,
    offer: Annotated[float, StateSpec()] = -1.0,
    separation: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.05,
    reservation: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
    decline: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.1,
) -> None:
    super().__init__(label, visual=visual)
    self.ask = ask
    self.supply = supply
    self.employer = employer
    self.wage = wage
    self.duration = duration
    self.hired = hired
    self.offer = offer
    self.separation = separation
    self.reservation = reservation
    self.decline = decline

get_search() staticmethod

Return the states a worker sets in the phase of offers before the separations, in their order.

A worker unemployed at the start of the step searches, with a reservation wage that falls with the duration of its search.

Returns:

Name	Type	Description
search	tuple[tuple[str, Callable[[JobseekerView, SystemView], float]], ...]	The name of every state and the function of its value, of the worker and the system view.

Source code in src/skcomplex/systems/economics/markets/_labour.py

@staticmethod
def get_search() -> tuple[tuple[str, Callable[[JobseekerView, SystemView], float]], ...]:
    """Return the states a worker sets in the phase of offers before the separations, in their order.

    A worker unemployed at the start of the step searches, with a reservation wage that falls with the duration of
    its search.

    Returns:
        search:
            The name of every state and the function of its value, of the worker and the system view.
    """
    return (
        ('supply', lambda agent, system: float(agent.employer < 0)),
        (
            'ask',
            lambda agent, system: (
                agent.reservation * (1 - agent.decline) ** agent.duration if agent.supply else agent.ask
            ),
        ),
    )

get_spell() staticmethod

Return the states a worker sets in the phase of offers after the separations, in their order.

A worker who lost its job earns no wage, and the duration of a search grows while the worker searches.

Returns:

Name	Type	Description
spell	tuple[tuple[str, Callable[[JobseekerView, SystemView], float]], ...]	The name of every state and the function of its value, of the worker and the system view.

Source code in src/skcomplex/systems/economics/markets/_labour.py

@staticmethod
def get_spell() -> tuple[tuple[str, Callable[[JobseekerView, SystemView], float]], ...]:
    """Return the states a worker sets in the phase of offers after the separations, in their order.

    A worker who lost its job earns no wage, and the duration of a search grows while the worker searches.

    Returns:
        spell:
            The name of every state and the function of its value, of the worker and the system view.
    """
    return (
        ('wage', lambda agent, system: 0.0 if agent.employer < 0 else agent.wage),
        (
            'duration',
            lambda agent, system: (
                agent.duration + 1 if agent.supply else 0.0 if agent.employer < 0 else agent.duration
            ),
        ),
    )

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

Bases: Element

Market of one good described by its price.

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
price	Annotated[float, StateSpec()]	The initial price.	1.0

Examples:

>>> from skcomplex.systems.economics.markets import MarketElement
>>> MarketElement('wheat', price=2.0).price
2.0

Source code in src/skcomplex/systems/economics/markets/_cobweb.py

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

PricePointElement(label, *, distribution=0.5, price=1.0, alpha=0.25, beta=0.5, n_firms=2.0)

Bases: Element

Equilibrium distribution of prices at one price, which makes a firm indifferent among its prices.

A firm that charges P sells to its loyal share alpha, and to the switchers beta when it undercuts all its m - 1 rivals, each with the probability 1 - F(P), for the expected profit P (alpha + beta (1 - F)^(m - 1)). In equilibrium every price gives the profit alpha of selling to the loyal share at the highest price 1, which F(P) = 1 - (alpha (1 - P) / (beta P))^(1 / (m - 1)) solves (Forrest 2018 Lemma 20.1).

Parameters:

Name	Type	Description	Default
label	str	The label of the point or the group.	required
distribution	Annotated[float, StateSpec()]	The initial guess of F(P), which the equations replace.	0.5
price	Annotated[float, StateSpec(domain=open_closed(0, 1))]	The price P.	1.0
alpha	Annotated[float, ParameterSpec(domain=open(0, oo))]	The loyal share of every firm.	0.25
beta	Annotated[float, ParameterSpec(domain=open(0, oo))]	The share of switchers.	0.5
n_firms	Annotated[float, ParameterSpec(domain=closed_open(2, oo))]	The number of firms m.	2.0

Examples:

>>> from skcomplex.systems.economics.markets import PricePointElement
>>> PricePointElement('point', price=0.5).price
0.5

Source code in src/skcomplex/systems/economics/markets/_oligopoly.py

def __init__(
    self,
    label: str,
    *,
    distribution: Annotated[float, StateSpec()] = 0.5,
    price: Annotated[float, StateSpec(domain=RealSet.open_closed(0, 1))] = 1.0,
    alpha: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 0.25,
    beta: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 0.5,
    n_firms: Annotated[float, ParameterSpec(domain=RealSet.closed_open(2, oo))] = 2.0,
) -> None:
    super().__init__(label)
    self.distribution = distribution
    self.price = price
    self.alpha = alpha
    self.beta = beta
    self.n_firms = n_firms

get_profit(member) staticmethod

Return the expected profit of a firm at the price of a member, for the observables of a run.

Parameters:

Name	Type	Description	Default
member	SimpleNamespace	The namespace of the states and parameters of the member.	required

Returns:

Name	Type	Description
profit	Expression	The expected profit P (alpha + beta (1 - F)^(m - 1)).

Source code in src/skcomplex/systems/economics/markets/_oligopoly.py

@staticmethod
def get_profit(member: SimpleNamespace) -> Expression:
    """Return the expected profit of a firm at the price of a member, for the observables of a run.

    Args:
        member:
            The namespace of the states and parameters of the member.

    Returns:
        profit:
            The expected profit `P (alpha + beta (1 - F)^(m - 1))`.
    """
    return member.price * (member.alpha + member.beta * (1 - member.distribution) ** (member.n_firms - 1))

PricingFirmElement(label, price=1.0, scores=0.0)

Bases: Element

Firm that charges one price of a grid and counts what every price would have earned.

Parameters:

Name	Type	Description	Default
label	str	The label of the firm or the group.	required
price	Annotated[float, StateSpec()]	The price the firm charges.	1.0
scores	Annotated[ArrayLike, StateSpec(vector=True)]	The sum of the payoffs of every price of the grid against the prices of the rival so far.	0.0

Examples:

>>> from skcomplex.core.components.elements import GroupElement
>>> from skcomplex.systems.economics.markets import PricingFirmElement
>>> GroupElement('firms', element=PricingFirmElement, size=2).size
2

Source code in src/skcomplex/systems/economics/markets/_oligopoly.py

def __init__(
    self,
    label: str,
    price: Annotated[float, StateSpec()] = 1.0,
    scores: Annotated[npt.ArrayLike, StateSpec(vector=True)] = 0.0,
) -> None:
    super().__init__(label)
    self.price = price
    self.scores = scores

PricingGameSpec(prices, alpha, beta) dataclass

Bases: BaseSpec

Grid of prices of two firms, each with a loyal share, that share the switchers who pay the lower price.

Parameters:

Name	Type	Description	Default
prices	Array	The prices the firms choose among, increasing.	required
alpha	float	The loyal share of every firm.	required
beta	float	The share of switchers.	required

Examples:

>>> import numpy as np
>>> from skcomplex.systems.economics.markets import PricingGameSpec
>>> PricingGameSpec(np.array([0.5, 1.0]), 0.25, 0.5).beta
0.5

payoff(agent, system, k)

Return what the price of index k earns against the price the rival charged in this step.

Parameters:

Name	Type	Description	Default
agent	PricingFirmView	The firm.	required
system	SystemView	The system view, which holds the rival.	required
k	int	The index of the price.	required

Returns:

Name	Type	Description
payoff	float	The profit of the price.

Source code in src/skcomplex/systems/economics/markets/_oligopoly.py

def payoff(self, agent: PricingFirmView, system: SystemView, k: int) -> float:
    """Return what the price of index `k` earns against the price the rival charged in this step.

    Args:
        agent:
            The firm.

        system:
            The system view, which holds the rival.

        k:
            The index of the price.

    Returns:
        payoff:
            The profit of the price.
    """
    rival = cast(PricingFirmView, next(item for item in system.agents(agent.group) if item.id != agent.id))
    price, against = float(self.prices[k]), float(rival.price)
    share = 1.0 if price < against else 0.5 if price == against else 0.0
    return price * (self.alpha + self.beta * share)

utility(agent, system, option)

Return what a price has earned so far against the prices of the rival, its score.

Parameters:

Name	Type	Description	Default
agent	PricingFirmView	The firm.	required
system	SystemView	The system view, which the score does not read.	required
option	float	The price.	required

Returns:

Name	Type	Description
utility	float	The score of the price.

Source code in src/skcomplex/systems/economics/markets/_oligopoly.py

def utility(self, agent: PricingFirmView, system: SystemView, option: float) -> float:
    """Return what a price has earned so far against the prices of the rival, its score.

    Args:
        agent:
            The firm.

        system:
            The system view, which the score does not read.

        option:
            The price.

    Returns:
        utility:
            The score of the price.
    """
    return float(agent.scores[int(np.searchsorted(self.prices, option))])

ProductElement(label, visual=None, *, shelved=0.0, tau_a=1.0)

Bases: Element

Product on the shelf of a shop, which ages from the time it was shelved, its freshness exp(-age / tau_a).

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
shelved	Annotated[float, StateSpec()]	The time at which the product was put on the shelf.	0.0
tau_a	Annotated[float, ParameterSpec(domain=open(0, oo))]	The time over which the product ages.	1.0

Examples:

>>> from skcomplex.core.components.elements import GroupElement
>>> from skcomplex.systems.economics.markets import ProductElement
>>> GroupElement('products', element=ProductElement, size=4, location=[0, 0, 1, 1]).size
4

Source code in src/skcomplex/systems/economics/markets/_perishables.py

def __init__(
    self,
    label: str,
    visual: VisualSpec | None = None,
    *,
    shelved: Annotated[float, StateSpec()] = 0.0,
    tau_a: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 1.0,
) -> None:
    super().__init__(label, visual=visual)
    self.shelved = shelved
    self.tau_a = tau_a

restock(product, customer, system) staticmethod

Return the amount shelved - t that resets the shelving time of a sold product to the time of the sale.

Parameters:

Name	Type	Description	Default
product	ProductView	The product sold.	required
customer	CustomerView	The customer who buys it.	required
system	SystemView	The system view of the purchase.	required

Returns:

Name	Type	Description
amount	float	The amount taken from the shelving time.

Source code in src/skcomplex/systems/economics/markets/_perishables.py

@staticmethod
def restock(product: ProductView, customer: CustomerView, system: SystemView) -> float:
    """Return the amount `shelved - t` that resets the shelving time of a sold product to the time of the sale.

    Args:
        product:
            The product sold.

        customer:
            The customer who buys it.

        system:
            The system view of the purchase.

    Returns:
        amount:
            The amount taken from the shelving time.
    """
    return product.shelved - system.time

sell(product, customer, system) staticmethod

Return the amount that sets the freshness the customer holds to exp(-(t - shelved) / tau_a).

Parameters:

Name	Type	Description	Default
product	ProductView	The product sold.	required
customer	CustomerView	The customer who buys it.	required
system	SystemView	The system view of the purchase.	required

Returns:

Name	Type	Description
amount	float	The change of the freshness the customer holds.

Source code in src/skcomplex/systems/economics/markets/_perishables.py

@staticmethod
def sell(product: ProductView, customer: CustomerView, system: SystemView) -> float:
    """Return the amount that sets the freshness the customer holds to `exp(-(t - shelved) / tau_a)`.

    Args:
        product:
            The product sold.

        customer:
            The customer who buys it.

        system:
            The system view of the purchase.

    Returns:
        amount:
            The change of the freshness the customer holds.
    """
    return float(np.exp(-(system.time - product.shelved) / product.tau_a)) - customer.freshness

ProgramMemberElement(label, visual=VisualSpec('agents', {'color_by': 'level'}), *, acres=0.0, level=0, traded=0.0)

Bases: Element

Tomato buyer or farm with its acres of tomatoes, its level in the Fair Food Program, and what it traded.

A buyer needs acres of tomatoes every period and has the level 0 outside the programme, 1 when it buys from member farms first, and 2 when it buys only from member farms. A farm grows acres of tomatoes and has the level 0 outside the programme and 1 inside. traded is what a buyer bought or a farm sold in the period.

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': 'level'})
acres	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The tomatoes a buyer needs or a farm grows every period, in acres.	0.0
level	Annotated[float, StateSpec(domain=NN)]	The level of involvement in the programme.	0
traded	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The tomatoes bought or sold in the period.	0.0

Examples:

>>> from skcomplex.core.components.elements import GroupElement
>>> from skcomplex.systems.economics.markets import ProgramMemberElement
>>> GroupElement('farms', element=ProgramMemberElement, size=3, values={'acres': 10.0}).size
3

Source code in src/skcomplex/systems/economics/markets/_fair_food_program.py

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

QuadraticSupplyMarketElement(label, visual=None, *, P=1.0, k=1.0, beta=10.0, lam=1.0, delta=1.0, alpha=2.0, gamma=-0.1)

Bases: Element

Market of one good whose price adjusts to the excess of a linear demand over a quadratic supply.

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
P	Annotated[float, StateSpec()]	The initial price.	1.0
k	Annotated[float, ParameterSpec(domain=open(0, oo))]	The speed of the price adjustment.	1.0
beta	Annotated[float, ParameterSpec(domain=open(0, oo))]	The demand at price zero.	10.0
lam	Annotated[float, ParameterSpec(domain=open(0, oo))]	The slope of the demand.	1.0
delta	Annotated[float, ParameterSpec(domain=open(0, oo))]	The supply at price zero.	1.0
alpha	Annotated[float, ParameterSpec(domain=open(0, oo))]	The linear intensity of the supply.	2.0
gamma	Annotated[float, ParameterSpec()]	The quadratic intensity of the supply, negative for saturation.	-0.1

Examples:

>>> from skcomplex.systems.economics.markets import QuadraticSupplyMarketElement
>>> QuadraticSupplyMarketElement('market', P=2.0).P
2.0

Source code in src/skcomplex/systems/economics/markets/_prices.py

def __init__(
    self,
    label: str,
    visual: VisualSpec | None = None,
    *,
    P: Annotated[float, StateSpec()] = 1.0,
    k: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 1.0,
    beta: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 10.0,
    lam: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 1.0,
    delta: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 1.0,
    alpha: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 2.0,
    gamma: Annotated[float, ParameterSpec()] = -0.1,
) -> None:
    super().__init__(label, visual=visual)
    self.P = P
    self.k = k
    self.beta = beta
    self.lam = lam
    self.delta = delta
    self.alpha = alpha
    self.gamma = gamma

ResidentElement(label, visual=VisualSpec('agents', {'color_by': 'travel', 'color_range': (0.0, 1.0)}), *, bid=20.0, demand=1.0, money=1000.0, travel=1.0, distance_cost=1.0, online=0.0)

Bases: Element

Shopper who wants one unit a step and shops in person, paying for travel, or online, where distance is free.

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': 'travel', 'color_range': (0.0, 1.0)})
bid	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The highest price the shopper pays.	20.0
demand	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The units still wanted in the step.	1.0
money	Annotated[float, StateSpec()]	The money of the shopper.	1000.0
travel	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The travel cost per cell of the current step, zero online.	1.0
distance_cost	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The travel cost per cell of shopping in person.	1.0
online	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The probability of shopping online in a step.	0.0

Examples:

>>> from skcomplex.core.components.elements import GroupElement
>>> from skcomplex.systems.economics.markets import ResidentElement
>>> GroupElement('shoppers', element=ResidentElement, size=3, values={'online': 1.0}).size
3

Source code in src/skcomplex/systems/economics/markets/_shops.py

def __init__(
    self,
    label: str,
    visual: VisualSpec | None = VisualSpec('agents', {'color_by': 'travel', 'color_range': (0.0, 1.0)}),
    *,
    bid: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 20.0,
    demand: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
    money: Annotated[float, StateSpec()] = 1000.0,
    travel: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
    distance_cost: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
    online: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
) -> None:
    super().__init__(label, visual=visual)
    self.bid = bid
    self.demand = demand
    self.money = money
    self.travel = travel
    self.distance_cost = distance_cost
    self.online = online

get_preference(shopper, shop) staticmethod

Return the preference of a shopper for a shop, the opposite of the delivered price.

The delivered price is the shop's ask plus the shopper's travel cost per cell times the distance to the shop.

Parameters:

Name	Type	Description	Default
shopper	object	The shopper.	required
shop	object	The shop.	required

Returns:

Name	Type	Description
preference	float	The opposite of the delivered price, higher for a preferred shop.

Source code in src/skcomplex/systems/economics/markets/_shops.py

@staticmethod
def get_preference(shopper: object, shop: object) -> float:
    """Return the preference of a shopper for a shop, the opposite of the delivered price.

    The delivered price is the shop's ask plus the shopper's travel cost per cell times the distance to the shop.

    Args:
        shopper:
            The shopper.

        shop:
            The shop.

    Returns:
        preference:
            The opposite of the delivered price, higher for a preferred shop.
    """
    if not isinstance(shopper, ResidentView) or not isinstance(shop, StoreView):
        return -math.inf
    return -(shop.ask + shopper.travel * shopper.distance_to(shop))

get_travel(agent, system) staticmethod

Return the travel cost per cell of the step, zero when the shopper shops online with its probability.

Parameters:

Name	Type	Description	Default
agent	ResidentView	The shopper.	required
system	SystemView	The system view of the plan.	required

Returns:

Name	Type	Description
travel	float	The travel cost per cell.

Source code in src/skcomplex/systems/economics/markets/_shops.py

@staticmethod
def get_travel(agent: ResidentView, system: SystemView) -> float:
    """Return the travel cost per cell of the step, zero when the shopper shops online with its probability.

    Args:
        agent:
            The shopper.

        system:
            The system view of the plan.

    Returns:
        travel:
            The travel cost per cell.
    """
    return 0.0 if system.rng.random() < agent.online else agent.distance_cost

RetailDemandInteraction(label, *, retailer)

Bases: TransferInteraction

Customers of a supply chain, who take the shipments of the retailer.

The sales are a transfer from the inventory of the retailer to the customers.

Parameters:

Name	Type	Description	Default
label	str	The label of the customers.	required
retailer	Annotated[str, RoleSpec()]	The path of the retailer.	required

Examples:

>>> from skcomplex.systems.economics.markets import RetailDemandInteraction
>>> RetailDemandInteraction('customers', retailer='retailer').retailer
'retailer'

Source code in src/skcomplex/systems/economics/markets/_supply_chains.py

def __init__(
    self,
    label: str,
    *,
    retailer: Annotated[str, RoleSpec()],
) -> None:
    super().__init__(
        label,
        source=retailer,
        target=retailer,
        state='inventory',
        into=Sink('customers'),
        amount=lambda s: EchelonElement._shipments(s.retailer),
        flow='sales',
    )
    self.retailer = retailer

RetailerElement(label, visual=None, *, inventory=12.0, in_transit=8.0, backlog=4.0, expected=4.0, supply_line=16.0, desired=13.44, theta=0.36, alpha=0.26, beta=0.34, dispatch_time=1.0, mail_delay=2.0, shipping_delay=2.0, demand=4.0, demand_step=4.0, step_time=5.0)

Bases: EchelonElement

Retailer of a supply chain, an echelon whose customers order their demand, which steps once.

Parameters:

Name	Type	Description	Default
label	str	The label of the retailer.	required
visual	VisualSpec | None	How the element is drawn, or None for default_visual_.	None
inventory	Annotated[float, StateSpec(domain=closed_open(0, oo), quantity=beer)]	The initial inventory, in cases.	12.0
in_transit	Annotated[float, StateSpec(domain=closed_open(0, oo), quantity=beer)]	The initial beer on its way to the echelon.	8.0
backlog	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The initial orders waiting to ship.	4.0
expected	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The initial expected demand.	4.0
supply_line	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The initial beer ordered and not received.	16.0
desired	Annotated[float, ParameterSpec()]	The desired stock S'.	13.44
theta	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The rate at which the expected demand adapts to incoming orders.	0.36
alpha	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The fraction of the stock gap ordered per week.	0.26
beta	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The weight of the supply line in the stock gap.	0.34
dispatch_time	Annotated[float, ParameterSpec(domain=open(0, oo))]	The time over which the echelon ships its backlog.	1.0
mail_delay	Annotated[float, ParameterSpec(domain=open(0, oo))]	The average time an order of the echelon takes to reach its supplier.	2.0
shipping_delay	Annotated[float, ParameterSpec(domain=open(0, oo))]	The average time a shipment takes to reach the echelon from its supplier.	2.0
demand	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The customer demand, in cases a week.	4.0
demand_step	Annotated[float, ParameterSpec()]	The permanent change of the demand.	4.0
step_time	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The week of the change of the demand.	5.0

Examples:

>>> from skcomplex.systems.economics.markets import RetailerElement
>>> RetailerElement('retailer').demand
4.0

Source code in src/skcomplex/systems/economics/markets/_supply_chains.py

def __init__(
    self,
    label: str,
    visual: VisualSpec | None = None,
    *,
    inventory: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo), quantity='beer')] = 12.0,
    in_transit: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo), quantity='beer')] = 8.0,
    backlog: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 4.0,
    expected: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 4.0,
    supply_line: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 16.0,
    desired: Annotated[float, ParameterSpec()] = 13.44,
    theta: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.36,
    alpha: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.26,
    beta: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.34,
    dispatch_time: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 1.0,
    mail_delay: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 2.0,
    shipping_delay: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 2.0,
    demand: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 4.0,
    demand_step: Annotated[float, ParameterSpec()] = 4.0,
    step_time: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 5.0,
) -> None:
    super().__init__(
        label,
        visual=visual,
        inventory=inventory,
        in_transit=in_transit,
        backlog=backlog,
        expected=expected,
        supply_line=supply_line,
        desired=desired,
        theta=theta,
        alpha=alpha,
        beta=beta,
        dispatch_time=dispatch_time,
        mail_delay=mail_delay,
        shipping_delay=shipping_delay,
    )
    self.demand = demand
    self.demand_step = demand_step
    self.step_time = step_time

ShopperElement(label, visual=None, *, wanted=0, visited=0, missed=0, spent=0.0)

Bases: Element

Shopper who visits one stall a step that it has not visited yet and buys what it wants there.

A shopper gives up the items that it has not found once it has visited every stall.

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
wanted	Annotated[float, StateSpec(domain=NN)]	The bit mask of the items on the shopping list.	0
visited	Annotated[float, StateSpec(domain=NN)]	The bit mask of the stalls already visited.	0
missed	Annotated[float, StateSpec(domain=NN)]	The bit mask of the items given up.	0
spent	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The money spent.	0.0

Examples:

>>> from skcomplex.core.components.elements import GroupElement
>>> from skcomplex.systems.economics.markets import ShopperElement
>>> GroupElement('shoppers', element=ShopperElement, size=5, values={'wanted': 3}).size
5

Source code in src/skcomplex/systems/economics/markets/_produce.py

def __init__(
    self,
    label: str,
    visual: VisualSpec | None = None,
    *,
    wanted: Annotated[float, StateSpec(domain=NN)] = 0,
    visited: Annotated[float, StateSpec(domain=NN)] = 0,
    missed: Annotated[float, StateSpec(domain=NN)] = 0,
    spent: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
) -> None:
    super().__init__(label, visual=visual)
    self.wanted = wanted
    self.visited = visited
    self.missed = missed
    self.spent = spent

shop(agent, system)

Visit a random stall not visited yet and buy every wanted item that it sells.

Parameters:

Name	Type	Description	Default
agent	ShopperView	The shopper.	required
system	SystemView	The system.	required

Source code in src/skcomplex/systems/economics/markets/_produce.py

def shop(self, agent: ShopperView, system: SystemView) -> None:
    """Visit a random stall not visited yet and buy every wanted item that it sells.

    Args:
        agent:
            The shopper.

        system:
            The system.
    """
    wanted, visited = int(agent.wanted), int(agent.visited)
    if not wanted:
        return
    unvisited = [stall for stall in system.agents(self.stalls, StallView) if not visited >> int(stall.stall) & 1]
    if not unvisited:
        agent.missed, agent.wanted = wanted, 0
        return
    stall = unvisited[int(system.rng.integers(len(unvisited)))]
    bought = wanted & int(stall.stock)
    agent.visited = visited | 1 << int(stall.stall)
    agent.spent += (1 + stall.markup) * sum(WHOLESALE_PRICES[item] for item in ShopperElement._items_of(bought))
    agent.wanted = wanted & ~bought

StallElement(label, visual=None, *, stall=0, stock=0, markup=0.1)

Bases: Element

Market stall that sells some kinds of produce at the wholesale price plus its markup.

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
stall	Annotated[float, StateSpec(domain=NN)]	The index of the stall.	0
stock	Annotated[float, StateSpec(domain=NN)]	The bit mask of the kinds of produce that the stall sells.	0
markup	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The relative markup on the wholesale prices.	0.1

Examples:

>>> from skcomplex.core.components.elements import GroupElement
>>> from skcomplex.systems.economics.markets import StallElement
>>> GroupElement('stalls', element=StallElement, size=3, values={'stall': [0, 1, 2]}).size
3

Source code in src/skcomplex/systems/economics/markets/_produce.py

def __init__(
    self,
    label: str,
    visual: VisualSpec | None = None,
    *,
    stall: Annotated[float, StateSpec(domain=NN)] = 0,
    stock: Annotated[float, StateSpec(domain=NN)] = 0,
    markup: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.1,
) -> None:
    super().__init__(label, visual=visual)
    self.stall = stall
    self.stock = stock
    self.markup = markup

StoreElement(label, visual=VisualSpec('agents', {'color_by': 'price'}), *, price=10.0, ask=10.0, supply=0.0, sales=0.0, money=100.0, profit=0.0, direction=1.0, capacity=1000.0, unit_cost=1.0, fixed_cost=20.0, adjustment=0.05)

Bases: Element

Shop that offers its capacity at its price, searches for the price of the largest profit, and may close.

The shop searches for its price by trial and error: it moves the price by adjustment in the same direction as long as its profit grows, and turns back when the profit falls, never below the unit cost. It pays the unit cost of every unit sold and the fixed cost every step, and closes when its money runs out.

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': 'price'})
price	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The price of a unit.	10.0
ask	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The price offered in the current step.	10.0
supply	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The units still offered in the step.	0.0
sales	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The units sold in the last step.	0.0
money	Annotated[float, StateSpec()]	The money of the shop.	100.0
profit	Annotated[float, StateSpec()]	The profit of the last step.	0.0
direction	Annotated[float, StateSpec()]	The direction of the last change of the price, 1 up or -1 down.	1.0
capacity	Annotated[float, ParameterSpec(domain=open(0, oo))]	The units the shop can sell in a step.	1000.0
unit_cost	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The cost of a unit sold.	1.0
fixed_cost	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The cost of running the shop for a step.	20.0
adjustment	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The relative change of the price in a step.	0.05

Examples:

>>> from skcomplex.core.components.elements import GroupElement
>>> from skcomplex.systems.economics.markets import StoreElement
>>> GroupElement('shops', element=StoreElement, size=4, values={'price': 10.0}).size
4

Source code in src/skcomplex/systems/economics/markets/_shops.py

def __init__(
    self,
    label: str,
    visual: VisualSpec | None = VisualSpec('agents', {'color_by': 'price'}),
    *,
    price: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 10.0,
    ask: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 10.0,
    supply: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
    sales: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
    money: Annotated[float, StateSpec()] = 100.0,
    profit: Annotated[float, StateSpec()] = 0.0,
    direction: Annotated[float, StateSpec()] = 1.0,
    capacity: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 1000.0,
    unit_cost: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
    fixed_cost: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 20.0,
    adjustment: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.05,
) -> None:
    super().__init__(label, visual=visual)
    self.price = price
    self.ask = ask
    self.supply = supply
    self.sales = sales
    self.money = money
    self.profit = profit
    self.direction = direction
    self.capacity = capacity
    self.unit_cost = unit_cost
    self.fixed_cost = fixed_cost
    self.adjustment = adjustment

get_opening() staticmethod

Return the states a shop sets when it opens for the market, its capacity offered at its price.

Returns:

Name	Type	Description
opening	tuple[tuple[str, Callable[[StoreView, SystemView], float]], ...]	The name of every state and the function of its value, of the shop and the system view.

Source code in src/skcomplex/systems/economics/markets/_shops.py

@staticmethod
def get_opening() -> tuple[tuple[str, Callable[[StoreView, SystemView], float]], ...]:
    """Return the states a shop sets when it opens for the market, its capacity offered at its price.

    Returns:
        opening:
            The name of every state and the function of its value, of the shop and the system view.
    """
    return (
        ('supply', lambda agent, system: agent.capacity),
        ('ask', lambda agent, system: agent.price),
        ('sales', lambda agent, system: 0.0),
    )

get_profit(agent) staticmethod

Return the profit of the step, the margin on the units sold less the fixed cost.

Parameters:

Name	Type	Description	Default
agent	StoreView	The shop.	required

Returns:

Name	Type	Description
profit	float	The profit of the step.

Source code in src/skcomplex/systems/economics/markets/_shops.py

@staticmethod
def get_profit(agent: StoreView) -> float:
    """Return the profit of the step, the margin on the units sold less the fixed cost.

    Args:
        agent:
            The shop.

    Returns:
        profit:
            The profit of the step.
    """
    return agent.sales * (agent.ask - agent.unit_cost) - agent.fixed_cost

get_settlement() staticmethod

Return the states a shop sets when it settles its accounts, in their order.

The shop turns its price back when its profit falls, records the profit, and pays its costs. The pricing of the system then steps the price by the sign of the direction.

Returns:

Name	Type	Description
settlement	tuple[tuple[str, Callable[[StoreView, SystemView], float]], ...]	The name of every state and the function of its value, of the shop and the system view.

Source code in src/skcomplex/systems/economics/markets/_shops.py

@staticmethod
def get_settlement() -> tuple[tuple[str, Callable[[StoreView, SystemView], float]], ...]:
    """Return the states a shop sets when it settles its accounts, in their order.

    The shop turns its price back when its profit falls, records the profit, and pays its costs. The pricing of
    the system then steps the price by the sign of the direction.

    Returns:
        settlement:
            The name of every state and the function of its value, of the shop and the system view.
    """
    profit = StoreElement.get_profit
    return (
        ('direction', lambda agent, system: -agent.direction if profit(agent) < agent.profit else agent.direction),
        ('profit', lambda agent, system: profit(agent)),
        ('money', lambda agent, system: agent.money - agent.sales * agent.unit_cost - agent.fixed_cost),
    )

record_sale(match, system) staticmethod

Count the units of a purchase in the sales of the shop.

Parameters:

Name	Type	Description	Default
match	MatchSpec	The purchase of a shopper, the requester, from a shop, the provider.	required
system	SystemView	The system view of the market.	required

Source code in src/skcomplex/systems/economics/markets/_shops.py

@staticmethod
def record_sale(match: MatchSpec, system: SystemView) -> None:
    """Count the units of a purchase in the sales of the shop.

    Args:
        match:
            The purchase of a shopper, the requester, from a shop, the provider.

        system:
            The system view of the market.
    """
    shop = match.provider
    if isinstance(shop, StoreView):
        shop.sales += match.quantity

SugarPatchElement(label, visual=VisualSpec('agents', {'color_by': 'sugar', 'color_range': (0.0, 4.0)}), *, sugar=0.0, capacity=0.0, growback=1.0)

Bases: Element

Cell of the landscape whose sugar grows back by a fixed amount a step up to its capacity.

The growback is a RegrowthInteraction of the group of patches.

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': 'sugar', 'color_range': (0.0, 4.0)})
sugar	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The sugar on the patch.	0.0
capacity	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The most sugar the patch holds.	0.0
growback	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The sugar that grows back in a step.	1.0

Examples:

>>> from skcomplex.core.components.elements import GroupElement
>>> from skcomplex.systems.economics.markets import SugarPatchElement
>>> GroupElement('patches', element=SugarPatchElement, size=4, values={'capacity': 2.0}).size
4

Source code in src/skcomplex/systems/economics/markets/_sugarscape.py

def __init__(
    self,
    label: str,
    visual: VisualSpec | None = VisualSpec('agents', {'color_by': 'sugar', 'color_range': (0.0, 4.0)}),
    *,
    sugar: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
    capacity: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
    growback: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
) -> None:
    super().__init__(label, visual=visual)
    self.sugar = sugar
    self.capacity = capacity
    self.growback = growback

SupplyLinkInteraction(label, *, downstream, upstream)

Bases: TransferInteraction

Link of a supply chain, which ships beer from the upstream echelon to the downstream echelon.

The shipments and their arrivals, after a third-order delay of the shipping delay of the downstream echelon, are the legs of one transfer, from the inventory of the upstream echelon to the beer in transit to the downstream echelon and on to its inventory.

Parameters:

Name	Type	Description	Default
label	str	The label of the link.	required
downstream	Annotated[str, RoleSpec()]	The path of the echelon that orders.	required
upstream	Annotated[str, RoleSpec()]	The path of the echelon that ships.	required

Examples:

>>> from skcomplex.systems.economics.markets import SupplyLinkInteraction
>>> SupplyLinkInteraction('link', downstream='retailer', upstream='wholesaler').upstream
'wholesaler'

Source code in src/skcomplex/systems/economics/markets/_supply_chains.py

def __init__(
    self,
    label: str,
    *,
    downstream: Annotated[str, RoleSpec()],
    upstream: Annotated[str, RoleSpec()],
) -> None:
    super().__init__(
        label,
        source=upstream,
        target=downstream,
        state='inventory',
        into='in_transit',
        amount=lambda s: EchelonElement._shipments(s.upstream),
        flow='shipments',
        legs=[
            (
                'downstream.in_transit',
                'downstream.inventory',
                lambda s: SupplyLinkInteraction._arrivals(s.upstream, s.downstream),
                'arrivals',
            ),
        ],
    )
    self.downstream = downstream
    self.upstream = upstream

TillElement(label, visual=None, *, revenue=0.0, washes=0, valets=0, turned_away=0)

Bases: Element

Till of the car wash, which counts the revenue and the cars served and turned away.

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
revenue	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The money taken.	0.0
washes	Annotated[float, StateSpec(domain=NN)]	The number of cars that left after a wash only.	0
valets	Annotated[float, StateSpec(domain=NN)]	The number of cars that left after a valet.	0
turned_away	Annotated[float, StateSpec(domain=NN)]	The number of drivers who did not turn in because of the queue.	0

Examples:

>>> from skcomplex.core.components.elements import GroupElement
>>> from skcomplex.systems.economics.markets import TillElement
>>> GroupElement('till', element=TillElement, size=1).size
1

Source code in src/skcomplex/systems/economics/markets/_queues.py

def __init__(
    self,
    label: str,
    visual: VisualSpec | None = None,
    *,
    revenue: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
    washes: Annotated[float, StateSpec(domain=NN)] = 0,
    valets: Annotated[float, StateSpec(domain=NN)] = 0,
    turned_away: Annotated[float, StateSpec(domain=NN)] = 0,
) -> None:
    super().__init__(label, visual=visual)
    self.revenue = revenue
    self.washes = washes
    self.valets = valets
    self.turned_away = turned_away

ToyMarketInteraction(label, *, rule=None, phase=None, traders, sight=1.0, mutation=0.0, imitation=0.1, elite=0.05, size=51.0, n_goods=1, fitness='none')

Bases: FeedbackInteraction[SystemView]

Market of the Toy Trader: every trader buys the cheapest good it can see, sellers price by inventory, all move.

Every iteration the traders, in random order, try to buy one unit of every good from the trader with the lowest price within the radius sight, picking at random among equal prices, and buy it when they can pay. After a sale the seller lowers its price by its factor price_down when its stock is above the target and raises it by price_up when it is below, never below 0.5. Then every trader moves one unit in a random direction on the torus. Toys are priced by the trader's own factors, which mutate with the probability mutation by the difference of two uniform draws on [0, 0.01), and crayons and blackboards by the factors 1.0101 and 0.99. With a fitness test, a trader of the poorest half copies with the probability imitation the toy price of a random trader of the richest elite share and, each with probability one half, its factors plus a uniform mutation on [-0.01, 0.01].

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
sight	Annotated[float, ParameterSpec(domain=open(0, oo))]	The radius within which a trader sees prices, its market sight.	1.0
mutation	Annotated[float, ParameterSpec(domain=closed(0, 1))]	The probability per iteration that a factor of the toy price mutates.	0.0
imitation	Annotated[float, ParameterSpec(domain=closed(0, 1))]	The probability per iteration that a poor trader imitates a rich one.	0.1
elite	Annotated[float, ParameterSpec(domain=open(0, 1))]	The share of the richest traders that poor traders imitate.	0.05
size	Annotated[float, ParameterSpec(domain=open(0, oo))]	The side of the torus.	51.0
n_goods	int	The goods traded, 1 for toys only or 3 for toys, crayons, and blackboards.	1
fitness	str	The fitness test of imitation, 'none', 'money', or 'assets'.	'none'

Raises:

Type	Description
ValidationError	The fitness test or the number of goods is unknown, raised by prepare.

Examples:

>>> from skcomplex.systems.economics.markets import ToyMarketInteraction
>>> ToyMarketInteraction('market', traders='traders').reports[:2]
('average_price', 'trades')

Source code in src/skcomplex/systems/economics/markets/_toy_trader.py

def __init__(
    self,
    label: str,
    *,
    rule: None = None,
    phase: str | None = None,
    traders: Annotated[str, RoleSpec()],
    sight: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 1.0,
    mutation: Annotated[float, ParameterSpec(domain=RealSet.closed(0, 1))] = 0.0,
    imitation: Annotated[float, ParameterSpec(domain=RealSet.closed(0, 1))] = 0.1,
    elite: Annotated[float, ParameterSpec(domain=RealSet.open(0, 1))] = 0.05,
    size: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 51.0,
    n_goods: int = 1,
    fitness: str = 'none',
) -> None:
    super().__init__(label, rule=rule, phase=phase)
    self.traders = traders
    self.sight = sight
    self.mutation = mutation
    self.imitation = imitation
    self.elite = elite
    self.size = size
    self.n_goods = n_goods
    self.fitness = fitness

act(system)

Run one iteration: trade, evolve the pricing rules, and move.

Parameters:

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

Source code in src/skcomplex/systems/economics/markets/_toy_trader.py

def act(self, system: SystemView) -> None:
    """Run one iteration: trade, evolve the pricing rules, and move.

    Args:
        system:
            The system view of the market.
    """
    traders = sorted(system.agents('traders', ToyTraderView), key=lambda trader: trader.id)
    prices_of = ('toy_price', 'crayon_price', 'blackboard_price')
    state = {
        'places': np.array([[trader.x, trader.y] for trader in traders]),
        'money': np.array([trader.money for trader in traders]),
        'stocks': np.array([[getattr(trader, good) for good in GOODS] for trader in traders]),
        'prices': np.array([[getattr(trader, name) for name in prices_of] for trader in traders]),
        'up': np.array([trader.price_up for trader in traders]),
        'down': np.array([trader.price_down for trader in traders]),
    }
    sold = self._trade(system, state)
    last = system.history('average_price', 1)
    fallback = float(last[-1]) if len(last) else float(state['prices'][:, 0].mean())
    average = float(np.mean(sold)) if sold else fallback
    self._evolve(system, state, average)
    angles = system.rng.uniform(0, 2 * np.pi, len(traders))
    steps = np.column_stack((np.cos(angles), np.sin(angles)))
    state['places'] = (state['places'] + steps) % system.parameter('size')
    for index, trader in enumerate(traders):
        trader.x, trader.y = (float(value) for value in state['places'][index])
        trader.money = float(state['money'][index])
        for good, name, value in zip(GOODS, prices_of, state['prices'][index], strict=True):
            setattr(trader, good, float(state['stocks'][index, GOODS.index(good)]))
            setattr(trader, name, float(value))
        trader.price_up, trader.price_down = float(state['up'][index]), float(state['down'][index])
    system.report('average_price', average)
    system.report('trades', float(len(sold)))
    system.report('money_gini', self._gini(state['money']))
    system.report('asset_gini', self._gini(state['money'] + state['stocks'][:, 0] * average))

ToyTraderElement(label, visual=VisualSpec('agents', {'color_by': 'money'}), *, x=0.0, y=0.0, money=100.0, toys=5.0, crayons=0.0, blackboards=0.0, toy_price=10.0, crayon_price=10.0, blackboard_price=10.0, price_up=PRICE_UP, price_down=PRICE_DOWN)

Bases: Element

Trader on a torus with money, toys, crayons, and blackboards, its selling prices, and its pricing rule.

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': 'money'})
x	Annotated[float, StateSpec()]	The horizontal place of the trader.	0.0
y	Annotated[float, StateSpec()]	The vertical place of the trader.	0.0
money	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The money of the trader.	100.0
toys	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The toys of the trader.	5.0
crayons	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The crayons of the trader.	0.0
blackboards	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The blackboards of the trader.	0.0
toy_price	Annotated[float, StateSpec(domain=open(0, oo))]	The price the trader sells a toy at.	10.0
crayon_price	Annotated[float, StateSpec(domain=open(0, oo))]	The price the trader sells a crayon at.	10.0
blackboard_price	Annotated[float, StateSpec(domain=open(0, oo))]	The price the trader sells a blackboard at.	10.0
price_up	Annotated[float, StateSpec(domain=closed_open(1, oo))]	The factor that raises the toy price when the toys run short.	PRICE_UP
price_down	Annotated[float, StateSpec(domain=closed(0, 1))]	The factor that lowers the toy price when the toys pile up.	PRICE_DOWN

Examples:

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

Source code in src/skcomplex/systems/economics/markets/_toy_trader.py

def __init__(
    self,
    label: str,
    visual: VisualSpec | None = VisualSpec('agents', {'color_by': 'money'}),
    *,
    x: Annotated[float, StateSpec()] = 0.0,
    y: Annotated[float, StateSpec()] = 0.0,
    money: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 100.0,
    toys: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 5.0,
    crayons: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
    blackboards: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
    toy_price: Annotated[float, StateSpec(domain=RealSet.open(0, oo))] = 10.0,
    crayon_price: Annotated[float, StateSpec(domain=RealSet.open(0, oo))] = 10.0,
    blackboard_price: Annotated[float, StateSpec(domain=RealSet.open(0, oo))] = 10.0,
    price_up: Annotated[float, StateSpec(domain=RealSet.closed_open(1, oo))] = PRICE_UP,
    price_down: Annotated[float, StateSpec(domain=RealSet.closed(0, 1))] = PRICE_DOWN,
) -> None:
    super().__init__(label, visual=visual)
    self.x = x
    self.y = y
    self.money = money
    self.toys = toys
    self.crayons = crayons
    self.blackboards = blackboards
    self.toy_price = toy_price
    self.crayon_price = crayon_price
    self.blackboard_price = blackboard_price
    self.price_up = price_up
    self.price_down = price_down

WasherElement(label, visual=None)

Bases: Element

Worker who joins the front car of the wash or of the valet while it has fewer than two workers.

A worker who finds both areas staffed helps at a random one of them. One second of work removes one person-second from the car.

Parameters:

Name	Type	Description	Default
label	str	The label of the agent or group.	required

Examples:

>>> from skcomplex.core.components.elements import GroupElement
>>> from skcomplex.systems.economics.markets import WasherElement
>>> GroupElement('workers', element=WasherElement, size=4).size
4

Source code in src/skcomplex/core/components/elements/_base.py

def __init__(self, label: str, visual: VisualSpec | None = None) -> None:
    super().__init__(label)
    self.visual = visual

work(agent, system)

Work one second on the front car of an area.

Parameters:

Name	Type	Description	Default
agent	AgentView	The worker.	required
system	SystemView	The system.	required

Source code in src/skcomplex/systems/economics/markets/_queues.py

def work(self, agent: AgentView, system: SystemView) -> None:
    """Work one second on the front car of an area.

    Args:
        agent:
            The worker.

        system:
            The system.
    """
    cars = system.agents(self.cars, CarView)
    fronts = [
        car for car in (WasherElement._front(cars, WASH), WasherElement._front(cars, VALET)) if car is not None
    ]
    if not fronts:
        return
    short = [car for car in fronts if car.crew < CREW_SIZE]
    car = short[0] if short else fronts[int(system.rng.integers(len(fronts)))]
    car.crew += 1
    car.work -= 1.0

ZeroIntelligenceBuyerElement(label, visual=None, *, value=100.0, bid=0.0, demand=1.0, profit=0.0, ceiling=200.0, constrained=1.0)

Bases: Element

Buyer of one unit per trading period who bids a random price, below its value under the budget constraint.

In every round of a trading period, a buyer that has not bought yet bids a price drawn uniformly from 0 to its value, or to the price ceiling without the constraint, the zero-intelligence traders of Gode and Sunder (1993).

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
value	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The value of the unit to the buyer, its redemption value.	100.0
bid	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The current bid.	0.0
demand	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The number of units still wanted in the trading period.	1.0
profit	Annotated[float, StateSpec()]	The accumulated surplus of the buyer's trades.	0.0
ceiling	Annotated[float, ParameterSpec(domain=open(0, oo))]	The highest price of the market.	200.0
constrained	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	One when the buyer never bids above its value, zero otherwise.	1.0

Examples:

>>> from skcomplex.core.components.elements import GroupElement
>>> from skcomplex.systems.economics.markets import ZeroIntelligenceBuyerElement
>>> GroupElement('buyers', element=ZeroIntelligenceBuyerElement, size=5, values={'value': 100.0}).size
5

Source code in src/skcomplex/systems/economics/markets/_double_auction.py

def __init__(
    self,
    label: str,
    visual: VisualSpec | None = None,
    *,
    value: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 100.0,
    bid: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
    demand: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
    profit: Annotated[float, StateSpec()] = 0.0,
    ceiling: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 200.0,
    constrained: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
) -> None:
    super().__init__(label, visual=visual)
    self.value = value
    self.bid = bid
    self.demand = demand
    self.profit = profit
    self.ceiling = ceiling
    self.constrained = constrained

get_bid(agent, system) staticmethod

Return a bid drawn uniformly from 0 to the value, or to the price ceiling without the constraint.

Parameters:

Name	Type	Description	Default
agent	ZeroIntelligenceBuyerView	The buyer.	required
system	SystemView	The system view of the quotes.	required

Returns:

Name	Type	Description
bid	float	The bid of the round.

Source code in src/skcomplex/systems/economics/markets/_double_auction.py

@staticmethod
def get_bid(agent: ZeroIntelligenceBuyerView, system: SystemView) -> float:
    """Return a bid drawn uniformly from 0 to the value, or to the price ceiling without the constraint.

    Args:
        agent:
            The buyer.

        system:
            The system view of the quotes.

    Returns:
        bid:
            The bid of the round.
    """
    return float(system.rng.uniform(0.0, agent.value if agent.constrained else agent.ceiling))

ZeroIntelligenceSellerElement(label, visual=None, *, value=100.0, ask=200.0, supply=1.0, profit=0.0, ceiling=200.0, constrained=1.0)

Bases: Element

Seller of one unit per trading period who asks a random price, above its cost under the budget constraint.

In every round of a trading period, a seller that has not sold yet asks a price drawn uniformly from its cost, or from 0 without the constraint, to the price ceiling.

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
value	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The cost of the unit to the seller.	100.0
ask	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The current ask.	200.0
supply	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The number of units still offered in the trading period.	1.0
profit	Annotated[float, StateSpec()]	The accumulated surplus of the seller's trades.	0.0
ceiling	Annotated[float, ParameterSpec(domain=open(0, oo))]	The highest price of the market.	200.0
constrained	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	One when the seller never asks below its cost, zero otherwise.	1.0

Examples:

>>> from skcomplex.core.components.elements import GroupElement
>>> from skcomplex.systems.economics.markets import ZeroIntelligenceSellerElement
>>> GroupElement('sellers', element=ZeroIntelligenceSellerElement, size=5, values={'value': 50.0}).size
5

Source code in src/skcomplex/systems/economics/markets/_double_auction.py

def __init__(
    self,
    label: str,
    visual: VisualSpec | None = None,
    *,
    value: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 100.0,
    ask: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 200.0,
    supply: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
    profit: Annotated[float, StateSpec()] = 0.0,
    ceiling: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 200.0,
    constrained: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
) -> None:
    super().__init__(label, visual=visual)
    self.value = value
    self.ask = ask
    self.supply = supply
    self.profit = profit
    self.ceiling = ceiling
    self.constrained = constrained

get_ask(agent, system) staticmethod

Return an ask drawn uniformly from the cost, or from 0 without the constraint, to the price ceiling.

Parameters:

Name	Type	Description	Default
agent	ZeroIntelligenceSellerView	The seller.	required
system	SystemView	The system view of the quotes.	required

Returns:

Name	Type	Description
ask	float	The ask of the round.

Source code in src/skcomplex/systems/economics/markets/_double_auction.py

@staticmethod
def get_ask(agent: ZeroIntelligenceSellerView, system: SystemView) -> float:
    """Return an ask drawn uniformly from the cost, or from 0 without the constraint, to the price ceiling.

    Args:
        agent:
            The seller.

        system:
            The system view of the quotes.

    Returns:
        ask:
            The ask of the round.
    """
    bottom = agent.value if agent.constrained else 0.0
    return float(system.rng.uniform(bottom, max(bottom, agent.ceiling)))

barter(n_barterers=2, *, chocolate=(9.0, 1.0), cigarettes=(2.0, 8.0), weights=(0.6, 0.3), quantity=0.02, random_state=0)

Return barter of chocolate and cigarettes, in an Edgeworth box of two barterers or among many prisoners.

Every step the barterers are paired at random, and every pair trades small quantities of one good for the other at the geometric mean of their marginal rates of substitution while both gain, so the totals of both goods are unchanged. Two barterers with the given endowments and weights of chocolate trade within the Edgeworth box until no trade improves both, which is a point of the contract curve, where their marginal rates of substitution are equal. More barterers are prisoners of war who receive random Red Cross parcels and random tastes, and their trades settle on a common price, as Radford (1945) observed (Hamill and Gilbert 2016 ch. 5).

Parameters:

Name	Type	Description	Default
n_barterers	int	The number of barterers, 2 for the Edgeworth box.	2
chocolate	tuple[float, float]	The chocolate of the two barterers of the Edgeworth box, and the range of the parcels of more barterers.	(9.0, 1.0)
cigarettes	tuple[float, float]	The cigarettes of the two barterers of the Edgeworth box, and the range of the parcels of more barterers.	(2.0, 8.0)
weights	tuple[float, float]	The weights of chocolate of the two barterers of the Edgeworth box, and the range of the weights of more barterers.	(0.6, 0.3)
quantity	float	The quantity of the bought good in every trade.	0.02
random_state	int | Generator | None	The seed or generator of the parcels and the weights of more than two barterers.	0

Returns:

Name	Type	Description
system	System	The system, with the group barterers and the exchange exchange, whose reporters are the price of cigarettes in chocolate exchange.price, its exchange.dispersion, and exchange.volume.

Raises:

Type	Description
ValueError	There are fewer than two barterers.

Examples:

>>> from skcomplex.systems.economics.markets import barter
>>> barter().elements[0].prepare().size_
2

Source code in src/skcomplex/systems/economics/markets/_barter.py

@register_system(
    field='economics/markets',
    tags=('market', 'agents', 'barter', 'Edgeworth box', 'contract curve'),
    references=('Hamill and Gilbert 2016 ch. 5', 'Radford 1945'),
    defaults={'until': 20},
    observables=_barter_observables,
)
def barter(
    n_barterers: int = 2,
    *,
    chocolate: tuple[float, float] = (9.0, 1.0),
    cigarettes: tuple[float, float] = (2.0, 8.0),
    weights: tuple[float, float] = (0.6, 0.3),
    quantity: float = 0.02,
    random_state: int | np.random.Generator | None = 0,
) -> System:
    """Return barter of chocolate and cigarettes, in an Edgeworth box of two barterers or among many prisoners.

    Every step the barterers are paired at random, and every pair trades small quantities of one good for the other
    at the geometric mean of their marginal rates of substitution while both gain, so the totals of both goods are
    unchanged. Two barterers with the given endowments and weights of chocolate trade within the Edgeworth box until
    no trade improves both, which is a point of the contract curve, where their marginal rates of substitution are
    equal. More barterers are prisoners of war who receive random Red Cross parcels and random tastes, and their
    trades settle on a common price, as Radford (1945) observed (Hamill and Gilbert 2016 ch. 5).

    Args:
        n_barterers:
            The number of barterers, 2 for the Edgeworth box.

        chocolate:
            The chocolate of the two barterers of the Edgeworth box, and the range of the parcels of more barterers.

        cigarettes:
            The cigarettes of the two barterers of the Edgeworth box, and the range of the parcels of more
            barterers.

        weights:
            The weights of chocolate of the two barterers of the Edgeworth box, and the range of the weights of more
            barterers.

        quantity:
            The quantity of the bought good in every trade.

        random_state:
            The seed or generator of the parcels and the weights of more than two barterers.

    Returns:
        system:
            The system, with the group `barterers` and the exchange `exchange`, whose reporters are the price of
            cigarettes in chocolate `exchange.price`, its `exchange.dispersion`, and `exchange.volume`.

    Raises:
        ValueError:
            There are fewer than two barterers.

    Examples:
        >>> from skcomplex.systems.economics.markets import barter
        >>> barter().elements[0].prepare().size_
        2
    """
    if n_barterers < EDGEWORTH_BARTERERS:
        error_msg = f'Barter needs at least two barterers, but got {n_barterers}.'
        raise ValueError(error_msg)
    rng = np.random.default_rng(random_state)
    if n_barterers == EDGEWORTH_BARTERERS:
        holdings: dict[str, list[float] | npt.NDArray[np.float64]] = {
            'chocolate': list(chocolate),
            'cigarettes': list(cigarettes),
        }
        shares = np.asarray(weights, dtype=float)
    else:
        holdings = {
            'chocolate': rng.uniform(min(chocolate), max(chocolate), n_barterers),
            'cigarettes': rng.uniform(min(cigarettes), max(cigarettes), n_barterers),
        }
        shares = rng.uniform(min(weights), max(weights), n_barterers)
    barterers = GroupElement(
        'barterers',
        element=BartererElement,
        size=n_barterers,
        values={'chocolate_weight': shares, 'cigarettes_weight': 1 - shares, **holdings},
        variable=True,
    )
    return System(
        elements=[barterers],
        interactions=[
            ExchangeInteraction('exchange', source='barterers', target='barterers', goods=GOODS, quantity=quantity),
        ],
        time=DiscreteTime(schedule=PhasesScheduleSpec([PhaseSpec('trade')])),
    )

beer_game(inventory=12.0, *, demand=4.0, demand_step=4.0, step_time=5.0, theta=0.36, alpha=0.26, beta=0.34, dispatch_time=1.0, mail_delay=2.0, shipping_delay=2.0, production_delay=3.0)

Return Sterman's beer distribution game, a supply chain whose orders swing wider at every echelon upstream.

A retailer, a wholesaler, a distributor, and a factory each hold an inventory of beer and a backlog of orders waiting to ship. Each ships its backlog over the dispatch time, limited by its inventory, and the beer reaches the next echelon downstream after the shipping delay. Orders reach the next echelon upstream after the mail delay, and the factory brews its orders over the production delay. Customer demand steps from 4 to 8 cases a week in week 5. Every echelon orders by Sterman's anchoring and adjustment rule, its expected demand plus alpha times the gap between its desired stock S' and its effective inventory, the inventory less the backlog, plus beta times its supply line, the beer it ordered and has not received. The expected demand adapts to incoming orders at the rate theta. The defaults of theta, alpha, and beta are the median estimates of Sterman 1989, and S' is set so the chain starts in equilibrium at 4 cases a week. The game is weekly and its delays are fixed, while this system is a continuous approximation in which third-order delays stand in for the fixed ones. The amplitude of the orders grows from the retailer to the factory, the bullwhip effect.

Parameters:

Name	Type	Description	Default
inventory	float	The initial inventory of every echelon, in cases.	12.0
demand	float	The customer demand, in cases a week.	4.0
demand_step	float	The permanent change of the customer demand.	4.0
step_time	float	The week of the change of the customer demand.	5.0
theta	float	The rate at which the expected demand adapts to incoming orders.	0.36
alpha	float	The fraction of the stock gap ordered per week.	0.26
beta	float	The weight of the supply line in the stock gap.	0.34
dispatch_time	float	The time over which an echelon ships its backlog.	1.0
mail_delay	float	The average time an order takes to reach the next echelon upstream.	2.0
shipping_delay	float	The average time a shipment takes to reach the next echelon downstream.	2.0
production_delay	float	The average time the factory takes to brew its orders.	3.0

Returns:

Name	Type	Description
system	System	The system, with the stocks <echelon>.inventory, <echelon>.in_transit, <echelon>.backlog, <echelon>.expected, and <echelon>.supply_line of the echelons retailer, wholesaler, distributor, and factory, and the flows <upstream>_to_<downstream>.shipments and <upstream>_to_<downstream>.arrivals of the links, customers.sales, and brewery.starts and brewery.completions. The rates <echelon>_backlog and <echelon>_expected add the orders that reach every echelon and adapt its expectation, and the rates <echelon>_supply_line keep its supply line. The observables of the system are the orders <echelon>.orders and the shipments <echelon>.shipments of every echelon, and the orders of the customers customers.orders.

Examples:

>>> from skcomplex.systems.economics.markets import beer_game
>>> [element.label for element in beer_game().elements]
['retailer', 'wholesaler', 'distributor', 'factory']

Source code in src/skcomplex/systems/economics/markets/_supply_chains.py

@register_system(
    field='economics/markets',
    tags=('supply chain', 'bullwhip effect', 'delay', 'stocks and flows', 'system dynamics', 'ode'),
    references=('Sterman 1989',),
    defaults={'until': 36.0, 'dt': 0.5},
    observables=lambda system: [
        *(ObservableSpec('orders', BreweryInteraction._orders, element=label) for label in ECHELONS),
        *(ObservableSpec('shipments', EchelonElement._shipments, element=label) for label in ECHELONS),
        ObservableSpec('customers.orders', _customer_orders),
    ],
)
def beer_game(
    inventory: float = 12.0,
    *,
    demand: float = 4.0,
    demand_step: float = 4.0,
    step_time: float = 5.0,
    theta: float = 0.36,
    alpha: float = 0.26,
    beta: float = 0.34,
    dispatch_time: float = 1.0,
    mail_delay: float = 2.0,
    shipping_delay: float = 2.0,
    production_delay: float = 3.0,
) -> System:
    """Return Sterman's beer distribution game, a supply chain whose orders swing wider at every echelon upstream.

    A retailer, a wholesaler, a distributor, and a factory each hold an inventory of beer and a backlog of orders
    waiting to ship. Each ships its backlog over the dispatch time, limited by its inventory, and the beer reaches the
    next echelon downstream after the shipping delay. Orders reach the next echelon upstream after the mail delay, and
    the factory brews its orders over the production delay. Customer demand steps from 4 to 8 cases a week in week 5.
    Every echelon orders by Sterman's anchoring and adjustment rule, its expected demand plus `alpha` times the gap
    between its desired stock `S'` and its effective inventory, the inventory less the backlog, plus `beta` times its
    supply line, the beer it ordered and has not received. The expected demand adapts to incoming orders at the rate
    `theta`. The defaults of `theta`, `alpha`, and `beta` are the median estimates of Sterman 1989, and `S'` is set so
    the chain starts in equilibrium at 4 cases a week. The game is weekly and its delays are fixed, while this system is
    a continuous approximation in which third-order delays stand in for the fixed ones. The amplitude of the orders
    grows from the retailer to the factory, the bullwhip effect.

    Args:
        inventory:
            The initial inventory of every echelon, in cases.

        demand:
            The customer demand, in cases a week.

        demand_step:
            The permanent change of the customer demand.

        step_time:
            The week of the change of the customer demand.

        theta:
            The rate at which the expected demand adapts to incoming orders.

        alpha:
            The fraction of the stock gap ordered per week.

        beta:
            The weight of the supply line in the stock gap.

        dispatch_time:
            The time over which an echelon ships its backlog.

        mail_delay:
            The average time an order takes to reach the next echelon upstream.

        shipping_delay:
            The average time a shipment takes to reach the next echelon downstream.

        production_delay:
            The average time the factory takes to brew its orders.

    Returns:
        system:
            The system, with the stocks `<echelon>.inventory`, `<echelon>.in_transit`, `<echelon>.backlog`,
            `<echelon>.expected`, and `<echelon>.supply_line` of the echelons `retailer`, `wholesaler`, `distributor`,
            and `factory`, and the flows `<upstream>_to_<downstream>.shipments` and
            `<upstream>_to_<downstream>.arrivals` of the links, `customers.sales`, and `brewery.starts` and
            `brewery.completions`. The rates `<echelon>_backlog` and `<echelon>_expected` add the orders that reach
            every echelon and adapt its expectation, and the rates `<echelon>_supply_line` keep its supply line. The
            observables of the system are the orders `<echelon>.orders` and the shipments
            `<echelon>.shipments` of every echelon, and the orders of the customers `customers.orders`.

    Examples:
        >>> from skcomplex.systems.economics.markets import beer_game
        >>> [element.label for element in beer_game().elements]
        ['retailer', 'wholesaler', 'distributor', 'factory']
    """
    throughput = demand * dispatch_time
    echelons: list[EchelonElement] = []
    for label in ECHELONS:
        pipeline = production_delay if label == 'factory' else shipping_delay
        lead_time = pipeline if label == 'factory' else mail_delay + shipping_delay
        supply_line = demand * lead_time
        values = {
            'inventory': inventory,
            'in_transit': demand * pipeline,
            'backlog': throughput,
            'expected': demand,
            'supply_line': supply_line,
            'desired': inventory - throughput + beta * supply_line,
            'theta': theta,
            'alpha': alpha,
            'beta': beta,
            'dispatch_time': dispatch_time,
            'mail_delay': mail_delay,
            'shipping_delay': shipping_delay,
        }
        if label == 'retailer':
            values.update(demand=demand, demand_step=demand_step, step_time=step_time)
        elif label == 'factory':
            values.update(production_delay=production_delay)
        kind = {'retailer': RetailerElement, 'factory': FactoryElement}.get(label, EchelonElement)
        echelons.append(kind(label).set_params(**values))
    links = [
        interaction
        for downstream, upstream in pairwise(ECHELONS)
        for interaction in (
            *_get_link_rates(downstream, upstream),
            SupplyLinkInteraction(f'{upstream}_to_{downstream}', downstream=downstream, upstream=upstream),
        )
    ]
    customers = [*_get_customer_rates(), RetailDemandInteraction('customers', retailer='retailer')]
    brewery = [_get_brewery_rate(), BreweryInteraction('brewery', factory='factory')]
    return System(elements=echelons, interactions=[*links, *customers, *brewery])

brock_hommes(n_traders=1000, *, intensity=1.0, rate=1.1, strength=1.2, cost=1.0, risk=1.0, noise=0.01, initial=0.1)

Return Brock and Hommes' asset market with traders switching between fundamentalists and trend chasers.

Every step the traders score every heuristic by its last realised fitness and choose a heuristic by a logit choice of intensity beta on its last realised fitness, and the market maker clears the deviation of the price from the fundamental at R x_t = sum_h n_h f_h. The fundamental steady state is stable while the share of trend chasers there, 1 / (1 + exp(-beta C)), is below R / g, and beyond the critical intensity beta* = ln(R / (g - R)) / C the price fluctuates persistently around the fundamental.

Parameters:

Name	Type	Description	Default
n_traders	int	The number of traders.	1000
intensity	float	The intensity of choice beta.	1.0
rate	float	The gross return R of the riskless asset.	1.1
strength	float	The extrapolation g of the trend chasers.	1.2
cost	float	The cost C of the fundamentalist heuristic.	1.0
risk	float	The risk aversion times the variance of returns, a sigma^2.	1.0
noise	float	The standard deviation of the noise of the price.	0.01
initial	float	The initial deviation of the price from its fundamental value.	0.1

Returns:

Name	Type	Description
system	System	The system, with the group traders, their scores learning and choice choice, and the institution pricing, whose reporters are the deviation of the price pricing.deviation, the share of trend chasers pricing.trend_share, and the fitnesses pricing.fitness_fundamentalist and pricing.fitness_trend.

Examples:

>>> from skcomplex.systems.economics.markets import brock_hommes
>>> brock_hommes(100).elements[0].prepare().size_
100

Source code in src/skcomplex/systems/economics/markets/_brock_hommes.py

@register_system(
    field='economics/markets',
    tags=('asset pricing', 'heuristics', 'bounded rationality', 'agents'),
    references=('Brock and Hommes 1998',),
    defaults={'until': 100},
)
def brock_hommes(
    n_traders: int = 1000,
    *,
    intensity: float = 1.0,
    rate: float = 1.1,
    strength: float = 1.2,
    cost: float = 1.0,
    risk: float = 1.0,
    noise: float = 0.01,
    initial: float = 0.1,
) -> System:
    """Return Brock and Hommes' asset market with traders switching between fundamentalists and trend chasers.

    Every step the traders score every heuristic by its last realised fitness and choose a heuristic by a logit choice
    of intensity `beta` on its last realised fitness, and the market maker clears the deviation of the price from the
    fundamental at `R x_t = sum_h n_h f_h`. The fundamental steady state is stable while the share of trend chasers
    there, `1 / (1 + exp(-beta C))`, is below `R / g`, and beyond the critical intensity `beta* = ln(R / (g - R)) / C`
    the price fluctuates persistently around the fundamental.

    Args:
        n_traders:
            The number of traders.

        intensity:
            The intensity of choice `beta`.

        rate:
            The gross return `R` of the riskless asset.

        strength:
            The extrapolation `g` of the trend chasers.

        cost:
            The cost `C` of the fundamentalist heuristic.

        risk:
            The risk aversion times the variance of returns, `a sigma^2`.

        noise:
            The standard deviation of the noise of the price.

        initial:
            The initial deviation of the price from its fundamental value.

    Returns:
        system:
            The system, with the group `traders`, their scores `learning` and choice `choice`, and the institution
            `pricing`, whose reporters are the deviation of the price `pricing.deviation`, the share of trend chasers
            `pricing.trend_share`, and the fitnesses `pricing.fitness_fundamentalist` and `pricing.fitness_trend`.

    Examples:
        >>> from skcomplex.systems.economics.markets import brock_hommes
        >>> brock_hommes(100).elements[0].prepare().size_
        100
    """
    traders = GroupElement('traders', element=HeuristicTraderElement, size=n_traders)
    learning = ReinforcementInteraction(
        'learning',
        target='traders',
        scores=tuple(f'{name}_fitness' for name in PREDICTORS),
        payoff=HeuristicTraderElement.get_realised_fitness,
        rate=1.0,
        form='agent',
        phase='learning',
    )
    choice = ChoiceInteraction(
        'choice',
        target='traders',
        state='predictor',
        options=(0, 1),
        utility=HeuristicTraderElement.get_fitness,
        intensity=intensity,
        form='agent',
        phase='choice',
    )
    pricing = HeuristicPricingInteraction(
        PRICING,
        traders='traders',
        phase='pricing',
        rate=rate,
        strength=strength,
        cost=cost,
        risk=risk,
        noise=noise,
        initial=initial,
    )
    return System(
        elements=[traders],
        interactions=[learning, choice, pricing],
        time=DiscreteTime(
            schedule=PhasesScheduleSpec([PhaseSpec('learning'), PhaseSpec('choice'), PhaseSpec('pricing')]),
        ),
    )

car_dealer(inventory=200.0, *, demand=20.0, demand_step=2.0, step_time=25.0, coverage=10.0, perception_delay=5.0, response_delay=3.0, delivery_delay=5.0)

Return Meadows' car dealership, whose inventory oscillates after a step in demand because of its delays.

The inventory is a stock of cars that deliveries from the factory fill and sales to customers drain. Sales are the customer demand, limited by the cars on the lot. The dealer smooths sales over the perception delay, wants an inventory of coverage days of perceived sales, and orders the perceived sales plus the gap to that inventory divided by the response delay. Orders arrive through a third-order material delay, a continuous stand-in for Meadows' fixed delivery delay. Lengthening the response delay damps the oscillation, while shortening the perception delay makes the dealer react faster to the new demand and slightly enlarges it.

Parameters:

Name	Type	Description	Default
inventory	float	The initial number of cars on the lot.	200.0
demand	float	The customer demand in cars per day.	20.0
demand_step	float	The permanent change of the demand.	2.0
step_time	float	The day of the change of the demand.	25.0
coverage	float	The days of perceived sales the dealer wants on the lot.	10.0
perception_delay	float	The time over which the dealer averages sales.	5.0
response_delay	float	The time over which the dealer closes the inventory gap.	3.0
delivery_delay	float	The average time between an order and its delivery.	5.0

Returns:

Name	Type	Description
system	System	The system, with the cars on the lot dealer.inventory, and the flows dealer.deliveries and dealer.sales. The observables of the system are dealer.perceived for the perceived sales, dealer.desired_inventory, and dealer.orders.

Examples:

>>> from skcomplex.systems.economics.markets import car_dealer
>>> car_dealer(response_delay=6.0).elements[0].response_delay
6.0

Source code in src/skcomplex/systems/economics/markets/_inventory.py

@register_system(
    field='economics/markets',
    tags=('inventory', 'delay', 'oscillations', 'stocks and flows', 'system dynamics', 'ode'),
    references=('Meadows 2008',),
    defaults={'until': 100.0, 'dt': 1.0},
    observables=lambda system: [
        ObservableSpec('perceived', _perceived, element='dealer'),
        ObservableSpec('desired_inventory', _desired_inventory, element='dealer'),
        ObservableSpec('orders', _orders, element='dealer'),
    ],
)
def car_dealer(
    inventory: float = 200.0,
    *,
    demand: float = 20.0,
    demand_step: float = 2.0,
    step_time: float = 25.0,
    coverage: float = 10.0,
    perception_delay: float = 5.0,
    response_delay: float = 3.0,
    delivery_delay: float = 5.0,
) -> System:
    """Return Meadows' car dealership, whose inventory oscillates after a step in demand because of its delays.

    The inventory is a stock of cars that deliveries from the factory fill and sales to customers drain. Sales are the
    customer demand, limited by the cars on the lot. The dealer smooths sales over the perception delay, wants an
    inventory of `coverage` days of perceived sales, and orders the perceived sales plus the gap to that inventory
    divided by the response delay. Orders arrive through a third-order material delay, a continuous stand-in for
    Meadows' fixed delivery delay. Lengthening the response delay damps the oscillation, while shortening the
    perception delay makes the dealer react faster to the new demand and slightly enlarges it.

    Args:
        inventory:
            The initial number of cars on the lot.

        demand:
            The customer demand in cars per day.

        demand_step:
            The permanent change of the demand.

        step_time:
            The day of the change of the demand.

        coverage:
            The days of perceived sales the dealer wants on the lot.

        perception_delay:
            The time over which the dealer averages sales.

        response_delay:
            The time over which the dealer closes the inventory gap.

        delivery_delay:
            The average time between an order and its delivery.

    Returns:
        system:
            The system, with the cars on the lot `dealer.inventory`, and the flows `dealer.deliveries` and
            `dealer.sales`. The observables of the system are `dealer.perceived` for the perceived sales,
            `dealer.desired_inventory`, and `dealer.orders`.

    Examples:
        >>> from skcomplex.systems.economics.markets import car_dealer
        >>> car_dealer(response_delay=6.0).elements[0].response_delay
        6.0
    """
    dealer = CarDealershipElement(
        'dealer',
        inventory=inventory,
        demand=demand,
        demand_step=demand_step,
        step_time=step_time,
        coverage=coverage,
        perception_delay=perception_delay,
        response_delay=response_delay,
        delivery_delay=delivery_delay,
    )
    cars = TransferInteraction(
        'cars',
        source='dealer',
        target='dealer',
        state=Source('factory'),
        into='inventory',
        amount=_deliveries,
        flow='deliveries',
        legs=[('inventory', Sink('customers'), lambda s: _sales(_dealer(s)), 'sales')],
    )
    return System(elements=[dealer], interactions=[cars])

car_wash(n_workers=4, n_cars=1, *, p_pass=0.33, p_enter=0.005, p_valet=0.5, wash_work=525.0, valet_work=1750.0, wash_price=4.0, valet_price=10.0, balk=3)

Return Chesney's Clean and Wax hand car wash, a shadow account of the revenue of its workers' labour.

Every step is one second. Passing cars turn in with a probability and queue for the wash, washed cars may queue for a valet, and the workers serve the front car of each area, two at a time if they can. The wash opens with cars already queuing.

Parameters:

Name	Type	Description	Default
n_workers	int	The number of workers.	4
n_cars	int	The number of cars queuing for the wash at the start, at least one.	1
p_pass	float	The probability that a car passes in a second.	0.33
p_enter	float	The probability that a passing car turns in.	0.005
p_valet	float	The probability that a washed car stays for a valet.	0.5
wash_work	float	The person-seconds of a wash.	525.0
valet_work	float	The person-seconds of a valet.	1750.0
wash_price	float	The price of a wash.	4.0
valet_price	float	The price of a wash with a valet.	10.0
balk	int	The length of the wash queue at which drivers no longer turn in.	3

Returns:

Name	Type	Description
system	System	The system, with the groups workers, cars, and till.

Examples:

>>> from skcomplex.systems.economics.markets import car_wash
>>> car_wash(5).elements[0].prepare().size_
5

Source code in src/skcomplex/systems/economics/markets/_queues.py

@register_system(
    field='economics/markets',
    tags=('queue', 'agents', 'labour'),
    references=('Chesney 2021',),
    defaults={'until': 600},
)
def car_wash(
    n_workers: int = 4,
    n_cars: int = 1,
    *,
    p_pass: float = 0.33,
    p_enter: float = 0.005,
    p_valet: float = 0.5,
    wash_work: float = 525.0,
    valet_work: float = 1750.0,
    wash_price: float = 4.0,
    valet_price: float = 10.0,
    balk: int = 3,
) -> System:
    """Return Chesney's Clean and Wax hand car wash, a shadow account of the revenue of its workers' labour.

    Every step is one second. Passing cars turn in with a probability and queue for the wash, washed cars may queue
    for a valet, and the workers serve the front car of each area, two at a time if they can. The wash opens with
    cars already queuing.

    Args:
        n_workers:
            The number of workers.

        n_cars:
            The number of cars queuing for the wash at the start, at least one.

        p_pass:
            The probability that a car passes in a second.

        p_enter:
            The probability that a passing car turns in.

        p_valet:
            The probability that a washed car stays for a valet.

        wash_work:
            The person-seconds of a wash.

        valet_work:
            The person-seconds of a valet.

        wash_price:
            The price of a wash.

        valet_price:
            The price of a wash with a valet.

        balk:
            The length of the wash queue at which drivers no longer turn in.

    Returns:
        system:
            The system, with the groups `workers`, `cars`, and `till`.

    Examples:
        >>> from skcomplex.systems.economics.markets import car_wash
        >>> car_wash(5).elements[0].prepare().size_
        5
    """
    operation = CarWashOperationInteraction(
        'wash',
        cars='cars',
        till='till',
        p_pass=p_pass,
        p_enter=p_enter,
        p_valet=p_valet,
        wash_work=wash_work,
        valet_work=valet_work,
        wash_price=wash_price,
        valet_price=valet_price,
        balk=float(balk),
    )
    return System(
        elements=[
            GroupElement('workers', element=WasherElement, size=n_workers, variable=True),
            GroupElement('cars', element=CarElement, size=n_cars, values={'work': wash_work}, variable=True),
            GroupElement('till', element=TillElement, size=1, variable=True),
        ],
        interactions=[operation],
    )

cobb_douglas(agent, goods, weights)

Return the logarithm of the Cobb-Douglas utility of an agent, sum_i a_i log x_i.

Parameters:

Name	Type	Description	Default
agent	object	The agent, with a state for the holding of every good and for the weight of every good.	required
goods	Sequence[str]	The names of the states that hold the goods.	required
weights	Sequence[str]	The names of the states that hold the weights of the goods, in the order of goods.	required

Returns:

Name	Type	Description
utility	float	The logarithm of the utility, minus infinity when a weighted good runs out.

Examples:

>>> from types import SimpleNamespace
>>> from skcomplex.systems.economics.markets import cobb_douglas
>>> agent = SimpleNamespace(x=4.0, y=1.0, a=0.5, b=0.5)
>>> round(cobb_douglas(agent, ('x', 'y'), ('a', 'b')), 6)
0.693147

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

def cobb_douglas(agent: object, goods: Sequence[str], weights: Sequence[str]) -> float:
    """Return the logarithm of the Cobb-Douglas utility of an agent, `sum_i a_i log x_i`.

    Args:
        agent:
            The agent, with a state for the holding of every good and for the weight of every good.

        goods:
            The names of the states that hold the goods.

        weights:
            The names of the states that hold the weights of the goods, in the order of `goods`.

    Returns:
        utility:
            The logarithm of the utility, minus infinity when a weighted good runs out.

    Examples:
        >>> from types import SimpleNamespace
        >>> from skcomplex.systems.economics.markets import cobb_douglas
        >>> agent = SimpleNamespace(x=4.0, y=1.0, a=0.5, b=0.5)
        >>> round(cobb_douglas(agent, ('x', 'y'), ('a', 'b')), 6)
        0.693147
    """
    total = 0.0
    for good, weight in zip(goods, weights, strict=True):
        share, holding = float(getattr(agent, weight)), float(getattr(agent, good))
        if share == 0:
            continue
        if holding <= 0:
            return -math.inf
        total += share * math.log(holding)
    return total

cobweb(price=1.0, *, a=10.0, b=2.0, c=1.0, d=1.0)

Return a market whose price converges to or oscillates around equilibrium as supply lags the price.

Parameters:

Name	Type	Description	Default
price	float	The initial price.	1.0
a	float	The demand intercept.	10.0
b	float	The demand slope.	2.0
c	float	The supply intercept.	1.0
d	float	The supply slope.	1.0

Returns:

Name	Type	Description
system	System	The system, with the market market.

Examples:

>>> from skcomplex.systems.economics.markets import cobweb
>>> cobweb(price=2.0).elements[0].price
2.0

Source code in src/skcomplex/systems/economics/markets/_cobweb.py

@register_system(
    field='economics/markets',
    tags=('market', 'discrete map'),
    references=('Ezekiel 1938',),
    defaults={'until': 30, 'dt': 1},
)
def cobweb(price: float = 1.0, *, a: float = 10.0, b: float = 2.0, c: float = 1.0, d: float = 1.0) -> System:
    """Return a market whose price converges to or oscillates around equilibrium as supply lags the price.

    Args:
        price:
            The initial price.

        a:
            The demand intercept.

        b:
            The demand slope.

        c:
            The supply intercept.

        d:
            The supply slope.

    Returns:
        system:
            The system, with the market `market`.

    Examples:
        >>> from skcomplex.systems.economics.markets import cobweb
        >>> cobweb(price=2.0).elements[0].price
        2.0
    """
    return System(
        elements=[MarketElement('market', price=price)],
        interactions=[CobwebInteraction('cobweb', target='market', a=a, b=b, c=c, d=d)],
    )

commodity_cycle(inventory=25.0, capacity=100.0, *, demand=100.0, demand_step=10.0, step_time=2.0, desired_coverage=0.25, reference_price=1.0, supply_elasticity=0.1, demand_elasticity=0.25, expectation_time=1.0, construction_delay=3.0, lifetime=15.0)

Return a commodity market whose price and capacity cycle because new capacity arrives after a delay.

The inventory is a stock of the commodity that production fills and consumption drains. The price is a falling table function of the inventory coverage, the years of demand the inventory holds, relative to the desired coverage. A high price raises the utilization of capacity and lowers consumption with constant elasticities. Producers form an expected price by smoothing the price, and order new capacity at the rate of retirements multiplied by a rising table function of the expected price. Orders become capacity through a third-order delay. After a step in demand the market overshoots, so the price and the capacity oscillate, the commodity cycle of Bala et al. Section 4.11.4. The book was not at hand, so the tables and the parameters are a documented reconstruction of that structure, in years and in units of the reference demand and price. The market starts in equilibrium.

Parameters:

Name	Type	Description	Default
inventory	float	The initial inventory of the commodity.	25.0
capacity	float	The initial production capacity per year.	100.0
demand	float	The demand per year at the reference price.	100.0
demand_step	float	The permanent change of the demand.	10.0
step_time	float	The year of the change of the demand.	2.0
desired_coverage	float	The years of demand producers want the inventory to hold.	0.25
reference_price	float	The price at the desired coverage.	1.0
supply_elasticity	float	The elasticity of the utilization of capacity with respect to the price.	0.1
demand_elasticity	float	The elasticity of consumption with respect to the price, as a positive number.	0.25
expectation_time	float	The time over which producers smooth the price into their expected price.	1.0
construction_delay	float	The average time between an order of capacity and its completion.	3.0
lifetime	float	The average lifetime of capacity.	15.0

Returns:

Name	Type	Description
system	System	The system, with the inventory market.inventory and the capacity market.capacity in years, the flows market.production, market.consumption, market.completions, and market.retirements. The observables of the system are market.coverage, market.price, market.expected_price, and market.capacity_orders.

Examples:

>>> from skcomplex.systems.economics.markets import commodity_cycle
>>> commodity_cycle(construction_delay=5.0).elements[0].construction_delay
5.0

Source code in src/skcomplex/systems/economics/markets/_commodities.py

@register_system(
    field='economics/markets',
    tags=('commodity cycle', 'delay', 'oscillations', 'stocks and flows', 'system dynamics', 'ode'),
    references=('Bala et al. 2017',),
    defaults={'until': 60.0, 'dt': 0.25},
    observables=lambda system: [
        ObservableSpec('coverage', _coverage, element='market'),
        ObservableSpec('price', _price, element='market'),
        ObservableSpec('expected_price', _expected_price, element='market'),
        ObservableSpec('capacity_orders', _capacity_orders, element='market'),
    ],
)
def commodity_cycle(
    inventory: float = 25.0,
    capacity: float = 100.0,
    *,
    demand: float = 100.0,
    demand_step: float = 10.0,
    step_time: float = 2.0,
    desired_coverage: float = 0.25,
    reference_price: float = 1.0,
    supply_elasticity: float = 0.1,
    demand_elasticity: float = 0.25,
    expectation_time: float = 1.0,
    construction_delay: float = 3.0,
    lifetime: float = 15.0,
) -> System:
    """Return a commodity market whose price and capacity cycle because new capacity arrives after a delay.

    The inventory is a stock of the commodity that production fills and consumption drains. The price is a falling
    table function of the inventory coverage, the years of demand the inventory holds, relative to the desired
    coverage. A high price raises the utilization of capacity and lowers consumption with constant elasticities.
    Producers form an expected price by smoothing the price, and order new capacity at the rate of retirements
    multiplied by a rising table function of the expected price. Orders become capacity through a third-order delay.
    After a step in demand the market overshoots, so the price and the capacity oscillate, the commodity cycle of
    Bala et al. Section 4.11.4. The book was not at hand, so the tables and the parameters are a documented
    reconstruction of that structure, in years and in units of the reference demand and price. The market starts in
    equilibrium.

    Args:
        inventory:
            The initial inventory of the commodity.

        capacity:
            The initial production capacity per year.

        demand:
            The demand per year at the reference price.

        demand_step:
            The permanent change of the demand.

        step_time:
            The year of the change of the demand.

        desired_coverage:
            The years of demand producers want the inventory to hold.

        reference_price:
            The price at the desired coverage.

        supply_elasticity:
            The elasticity of the utilization of capacity with respect to the price.

        demand_elasticity:
            The elasticity of consumption with respect to the price, as a positive number.

        expectation_time:
            The time over which producers smooth the price into their expected price.

        construction_delay:
            The average time between an order of capacity and its completion.

        lifetime:
            The average lifetime of capacity.

    Returns:
        system:
            The system, with the inventory `market.inventory` and the capacity `market.capacity` in years, the flows
            `market.production`, `market.consumption`, `market.completions`, and `market.retirements`. The observables
            of the system are `market.coverage`, `market.price`, `market.expected_price`, and `market.capacity_orders`.

    Examples:
        >>> from skcomplex.systems.economics.markets import commodity_cycle
        >>> commodity_cycle(construction_delay=5.0).elements[0].construction_delay
        5.0
    """
    market = CommodityMarketElement(
        'market',
        inventory=inventory,
        capacity=capacity,
        demand=demand,
        demand_step=demand_step,
        step_time=step_time,
        desired_coverage=desired_coverage,
        reference_price=reference_price,
        supply_elasticity=supply_elasticity,
        demand_elasticity=demand_elasticity,
        expectation_time=expectation_time,
        construction_delay=construction_delay,
        lifetime=lifetime,
    )
    commodity = TransferInteraction(
        'commodity',
        source='market',
        target='market',
        state=Source('producers'),
        into='inventory',
        amount=_production,
        flow='production',
        legs=[('inventory', Sink('consumers'), _consumption, 'consumption')],
    )
    capacity_flows = TransferInteraction(
        'capacity_flows',
        source='market',
        target='market',
        state=Source('construction'),
        into='capacity',
        amount=_completions,
        flow='completions',
        legs=[('capacity', Sink('retirement'), lambda s: s.source.capacity / s.source.lifetime, 'retirements')],
    )
    return System(elements=[market], interactions=[commodity, capacity_flows])

contract_curve(chocolate, first_weight, second_weight, total_chocolate, total_cigarettes)

Return the cigarettes of the first of two barterers on the contract curve, given its chocolate.

On the contract curve of two Cobb-Douglas barterers the marginal rates of substitution are equal, a y / ((1 - a) x) = b (Y - y) / ((1 - b) (X - x)), where a and b are the weights of chocolate, x and y the chocolate and cigarettes of the first barterer, and X and Y the totals.

Parameters:

Name	Type	Description	Default
chocolate	ArrayLike	The chocolate of the first barterer.	required
first_weight	float	The weight of chocolate of the first barterer.	required
second_weight	float	The weight of chocolate of the second barterer.	required
total_chocolate	float	The total chocolate of both.	required
total_cigarettes	float	The total cigarettes of both.	required

Returns:

Name	Type	Description
cigarettes	NDArray[float64]	The cigarettes of the first barterer on the contract curve.

Examples:

>>> from skcomplex.systems.economics.markets import contract_curve
>>> float(contract_curve(5.0, 0.5, 0.5, 10.0, 10.0))
5.0

Source code in src/skcomplex/systems/economics/markets/_barter.py

def contract_curve(
    chocolate: npt.ArrayLike,
    first_weight: float,
    second_weight: float,
    total_chocolate: float,
    total_cigarettes: float,
) -> npt.NDArray[np.float64]:
    """Return the cigarettes of the first of two barterers on the contract curve, given its chocolate.

    On the contract curve of two Cobb-Douglas barterers the marginal rates of substitution are equal,
    `a y / ((1 - a) x) = b (Y - y) / ((1 - b) (X - x))`, where `a` and `b` are the weights of chocolate, `x` and `y`
    the chocolate and cigarettes of the first barterer, and `X` and `Y` the totals.

    Args:
        chocolate:
            The chocolate of the first barterer.

        first_weight:
            The weight of chocolate of the first barterer.

        second_weight:
            The weight of chocolate of the second barterer.

        total_chocolate:
            The total chocolate of both.

        total_cigarettes:
            The total cigarettes of both.

    Returns:
        cigarettes:
            The cigarettes of the first barterer on the contract curve.

    Examples:
        >>> from skcomplex.systems.economics.markets import contract_curve
        >>> float(contract_curve(5.0, 0.5, 0.5, 10.0, 10.0))
        5.0
    """
    x = np.asarray(chocolate, dtype=float)
    a, b = first_weight, second_weight
    return b * (1 - a) * x * total_cigarettes / (a * (1 - b) * (total_chocolate - x) + b * (1 - a) * x)

cournot_duopoly(outputs=(100.0, 100.0), costs=(0.0, 0.0), *, a=1500.0, b=10.0, error=0.0)

Return two Cournot duopolists that adjust their output to their beliefs about each other.

Without errors the outputs converge to the Nash equilibrium, q_i = (a - 2 c_i + c_j) / (3 b), which is 50 units each for the defaults. Errors in the beliefs keep the total output fluctuating around it.

Parameters:

Name	Type	Description	Default
outputs	Sequence[float]	The initial outputs of the two firms.	(100.0, 100.0)
costs	Sequence[float]	The marginal costs of the two firms.	(0.0, 0.0)
a	float	The intercept of the inverse demand.	1500.0
b	float	The slope of the inverse demand.	10.0
error	float	The largest relative error of a firm's belief about the rival's output.	0.0

Returns:

Name	Type	Description
system	System	The system, with the group firms, the total output total, and the price price.

Examples:

>>> from skcomplex.systems.economics.markets import cournot_duopoly
>>> cournot_duopoly(error=0.01).interactions[0].error
0.01

Source code in src/skcomplex/systems/economics/markets/_duopoly.py

@register_system(
    field='economics/markets',
    tags=('market', 'oligopoly', 'agents'),
    references=('Hamill and Gilbert 2016', 'Cournot 1838'),
    defaults={'until': 30},
    observables=lambda system: [
        ObservableSpec('total', group='firms', reduce='sum', variable='output'),
        ObservableSpec('price', _price),
    ],
)
def cournot_duopoly(
    outputs: Sequence[float] = (100.0, 100.0),
    costs: Sequence[float] = (0.0, 0.0),
    *,
    a: float = 1500.0,
    b: float = 10.0,
    error: float = 0.0,
) -> System:
    """Return two Cournot duopolists that adjust their output to their beliefs about each other.

    Without errors the outputs converge to the Nash equilibrium, `q_i = (a - 2 c_i + c_j) / (3 b)`, which is 50 units
    each for the defaults. Errors in the beliefs keep the total output fluctuating around it.

    Args:
        outputs:
            The initial outputs of the two firms.

        costs:
            The marginal costs of the two firms.

        a:
            The intercept of the inverse demand.

        b:
            The slope of the inverse demand.

        error:
            The largest relative error of a firm's belief about the rival's output.

    Returns:
        system:
            The system, with the group `firms`, the total output `total`, and the price `price`.

    Examples:
        >>> from skcomplex.systems.economics.markets import cournot_duopoly
        >>> cournot_duopoly(error=0.01).interactions[0].error
        0.01
    """
    firms = GroupElement(
        'firms',
        element=DuopolistElement,
        size=N_FIRMS,
        values={'output': list(outputs), 'cost': list(costs)},
    )
    return System(
        elements=[firms],
        interactions=[CournotAdjustmentInteraction('market', target='firms', a=a, b=b, error=error)],
        time=DiscreteTime(),
    )

double_auction(n_buyers=50, n_sellers=50, *, protocol='order_book', rounds=20, constrained=True, ceiling=200.0, random_state=0)

Return Gode and Sunder's double auction of zero-intelligence traders, a step per trading period.

Every buyer values one unit and every seller can produce one unit at a cost, both drawn uniformly below the price ceiling. During a trading period of several rounds, every trader that has not traded yet quotes a random price and the market matches the quotes, continuously in a limit order book or at one uniform price in a call auction. With the budget constraint (ZI-C) no trader accepts a loss, and the market alone reaches nearly the whole competitive surplus. Without it (ZI-U), traders make losing trades and the efficiency drops.

The traders are the zero-intelligence traders of Gode and Sunder (1993). The double auction of Gooding (2019), whose traders adapt their bids and asks, is a different model.

Parameters:

Name	Type	Description	Default
n_buyers	int	The number of buyers.	50
n_sellers	int	The number of sellers.	50
protocol	str	The protocol of the market, 'order_book' or 'call_auction'.	'order_book'
rounds	int	The number of quoting rounds of a trading period.	20
constrained	bool	Whether traders never quote at a loss.	True
ceiling	float	The highest price of the market.	200.0
random_state	int | Generator | None	The seed or generator of the values and the costs.	0

Returns:

Name	Type	Description
system	System	The system, with the groups buyers and sellers, the market auction, and the realised ledger.surplus and ledger.efficiency of every trading period.

Raises:

Type	Description
ValueError	The protocol is unknown.

Examples:

>>> from skcomplex.systems.economics.markets import double_auction
>>> [group.size for group in double_auction(10, 8).elements]
[10, 8]

Source code in src/skcomplex/systems/economics/markets/_double_auction.py

@register_system(
    field='economics/markets',
    tags=('market', 'agents', 'double auction', 'zero intelligence'),
    references=('Gode and Sunder 1993',),
    defaults={'until': 20},
)
def double_auction(
    n_buyers: int = 50,
    n_sellers: int = 50,
    *,
    protocol: str = 'order_book',
    rounds: int = 20,
    constrained: bool = True,
    ceiling: float = 200.0,
    random_state: int | np.random.Generator | None = 0,
) -> System:
    """Return Gode and Sunder's double auction of zero-intelligence traders, a step per trading period.

    Every buyer values one unit and every seller can produce one unit at a cost, both drawn uniformly below the price
    ceiling. During a trading period of several rounds, every trader that has not traded yet quotes a random price and
    the market matches the quotes, continuously in a limit order book or at one uniform price in a call auction. With
    the budget constraint (ZI-C) no trader accepts a loss, and the market alone reaches nearly the whole competitive
    surplus. Without it (ZI-U), traders make losing trades and the efficiency drops.

    The traders are the zero-intelligence traders of Gode and Sunder (1993). The double auction of Gooding (2019),
    whose traders adapt their bids and asks, is a different model.

    Args:
        n_buyers:
            The number of buyers.

        n_sellers:
            The number of sellers.

        protocol:
            The protocol of the market, `'order_book'` or `'call_auction'`.

        rounds:
            The number of quoting rounds of a trading period.

        constrained:
            Whether traders never quote at a loss.

        ceiling:
            The highest price of the market.

        random_state:
            The seed or generator of the values and the costs.

    Returns:
        system:
            The system, with the groups `buyers` and `sellers`, the market `auction`, and the realised `ledger.surplus`
            and `ledger.efficiency` of every trading period.

    Raises:
        ValueError:
            The protocol is unknown.

    Examples:
        >>> from skcomplex.systems.economics.markets import double_auction
        >>> [group.size for group in double_auction(10, 8).elements]
        [10, 8]
    """
    if protocol not in PROTOCOLS:
        error_msg = f'Unknown protocol {protocol!r}. Use one of {PROTOCOLS}.'
        raise ValueError(error_msg)
    rng = np.random.default_rng(random_state)
    values = rng.uniform(0.25 * ceiling, ceiling, n_buyers)
    costs = rng.uniform(0.0, 0.75 * ceiling, n_sellers)
    limit = float(constrained)
    buyers = GroupElement(
        'buyers',
        element=ZeroIntelligenceBuyerElement,
        size=n_buyers,
        values={'value': values, 'ceiling': ceiling, 'constrained': limit},
    )
    sellers = GroupElement(
        'sellers',
        element=ZeroIntelligenceSellerElement,
        size=n_sellers,
        values={'value': costs, 'ceiling': ceiling, 'constrained': limit},
    )
    market = MatchingInteraction(
        'auction',
        requesters='buyers',
        providers='sellers',
        rule=ContinuousMatchingSpec() if protocol == 'order_book' else BatchMatchingSpec(),
        limit=('bid', 'ask'),
        quantity=('demand', 'supply'),
        phase='trade',
        on_match=_book_profit,
    )
    quotes = [
        MapInteraction(label, target=group, state=state, function=function, form='agent', phase='trade')
        for label, group, state, function in (
            ('bids', 'buyers', 'bid', ZeroIntelligenceBuyerElement.get_bid),
            ('asks', 'sellers', 'ask', ZeroIntelligenceSellerElement.get_ask),
        )
    ]
    return System(
        elements=[buyers, sellers],
        interactions=[*quotes, market, AuctionLedgerInteraction('ledger', phase='close')],
        time=DiscreteTime(schedule=PhasesScheduleSpec([PhaseSpec('trade', repeat=rounds), PhaseSpec('close')])),
    )

fair_food(width=20, height=20, *, premium=0.3, ethics_weight=2.0, price_weight=5.0, social_weight=2.0, intensity=1.0, low_budget=1.0, high_budget=3.0, random_state=0)

Return the choice between fair-trade and regular food of consumers who influence their neighbours.

Every consumer, of uniform ethical concern and budget, sits on a cell of a grid and every step buys fair or regular food, all at once from the choices of their neighbours in the last step. The share of fair food rises with the weight of ethics and falls with the premium, and social influence raises it.

The choice is the binary logit choice with social interactions of Brock and Durlauf (2001), with a private utility of ethics and price and a social utility of the share of neighbours who buy fair food. Neither Ajani's Choice nor the Fair Food Program of Chesney (2021) is this model.

Parameters:

Name	Type	Description	Default
width	int	The number of columns of the grid.	20
height	int	The number of rows of the grid.	20
premium	float	The extra price of fair food over regular food, which costs one.	0.3
ethics_weight	float	The weight of ethical concern in the utility of fair food.	2.0
price_weight	float	The weight of the premium relative to the budget.	5.0
social_weight	float	The weight of the share of neighbours who buy fair food.	2.0
intensity	float	The intensity of the logit choice.	1.0
low_budget	float	The lowest budget.	1.0
high_budget	float	The highest budget.	3.0
random_state	int | Generator | None	The seed or generator of the ethics and the budgets.	0

Returns:

Name	Type	Description
system	System	The system, with the group consumers, their logit choice shopping, and the share of fair food fair_share.

Examples:

>>> from skcomplex.systems.economics.markets import fair_food
>>> fair_food(5, 4).elements[0].prepare().size_
20

Source code in src/skcomplex/systems/economics/markets/_fair_food.py

@register_system(
    field='economics/markets',
    tags=('consumer choice', 'social influence', 'agents', 'grid'),
    references=('Brock and Durlauf 2001',),
    defaults={'until': 50},
    observables=lambda system: [ObservableSpec('fair_share', group='consumers', variable='fair')],
)
def fair_food(
    width: int = 20,
    height: int = 20,
    *,
    premium: float = 0.3,
    ethics_weight: float = 2.0,
    price_weight: float = 5.0,
    social_weight: float = 2.0,
    intensity: float = 1.0,
    low_budget: float = 1.0,
    high_budget: float = 3.0,
    random_state: int | np.random.Generator | None = 0,
) -> System:
    """Return the choice between fair-trade and regular food of consumers who influence their neighbours.

    Every consumer, of uniform ethical concern and budget, sits on a cell of a grid and every step buys fair or
    regular food, all at once from the choices of their neighbours in the last step. The share of fair food rises with
    the weight of ethics and falls with the premium, and social influence raises it.

    The choice is the binary logit choice with social interactions of Brock and Durlauf (2001), with a private utility
    of ethics and price and a social utility of the share of neighbours who buy fair food. Neither Ajani's Choice nor
    the Fair Food Program of Chesney (2021) is this model.

    Args:
        width:
            The number of columns of the grid.

        height:
            The number of rows of the grid.

        premium:
            The extra price of fair food over regular food, which costs one.

        ethics_weight:
            The weight of ethical concern in the utility of fair food.

        price_weight:
            The weight of the premium relative to the budget.

        social_weight:
            The weight of the share of neighbours who buy fair food.

        intensity:
            The intensity of the logit choice.

        low_budget:
            The lowest budget.

        high_budget:
            The highest budget.

        random_state:
            The seed or generator of the ethics and the budgets.

    Returns:
        system:
            The system, with the group `consumers`, their logit choice `shopping`, and the share of fair food
            `fair_share`.

    Examples:
        >>> from skcomplex.systems.economics.markets import fair_food
        >>> fair_food(5, 4).elements[0].prepare().size_
        20
    """
    rng = np.random.default_rng(random_state)
    size = width * height
    consumers = GroupElement(
        'consumers',
        element=ConsumerElement,
        size=size,
        values={
            'ethics': rng.random(size),
            'budget': rng.uniform(low_budget, high_budget, size),
            'premium': premium,
            'ethics_weight': ethics_weight,
            'price_weight': price_weight,
            'social_weight': social_weight,
        },
        location=[(x, y) for x in range(width) for y in range(height)],
    )
    choice = ChoiceInteraction(
        'shopping',
        target='consumers',
        state='fair',
        options=(0, 1),
        utility=_get_utility,
        intensity=intensity,
        form='agent',
    )
    return System(
        elements=[consumers],
        interactions=[choice],
        space=LatticeSpace((width, height), boundary='periodic', capacity=1),
        time=DiscreteTime(),
    )

fair_food_program(n_shops=30, n_farms=20, *, acres=1000.0, pressure=0.001, random_state=0)

Return Chesney's Fair Food Program, in which pressure on tomato buyers draws farms into fair wages.

The Coalition of Immokalee Workers presses tomato buyers to join the Fair Food Program, whose member farms pay pickers a penny more per pound. Buyers join at level 1, buying from member farms first, then move to level 2, buying only from member farms, once the member farms grow enough, and farms join when the buyers outside the programme need less than they grow. Buyers and farms have random sizes that share the acres of farmland, and the buyers need as much as the farms grow. This is the model of Chesney (2021, Sections 7.2 to 7.6), with the floor and ceiling of its NetLogo code, whose sell_me subtracts the last tomatoes of a farm from the buyer instead of adding them, which this system corrects. Farms hold out until most buyers have joined, then join in a cascade, a tipping point that comes later the more unequal the agents.

Parameters:

Name	Type	Description	Default
n_shops	int	The number of buyers.	30
n_farms	int	The number of farms.	20
acres	float	The acres of farmland, which the buyers need and the farms grow.	1000.0
pressure	float	The probability per period that a buyer outside the programme joins it at level 1.	0.001
random_state	int | Generator | None	The seed or generator of the sizes of the buyers and farms.	0

Returns:

Name	Type	Description
system	System	The system, with the groups shops and farms and the market market, whose reporters are the buyers at level 1 market.committed_shops and level 2 market.member_shops, the member farms market.member_farms, the unfulfilled needs market.unfulfilled, and the unsold tomatoes market.unsold.

Examples:

>>> from skcomplex.systems.economics.markets import fair_food_program
>>> [group.prepare().size_ for group in fair_food_program(5, 3).elements]
[5, 3]

Source code in src/skcomplex/systems/economics/markets/_fair_food_program.py

@register_system(
    field='economics/markets',
    tags=('fair trade', 'supply chain', 'agents', 'tipping point', 'labour standards', 'modern slavery'),
    references=('Chesney 2021',),
    defaults={'until': 5000},
)
def fair_food_program(
    n_shops: int = 30,
    n_farms: int = 20,
    *,
    acres: float = 1000.0,
    pressure: float = 0.001,
    random_state: int | np.random.Generator | None = 0,
) -> System:
    """Return Chesney's Fair Food Program, in which pressure on tomato buyers draws farms into fair wages.

    The Coalition of Immokalee Workers presses tomato buyers to join the Fair Food Program, whose member farms pay
    pickers a penny more per pound. Buyers join at level 1, buying from member farms first, then move to level 2,
    buying only from member farms, once the member farms grow enough, and farms join when the buyers outside the
    programme need less than they grow. Buyers and farms have random sizes that share the acres of farmland, and the
    buyers need as much as the farms grow. This is the model of Chesney (2021, Sections 7.2 to 7.6), with the `floor`
    and `ceiling` of its NetLogo code, whose `sell_me` subtracts the last tomatoes of a farm from the buyer instead of
    adding them, which this system corrects. Farms hold out until most buyers have joined, then join in a cascade, a
    tipping point that comes later the more unequal the agents.

    Args:
        n_shops:
            The number of buyers.

        n_farms:
            The number of farms.

        acres:
            The acres of farmland, which the buyers need and the farms grow.

        pressure:
            The probability per period that a buyer outside the programme joins it at level 1.

        random_state:
            The seed or generator of the sizes of the buyers and farms.

    Returns:
        system:
            The system, with the groups `shops` and `farms` and the market `market`, whose reporters are the buyers at
            level 1 `market.committed_shops` and level 2 `market.member_shops`, the member farms `market.member_farms`,
            the unfulfilled needs `market.unfulfilled`, and the unsold tomatoes `market.unsold`.

    Examples:
        >>> from skcomplex.systems.economics.markets import fair_food_program
        >>> [group.prepare().size_ for group in fair_food_program(5, 3).elements]
        [5, 3]
    """
    rng = np.random.default_rng(random_state)
    _, needs = _sizes(n_shops, acres, rng)
    _, grown = _sizes(n_farms, acres, rng)
    shops = GroupElement('shops', element=ProgramMemberElement, size=n_shops, values={'acres': np.floor(needs)})
    farms = GroupElement('farms', element=ProgramMemberElement, size=n_farms, values={'acres': np.ceil(grown)})
    market = FairFoodMarketInteraction('market', shops='shops', farms='farms', pressure=pressure)
    return System(elements=[shops, farms], interactions=[market])

gooding_double_auction(n_buyers=100, n_sellers=100, *, value=120.0, cost=20.0, spread=0.1, units=5.0, step=1.0, unlimited=False, random_state=0)

Return Gooding's double auction, whose buyers and sellers move their quotes towards trade and then apart.

Buyers value a unit near value and sellers can supply it near cost, each differing by up to spread of it. As in the experiments the model replicates, buyers start bidding near the sellers' cost and sellers start asking near the buyers' value, and both move their quotes towards one another until they trade, after which each tries to push the price its way (Gooding 2019, Section 11.3 and the appendix). In the short run, with sellers of a few units and buyers that are satiated after a few, the price settles near the competitive price halfway between value and cost, 70 at the defaults. With unlimited supply and demand, the long-run price wanders between the cost and the value and the equilibrium disappears. The source describes the rules in prose, so the meetings and the fixed steps of the quotes are a reconstruction of them.

Parameters:

Name	Type	Description	Default
n_buyers	int	The number of buyers.	100
n_sellers	int	The number of sellers.	100
value	float	The mean value of a unit to the buyers.	120.0
cost	float	The mean cost of a unit to the sellers.	20.0
spread	float	The largest relative difference of a trader's value or cost from the mean.	0.1
units	float	The units every seller holds and every buyer wants.	5.0
step	float	The change of a quote after a meeting.	1.0
unlimited	bool	Whether the units of the traders never run out.	False
random_state	int | Generator | None	The seed or generator of the values and costs.	0

Returns:

Name	Type	Description
system	System	The system, with the groups buyers and sellers and the auction auction, whose reporters are the last traded price auction.price, the units traded auction.volume, and the mean quotes auction.mean_bid and auction.mean_ask.

Examples:

>>> from skcomplex.systems.economics.markets import gooding_double_auction
>>> [group.prepare().size_ for group in gooding_double_auction(4, 3).elements]
[4, 3]

Source code in src/skcomplex/systems/economics/markets/_adaptive_auction.py

@register_system(
    field='economics/markets',
    tags=('double auction', 'agents', 'price discovery', 'equilibrium', 'adaptive quotes'),
    references=('Gooding 2019',),
    defaults={'until': 300},
)
def gooding_double_auction(
    n_buyers: int = 100,
    n_sellers: int = 100,
    *,
    value: float = 120.0,
    cost: float = 20.0,
    spread: float = 0.1,
    units: float = 5.0,
    step: float = 1.0,
    unlimited: bool = False,
    random_state: int | np.random.Generator | None = 0,
) -> System:
    """Return Gooding's double auction, whose buyers and sellers move their quotes towards trade and then apart.

    Buyers value a unit near `value` and sellers can supply it near `cost`, each differing by up to `spread` of it. As
    in the experiments the model replicates, buyers start bidding near the sellers' cost and sellers start asking near
    the buyers' value, and both move their quotes towards one another until they trade, after which each tries to push
    the price its way (Gooding 2019, Section 11.3 and the appendix). In the short run, with sellers of a few units and
    buyers that are satiated after a few, the price settles near the competitive price halfway between value and cost,
    70 at the defaults. With unlimited supply and demand, the long-run price wanders between the cost and the value and
    the equilibrium disappears. The source describes the rules in prose, so the meetings and the fixed steps of the
    quotes are a reconstruction of them.

    Args:
        n_buyers:
            The number of buyers.

        n_sellers:
            The number of sellers.

        value:
            The mean value of a unit to the buyers.

        cost:
            The mean cost of a unit to the sellers.

        spread:
            The largest relative difference of a trader's value or cost from the mean.

        units:
            The units every seller holds and every buyer wants.

        step:
            The change of a quote after a meeting.

        unlimited:
            Whether the units of the traders never run out.

        random_state:
            The seed or generator of the values and costs.

    Returns:
        system:
            The system, with the groups `buyers` and `sellers` and the auction `auction`, whose reporters are the last
            traded price `auction.price`, the units traded `auction.volume`, and the mean quotes `auction.mean_bid` and
            `auction.mean_ask`.

    Examples:
        >>> from skcomplex.systems.economics.markets import gooding_double_auction
        >>> [group.prepare().size_ for group in gooding_double_auction(4, 3).elements]
        [4, 3]
    """
    rng = np.random.default_rng(random_state)
    values = value * (1 + spread * rng.uniform(-1, 1, n_buyers))
    costs = cost * (1 + spread * rng.uniform(-1, 1, n_sellers))
    buyers = GroupElement(
        'buyers',
        element=AdaptiveTraderElement,
        size=n_buyers,
        values={'value': values, 'quote': cost * (1 + spread * rng.uniform(-1, 1, n_buyers)), 'units': units},
    )
    sellers = GroupElement(
        'sellers',
        element=AdaptiveTraderElement,
        size=n_sellers,
        values={'value': costs, 'quote': value * (1 + spread * rng.uniform(-1, 1, n_sellers)), 'units': units},
    )
    auction = AdaptiveAuctionInteraction('auction', buyers='buyers', sellers='sellers', step=step, unlimited=unlimited)
    return System(elements=[buyers, sellers], interactions=[auction])

goods_traders(n_traders=200, *, goods=3, width=20, height=20, quantity=0.1, mobility=0.5, low=1.0, high=20.0, random_state=0)

Return traders who wander a grid and barter two or three goods with their neighbours.

Every trader holds random amounts of the goods and values them with random Cobb-Douglas weights. Every step the traders wander to a random neighbouring cell, then every trader is paired with a neighbour, and the pairs barter while both gain, at the geometric mean of their marginal rates of substitution. Trades only move goods between the traders, so the total of every good is unchanged, while the mean utility rises and the prices of different trades converge, so the dispersion of the logarithm of the price falls.

The trade rule is the bilateral barter of sugar and spice of Epstein and Axtell (1996), and salt extends it to a third good. The Toy Trader of Gooding (2019), with money, posted prices, and imitation of price rules, is a different model.

Parameters:

Name	Type	Description	Default
n_traders	int	The number of traders.	200
goods	int	The number of goods, 2 for sugar and spice or 3 with salt.	3
width	int	The number of columns of the grid.	20
height	int	The number of rows of the grid.	20
quantity	float	The quantity of the bought good in every trade.	0.1
mobility	float	The probability that a trader moves in a step.	0.5
low	float	The smallest initial holding of a good.	1.0
high	float	The largest initial holding of a good.	20.0
random_state	int | Generator | None	The seed or generator of the cells, holdings, and weights.	0

Returns:

Name	Type	Description
system	System	The system, with the group traders, their random walk wander, and the exchange exchange, whose reporters are exchange.price, exchange.dispersion, and exchange.volume.

Raises:

Type	Description
ValueError	The number of goods is not 2 or 3.

Examples:

>>> from skcomplex.systems.economics.markets import goods_traders
>>> goods_traders(20, goods=2).elements[0].prepare().size_
20

Source code in src/skcomplex/systems/economics/markets/_goods_traders.py

@register_system(
    field='economics/markets',
    tags=('market', 'agents', 'barter', 'price convergence', 'grid'),
    references=('Epstein and Axtell 1996',),
    defaults={'until': 50, 'invariants': dict.fromkeys(GOODS, 1e-6)},
    observables=_goods_observables,
)
def goods_traders(
    n_traders: int = 200,
    *,
    goods: int = 3,
    width: int = 20,
    height: int = 20,
    quantity: float = 0.1,
    mobility: float = 0.5,
    low: float = 1.0,
    high: float = 20.0,
    random_state: int | np.random.Generator | None = 0,
) -> System:
    """Return traders who wander a grid and barter two or three goods with their neighbours.

    Every trader holds random amounts of the goods and values them with random Cobb-Douglas weights. Every step the
    traders wander to a random neighbouring cell, then every trader is paired with a neighbour, and the pairs barter
    while both gain, at the geometric mean of their marginal rates of substitution. Trades only move goods between the
    traders, so the total of every good is unchanged, while the mean utility rises and the prices of different trades
    converge, so the dispersion of the logarithm of the price falls.

    The trade rule is the bilateral barter of sugar and spice of Epstein and Axtell (1996), and salt extends it to a
    third good. The Toy Trader of Gooding (2019), with money, posted prices, and imitation of price rules, is a
    different model.

    Args:
        n_traders:
            The number of traders.

        goods:
            The number of goods, 2 for sugar and spice or 3 with salt.

        width:
            The number of columns of the grid.

        height:
            The number of rows of the grid.

        quantity:
            The quantity of the bought good in every trade.

        mobility:
            The probability that a trader moves in a step.

        low:
            The smallest initial holding of a good.

        high:
            The largest initial holding of a good.

        random_state:
            The seed or generator of the cells, holdings, and weights.

    Returns:
        system:
            The system, with the group `traders`, their random walk `wander`, and the exchange `exchange`, whose
            reporters are `exchange.price`, `exchange.dispersion`, and `exchange.volume`.

    Raises:
        ValueError:
            The number of goods is not 2 or 3.

    Examples:
        >>> from skcomplex.systems.economics.markets import goods_traders
        >>> goods_traders(20, goods=2).elements[0].prepare().size_
        20
    """
    if goods not in (2, 3):
        error_msg = f'The traders barter 2 or 3 goods, but got {goods}.'
        raise ValueError(error_msg)
    rng = np.random.default_rng(random_state)
    cells = [(x, y) for x in range(width) for y in range(height)]
    locations = [cells[index] for index in rng.integers(len(cells), size=n_traders)]
    weights = rng.dirichlet(np.ones(goods), size=n_traders)
    holdings = rng.uniform(low, high, (n_traders, goods))
    values: dict[str, Array] = {GOODS[index]: holdings[:, index] for index in range(goods)}
    values |= {WEIGHTS[index]: weights[:, index] for index in range(goods)}
    traders = GroupElement(
        'traders',
        element=GoodsTraderElement,
        size=n_traders,
        values=values,
        variable=True,
        location=locations,
    )
    return System(
        elements=[traders],
        interactions=[
            MovementInteraction('wander', target='traders', probability=mobility, phase='move'),
            ExchangeInteraction(
                'exchange',
                source='traders',
                target='traders',
                goods=GOODS[:goods],
                pairing='neighbor-pairs',
                quantity=quantity,
            ),
        ],
        space=LatticeSpace((width, height), boundary='periodic', neighborhood='moore'),
        time=DiscreteTime(schedule=PhasesScheduleSpec([PhaseSpec('move'), PhaseSpec('trade')])),
    )

hire(trade, system)

Record the employer, the wage, and the steps of the hire and of its offer on the hired worker.

The hired worker keeps its search flag supply until the next step, which the market lowered when it cleared, so the observable searching counts every worker who searched in the step.

Parameters:

Name	Type	Description	Default
trade	MatchSpec	The match of a firm, the requester of labour, with a worker, its provider.	required
system	SystemView	The system view of the market.	required

Source code in src/skcomplex/systems/economics/markets/_labour.py

def hire(trade: MatchSpec, system: SystemView) -> None:
    """Record the employer, the wage, and the steps of the hire and of its offer on the hired worker.

    The hired worker keeps its search flag `supply` until the next step, which the market lowered when it cleared, so
    the observable `searching` counts every worker who searched in the step.

    Args:
        trade:
            The match of a firm, the requester of labour, with a worker, its provider.

        system:
            The system view of the market.
    """
    worker, firm = trade.provider, trade.requester
    if not isinstance(worker, JobseekerView) or not isinstance(firm, EmployerView):
        return
    worker.employer, worker.wage, worker.duration = float(firm.id), trade.value, 0.0
    worker.supply += trade.quantity
    worker.hired, worker.offer = system.time, firm.posted

household_demand(n_households=1000, *, demand='cobb-douglas', shares=FOOD_SHARES, share_deviation=0.02, elasticities=FOOD_ELASTICITIES, elasticity_deviation=0.1, minimum=100 / 3, change=0.1, random_state=0)

Return Hamill and Gilbert's heterogeneous households, whose demand for food traces a curve as its price rises.

Every household has a budget drawn from an exponential law shifted right by a minimum and normalized to a mean of 100, which gives a Gini coefficient of about a third and a P90/P10 ratio just over 4, and a budget share of food drawn around the mean of its budget quintile, from 15 percent for the poorest to 8 percent for the richest. Every step the price rises by the share change, from 1. With the Cobb-Douglas utility of Section 3.2 of Hamill and Gilbert (2016) the demand is alpha m / p, the own-price elasticity is -1, the expenditure stays the same, and the poorest lose the most utility. With the practical model of Section 3.3 the households also have own-price elasticities drawn around the means of their quintiles, from -0.6 to -0.3, so a price rise of 10 percent lowers the demand by about 4.4 percent and raises the expenditure by about 5.1 percent, more for the richest.

Parameters:

Name	Type	Description	Default
n_households	int	The number of households.	1000
demand	str	'cobb-douglas' for the utility model, 'elasticity' for the practical model.	'cobb-douglas'
shares	Sequence[float]	The mean budget share of the good of every budget quintile, from the poorest.	FOOD_SHARES
share_deviation	float	The standard deviation of the budget shares within a quintile.	0.02
elasticities	Sequence[float]	The mean own-price elasticity of every budget quintile, from the poorest.	FOOD_ELASTICITIES
elasticity_deviation	float	The standard deviation of the elasticities within a quintile.	0.1
minimum	float	The shift of the exponential law of the budgets before normalization.	100 / 3
change	float	The relative rise of the price every step.	0.1
random_state	int | Generator | None	The seed or generator of the budgets, shares, and elasticities.	0

Returns:

Name	Type	Description
system	System	The system, with the group households of the demands households.demand, the expenditures households.expenditure, the utilities households.utility, and the prices households.price. The observables of the system are total_demand, total_expenditure, and mean_utility.

Examples:

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

Source code in src/skcomplex/systems/economics/markets/_household_demand.py

@register_system(
    field='economics/markets',
    tags=('demand', 'heterogeneity', 'agents', 'price elasticity', 'income distribution', 'discrete map'),
    references=('Hamill and Gilbert 2016',),
    defaults={'until': 10},
    observables=lambda system: [
        ObservableSpec('total_demand', group='households', variable='demand', reduce='sum'),
        ObservableSpec('total_expenditure', group='households', variable='expenditure', reduce='sum'),
        ObservableSpec('mean_utility', group='households', variable='utility', reduce='mean'),
    ],
)
def household_demand(
    n_households: int = 1000,
    *,
    demand: str = 'cobb-douglas',
    shares: Sequence[float] = FOOD_SHARES,
    share_deviation: float = 0.02,
    elasticities: Sequence[float] = FOOD_ELASTICITIES,
    elasticity_deviation: float = 0.1,
    minimum: float = 100 / 3,
    change: float = 0.1,
    random_state: int | np.random.Generator | None = 0,
) -> System:
    """Return Hamill and Gilbert's heterogeneous households, whose demand for food traces a curve as its price rises.

    Every household has a budget drawn from an exponential law shifted right by a minimum and normalized to a mean of
    100, which gives a Gini coefficient of about a third and a P90/P10 ratio just over 4, and a budget share of food
    drawn around the mean of its budget quintile, from 15 percent for the poorest to 8 percent for the richest. Every
    step the price rises by the share `change`, from 1. With the Cobb-Douglas utility of Section 3.2 of Hamill and
    Gilbert (2016) the demand is `alpha m / p`, the own-price elasticity is -1, the expenditure stays the same, and the
    poorest lose the most utility. With the practical model of Section 3.3 the households also have own-price
    elasticities drawn around the means of their quintiles, from -0.6 to -0.3, so a price rise of 10 percent lowers
    the demand by about 4.4 percent and raises the expenditure by about 5.1 percent, more for the richest.

    Args:
        n_households:
            The number of households.

        demand:
            `'cobb-douglas'` for the utility model, `'elasticity'` for the practical model.

        shares:
            The mean budget share of the good of every budget quintile, from the poorest.

        share_deviation:
            The standard deviation of the budget shares within a quintile.

        elasticities:
            The mean own-price elasticity of every budget quintile, from the poorest.

        elasticity_deviation:
            The standard deviation of the elasticities within a quintile.

        minimum:
            The shift of the exponential law of the budgets before normalization.

        change:
            The relative rise of the price every step.

        random_state:
            The seed or generator of the budgets, shares, and elasticities.

    Returns:
        system:
            The system, with the group `households` of the demands `households.demand`, the expenditures
            `households.expenditure`, the utilities `households.utility`, and the prices `households.price`. The
            observables of the system are `total_demand`, `total_expenditure`, and `mean_utility`.

    Examples:
        >>> from skcomplex.systems.economics.markets import household_demand
        >>> household_demand(10).elements[0].prepare().size_
        10
    """
    rng = np.random.default_rng(random_state)
    budgets = _budgets(n_households, minimum, rng)
    alphas = np.clip(_by_quintile(budgets, shares, share_deviation, rng), 0.0, 1.0)
    slopes = _by_quintile(budgets, elasticities, elasticity_deviation, rng) if demand == 'elasticity' else -1.0
    quantities = alphas * budgets
    households = GroupElement(
        'households',
        element=BudgetHouseholdElement,
        size=n_households,
        values={
            'budget': budgets,
            'alpha': alphas,
            'elasticity': slopes,
            'demand': quantities,
            'expenditure': quantities,
            'utility': quantities**alphas * (budgets - quantities) ** (1 - alphas),
            'change': change,
            'kind': demand,
        },
    )
    return System(elements=[households])

labour_market(n_workers=200, n_firms=40, *, separation=0.05, vacancy_rate=0.3, reservation=1.0, decline=0.1, low_wage=0.5, high_wage=1.0, unemployment=0.5, random_state=0)

Return Hamill and Gilbert's labour market of firms posting wage offers and unemployed workers searching.

In the phase of offers, jobs are destroyed at the separation rate s, every firm posts a vacancy with a uniform wage offer at the vacancy rate, and unemployed workers set reservation wages that fall with the duration of their search. In the phase of matching, a deferred-acceptance market pairs vacancies with workers whose reservation wage they cover, every worker getting the best offer of a stable matching. The unemployment rate settles at the flow-balance value s / (s + f), with f the share of searching workers hired in a step.

Parameters:

Name	Type	Description	Default
n_workers	int	The number of workers.	200
n_firms	int	The number of firms, each with at most one vacancy per step.	40
separation	float	The probability s that a job is destroyed in a step.	0.05
vacancy_rate	float	The probability that a firm posts a vacancy in a step.	0.3
reservation	float	The reservation wage at the start of a search.	1.0
decline	float	The relative fall of the reservation wage per step of search.	0.1
low_wage	float	The lowest wage offer.	0.5
high_wage	float	The highest wage offer.	1.0
unemployment	float	The initial share of unemployed workers.	0.5
random_state	int | Generator | None	The seed or generator of the initial employment.	0

Returns:

Name	Type	Description
system	System	The system, with the groups firms and workers, the market labour with the reporters labour.value (the mean wage of the hires) and labour.volume (the hires), and the catalog observables of the unemployment rate unemployment and of the searching workers searching.

Examples:

>>> from skcomplex.systems.economics.markets import labour_market
>>> [group.size for group in labour_market(20, 5).elements]
[5, 20]

Source code in src/skcomplex/systems/economics/markets/_labour.py

@register_system(
    field='economics/markets',
    tags=('labour', 'market', 'matching', 'agents'),
    references=('Hamill and Gilbert 2016 ch. 7',),
    defaults={'until': 200},
    observables=lambda system: [
        ObservableSpec('unemployment', group='workers', variable='employer', reduce=_unemployment),
        ObservableSpec('searching', group='workers', variable='supply', reduce='sum'),
    ],
)
def labour_market(
    n_workers: int = 200,
    n_firms: int = 40,
    *,
    separation: float = 0.05,
    vacancy_rate: float = 0.3,
    reservation: float = 1.0,
    decline: float = 0.1,
    low_wage: float = 0.5,
    high_wage: float = 1.0,
    unemployment: float = 0.5,
    random_state: int | np.random.Generator | None = 0,
) -> System:
    """Return Hamill and Gilbert's labour market of firms posting wage offers and unemployed workers searching.

    In the phase of offers, jobs are destroyed at the separation rate `s`, every firm posts a vacancy with a uniform
    wage offer at the vacancy rate, and unemployed workers set reservation wages that fall with the duration of their
    search. In the phase of matching, a deferred-acceptance market pairs vacancies with workers whose reservation wage
    they cover, every worker getting the best offer of a stable matching. The unemployment rate settles at the
    flow-balance value `s / (s + f)`, with `f` the share of searching workers hired in a step.

    Args:
        n_workers:
            The number of workers.

        n_firms:
            The number of firms, each with at most one vacancy per step.

        separation:
            The probability `s` that a job is destroyed in a step.

        vacancy_rate:
            The probability that a firm posts a vacancy in a step.

        reservation:
            The reservation wage at the start of a search.

        decline:
            The relative fall of the reservation wage per step of search.

        low_wage:
            The lowest wage offer.

        high_wage:
            The highest wage offer.

        unemployment:
            The initial share of unemployed workers.

        random_state:
            The seed or generator of the initial employment.

    Returns:
        system:
            The system, with the groups `firms` and `workers`, the market `labour` with the reporters `labour.value`
            (the mean wage of the hires) and `labour.volume` (the hires), and the catalog observables of the
            unemployment rate `unemployment` and of the searching workers `searching`.

    Examples:
        >>> from skcomplex.systems.economics.markets import labour_market
        >>> [group.size for group in labour_market(20, 5).elements]
        [5, 20]
    """
    rng = np.random.default_rng(random_state)
    employed = rng.random(n_workers) >= unemployment
    employers = np.where(employed, rng.integers(n_firms, size=n_workers), -1).astype(float)
    firms = GroupElement(
        'firms',
        element=EmployerElement,
        size=n_firms,
        values={'vacancy_rate': vacancy_rate, 'low_wage': low_wage, 'high_wage': high_wage, 'bid': low_wage},
    )
    workers = GroupElement(
        'workers',
        element=JobseekerElement,
        size=n_workers,
        values={
            'employer': employers,
            'wage': np.where(employed, (low_wage + high_wage) / 2, 0.0),
            'supply': 0.0,
            'ask': reservation,
            'separation': separation,
            'reservation': reservation,
            'decline': decline,
        },
    )
    offers = [
        MapInteraction(f'{group}_{name}', target=group, state=name, function=function, form='agent', phase='offers')
        for group, steps in (('firms', EmployerElement.get_posting()), ('workers', JobseekerElement.get_search()))
        for name, function in steps
    ]
    losses = TransitionInteraction(
        'workers_separation',
        target='workers',
        state='employer',
        source=lambda value: value >= 0,
        destination=-1.0,
        probability=lambda agent, system: agent.separation,
        form='agent',
        phase='offers',
    )
    spells = [
        MapInteraction(f'workers_{name}', target='workers', state=name, function=function, form='agent', phase='offers')
        for name, function in JobseekerElement.get_spell()
    ]
    market = MatchingInteraction(
        'labour',
        requesters='firms',
        providers='workers',
        rule=StableMatchingSpec(clearing='requester'),
        limit=('bid', 'ask'),
        quantity=('demand', 'supply'),
        phase='matching',
        on_match=hire,
    )
    return System(
        elements=[firms, workers],
        interactions=[
            *offers,
            losses,
            *spells,
            market,
        ],
        time=DiscreteTime(schedule=PhasesScheduleSpec([PhaseSpec('offers'), PhaseSpec('matching')])),
    )

marginal_rate(agent, good, numeraire, weights)

Return the marginal rate of substitution of a good for a numeraire, the units of numeraire one unit is worth.

Parameters:

Name	Type	Description	Default
agent	object	The agent.	required
good	str	The name of the state that holds the good.	required
numeraire	str	The name of the state that holds the numeraire.	required
weights	tuple[str, str]	The names of the states that hold the weights of the good and of the numeraire.	required

Returns:

Name	Type	Description
rate	float	The ratio of the marginal utilities, (a_good / x_good) / (a_numeraire / x_numeraire).

Examples:

>>> from types import SimpleNamespace
>>> from skcomplex.systems.economics.markets import marginal_rate
>>> agent = SimpleNamespace(x=2.0, y=6.0, a=0.5, b=0.5)
>>> marginal_rate(agent, 'x', 'y', ('a', 'b'))
3.0

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

def marginal_rate(agent: object, good: str, numeraire: str, weights: tuple[str, str]) -> float:
    """Return the marginal rate of substitution of a good for a numeraire, the units of numeraire one unit is worth.

    Args:
        agent:
            The agent.

        good:
            The name of the state that holds the good.

        numeraire:
            The name of the state that holds the numeraire.

        weights:
            The names of the states that hold the weights of the good and of the numeraire.

    Returns:
        rate:
            The ratio of the marginal utilities, `(a_good / x_good) / (a_numeraire / x_numeraire)`.

    Examples:
        >>> from types import SimpleNamespace
        >>> from skcomplex.systems.economics.markets import marginal_rate
        >>> agent = SimpleNamespace(x=2.0, y=6.0, a=0.5, b=0.5)
        >>> marginal_rate(agent, 'x', 'y', ('a', 'b'))
        3.0
    """
    good_weight, numeraire_weight = (float(getattr(agent, name)) for name in weights)
    return (good_weight / float(getattr(agent, good))) / (numeraire_weight / float(getattr(agent, numeraire)))

maximum_surplus(values, costs)

Return the largest total surplus of single units, the competitive surplus of the supply and demand schedules.

Parameters:

Name	Type	Description	Default
values	ndarray	The value of a unit to every buyer.	required
costs	ndarray	The cost of a unit to every seller.	required

Returns:

Name	Type	Description
surplus	float	The sum of the positive differences between the highest values and the lowest costs.

Examples:

>>> import numpy as np
>>> from skcomplex.systems.economics.markets import maximum_surplus
>>> maximum_surplus(np.array([10.0, 6.0, 2.0]), np.array([1.0, 4.0, 8.0]))
11.0

Source code in src/skcomplex/systems/economics/markets/_double_auction.py

def maximum_surplus(values: np.ndarray, costs: np.ndarray) -> float:
    """Return the largest total surplus of single units, the competitive surplus of the supply and demand schedules.

    Args:
        values:
            The value of a unit to every buyer.

        costs:
            The cost of a unit to every seller.

    Returns:
        surplus:
            The sum of the positive differences between the highest values and the lowest costs.

    Examples:
        >>> import numpy as np
        >>> from skcomplex.systems.economics.markets import maximum_surplus
        >>> maximum_surplus(np.array([10.0, 6.0, 2.0]), np.array([1.0, 4.0, 8.0]))
        11.0
    """
    demand, supply = np.sort(values)[::-1], np.sort(costs)
    size = min(len(demand), len(supply))
    return float(np.maximum(demand[:size] - supply[:size], 0).sum())

oligopoly_equilibria(n_firms=2, *, alpha=0.25, n_prices=21, demand=10.0, c_i=1.0, c_j=2.0, lambda_i=0.0, lambda_j=0.0, dynamics='static')

Return the equilibria of a Cournot duopoly and of a pricing game of firms with loyal customers and switchers.

The static system solves the Cournot best responses (Forrest 2018 Eqs. 21.5-21.6) and, on a grid of prices, the symmetric mixed equilibrium of m firms of loyal share alpha and switchers beta = 1 - m alpha, whose price distribution makes a firm indifferent among its prices, at the profit alpha (Lemma 20.1). With dynamics='fictitious-play', two firms learn to price on the grid instead: every round each plays a best response to the prices of its rival so far, and the prices they charge are distributed near the mixed equilibrium.

Parameters:

Name	Type	Description	Default
n_firms	int	The number of firms of the pricing game, two under fictitious play.	2
alpha	float	The loyal share of every firm.	0.25
n_prices	int	The number of prices of the grid from alpha / (alpha + beta) to 1.	21
demand	float	The intercept of the inverse demand of the Cournot duopoly.	10.0
c_i	float	The unit cost of the first Cournot firm.	1.0
c_j	float	The unit cost of the second Cournot firm.	2.0
lambda_i	float	The cross holding of the first Cournot firm.	0.0
lambda_j	float	The cross holding of the second Cournot firm.	0.0
dynamics	str	'static' for the equilibria, or 'fictitious-play' for the learning of two pricing firms.	'static'

Returns:

Name	Type	Description
system	System	The static system, with the states cournot.q_i, cournot.q_j, prices.price, and prices.distribution and the observables lowest_profit and highest_profit, or the learning system, with the group firms and the observables lower_price and higher_price. The catalog lists the observables, so a run passes them in observables.

Raises:

Type	Description
ValidationError	The dynamics are unknown, or fictitious play has other than two firms.

Examples:

>>> from skcomplex.core.engine.runtime import Simulation
>>> from skcomplex.systems.economics.markets import oligopoly_equilibria
>>> trajectory = Simulation(oligopoly_equilibria()).prepare().run(1.0).trajectory_
>>> round(float(trajectory.values('cournot.q_i')[0]), 6)
3.333333

Source code in src/skcomplex/systems/economics/markets/_oligopoly.py

@register_system(
    field='economics/markets',
    tags=('oligopoly', 'equilibrium', 'static', 'game theory'),
    references=('Forrest 2018', 'Brown 1951'),
    observables=_price_observables,
)
def oligopoly_equilibria(
    n_firms: int = 2,
    *,
    alpha: float = 0.25,
    n_prices: int = 21,
    demand: float = 10.0,
    c_i: float = 1.0,
    c_j: float = 2.0,
    lambda_i: float = 0.0,
    lambda_j: float = 0.0,
    dynamics: str = 'static',
) -> System:
    """Return the equilibria of a Cournot duopoly and of a pricing game of firms with loyal customers and switchers.

    The static system solves the Cournot best responses (Forrest 2018 Eqs. 21.5-21.6) and, on a grid of prices, the
    symmetric mixed equilibrium of `m` firms of loyal share `alpha` and switchers `beta = 1 - m alpha`, whose price
    distribution makes a firm indifferent among its prices, at the profit `alpha` (Lemma 20.1). With
    `dynamics='fictitious-play'`, two firms learn to price on the grid instead: every round each plays a best response
    to the prices of its rival so far, and the prices they charge are distributed near the mixed equilibrium.

    Args:
        n_firms:
            The number of firms of the pricing game, two under fictitious play.

        alpha:
            The loyal share of every firm.

        n_prices:
            The number of prices of the grid from `alpha / (alpha + beta)` to 1.

        demand:
            The intercept of the inverse demand of the Cournot duopoly.

        c_i:
            The unit cost of the first Cournot firm.

        c_j:
            The unit cost of the second Cournot firm.

        lambda_i:
            The cross holding of the first Cournot firm.

        lambda_j:
            The cross holding of the second Cournot firm.

        dynamics:
            `'static'` for the equilibria, or `'fictitious-play'` for the learning of two pricing firms.

    Returns:
        system:
            The static system, with the states `cournot.q_i`, `cournot.q_j`, `prices.price`, and `prices.distribution`
            and the observables `lowest_profit` and `highest_profit`, or the learning system, with the group `firms` and
            the observables `lower_price` and `higher_price`. The catalog lists the observables, so a run passes them
            in `observables`.

    Raises:
        ValidationError:
            The dynamics are unknown, or fictitious play has other than two firms.

    Examples:
        >>> from skcomplex.core.engine.runtime import Simulation
        >>> from skcomplex.systems.economics.markets import oligopoly_equilibria
        >>> trajectory = Simulation(oligopoly_equilibria()).prepare().run(1.0).trajectory_
        >>> round(float(trajectory.values('cournot.q_i')[0]), 6)
        3.333333
    """
    if dynamics not in DYNAMICS or (dynamics == 'fictitious-play' and n_firms != DUOPOLY):
        error_msg = (
            f'oligopoly_equilibria has dynamics={dynamics!r} and n_firms={n_firms!r}. Use one of {DYNAMICS}, and '
            f'{DUOPOLY} firms under fictitious play.'
        )
        raise ValidationError(error_msg)
    beta = 1 - n_firms * alpha
    prices = np.linspace(alpha / (alpha + beta), 1.0, n_prices)
    if dynamics == 'fictitious-play':
        return _fictitious_play(prices, alpha, beta)
    points = GroupElement(
        'prices',
        element=PricePointElement,
        size=n_prices,
        values={'price': prices, 'alpha': alpha, 'beta': beta, 'n_firms': float(n_firms)},
    )
    cournot = CournotBestResponseElement(
        'cournot',
        demand=demand,
        c_i=c_i,
        c_j=c_j,
        lambda_i=lambda_i,
        lambda_j=lambda_j,
    )
    return System(elements=[cournot, points], time=StaticTime())

perishable_goods_duopoly(customers=200, *, stock=0.5, T=0.1, alpha=0.9, tau_a=1.0, random_state=0)

Return customers who choose between two shops of perishable goods, preferring the one that sold fresher goods.

Each shop holds stock products per customer, and every sale replaces a product with a fresh one, so the shop with more customers sells fresher goods and attracts more customers. With the control parameter R = 2 stock / tau_a, the symmetric state m = p_1 - p_2 = 0 loses its stability below the temperature T_c = R / (1 + R)^2, and one shop takes most customers. The transition is continuous for R > sqrt(3) - 1 and discontinuous below. One unit of time is N visits by customers drawn with replacement. A visited customer chooses a shop with the probability 1 / (1 + exp(-(S_1 - S_2) / T)) for the first one, moves to it, buys a random product of age tau, of freshness h = exp(-tau / tau_a), which the shop replaces with a fresh one, and updates its satisfaction with that shop alone to alpha S + (1 - alpha) h. The shops are the two nodes of a network on which their products stand.

Parameters:

Name	Type	Description	Default
customers	int	The number of customers N.	200
stock	float	The number of products of each shop per customer.	0.5
T	float	The temperature of the choice, the randomness of the customers.	0.1
alpha	float	The memory of the satisfaction, between 0 and 1.	0.9
tau_a	float	The time over which the products age.	1.0
random_state	int | Generator | None	The seed or generator of the initial satisfactions, which scatter by 0.01 around 0.5.	0

Returns:

Name	Type	Description
system	System	The system, with the groups customers and products, and the asymmetry m of the probabilities of visiting the two shops.

Examples:

>>> from skcomplex.systems.economics.markets import perishable_goods_duopoly
>>> perishable_goods_duopoly(20).elements[0].prepare().size_
20

Source code in src/skcomplex/systems/economics/markets/_perishables.py

@register_system(
    field='economics/markets',
    tags=('agents', 'discrete choice', 'symmetry breaking', 'phase transition', 'retail'),
    references=('Bertin 2021', 'Bouchaud 2013'),
    defaults={'until': 300},
    observables=lambda system: [ObservableSpec('m', group='customers', expression=CustomerElement.preference)],
)
def perishable_goods_duopoly(
    customers: int = 200,
    *,
    stock: float = 0.5,
    T: float = 0.1,
    alpha: float = 0.9,
    tau_a: float = 1.0,
    random_state: int | np.random.Generator | None = 0,
) -> System:
    """Return customers who choose between two shops of perishable goods, preferring the one that sold fresher goods.

    Each shop holds `stock` products per customer, and every sale replaces a product with a fresh one, so the shop
    with more customers sells fresher goods and attracts more customers. With the control parameter
    `R = 2 stock / tau_a`, the symmetric state `m = p_1 - p_2 = 0` loses its stability below the temperature
    `T_c = R / (1 + R)^2`, and one shop takes most customers. The transition is continuous for `R > sqrt(3) - 1` and
    discontinuous below. One unit of time is `N` visits by customers drawn with replacement. A visited customer
    chooses a shop with the probability `1 / (1 + exp(-(S_1 - S_2) / T))` for the first one, moves to it, buys a random
    product of age `tau`, of freshness `h = exp(-tau / tau_a)`, which the shop replaces with a fresh one, and updates
    its satisfaction with that shop alone to `alpha S + (1 - alpha) h`. The shops are the two nodes of a network on
    which their products stand.

    Args:
        customers:
            The number of customers `N`.

        stock:
            The number of products of each shop per customer.

        T:
            The temperature of the choice, the randomness of the customers.

        alpha:
            The memory of the satisfaction, between 0 and 1.

        tau_a:
            The time over which the products age.

        random_state:
            The seed or generator of the initial satisfactions, which scatter by 0.01 around 0.5.

    Returns:
        system:
            The system, with the groups `customers` and `products`, and the asymmetry `m` of the probabilities of
            visiting the two shops.

    Examples:
        >>> from skcomplex.systems.economics.markets import perishable_goods_duopoly
        >>> perishable_goods_duopoly(20).elements[0].prepare().size_
        20
    """
    rng = np.random.default_rng(random_state)
    shelf = max(round(stock * customers), 1)
    satisfaction = 0.5 + 0.01 * rng.standard_normal((N_SHOPS, customers))
    choice = ChoiceInteraction(
        'choice',
        target='customers',
        state='shop',
        options=tuple(float(shop) for shop in range(N_SHOPS)),
        utility=CustomerElement.utility,
        schedule='member',
        form='agent',
        phase=PHASE,
    )
    movement = MovementInteraction(
        'movement',
        target='customers',
        step=CustomerElement.destination,
        schedule='member',
        phase=PHASE,
    )
    purchase = TransferInteraction(
        'purchase',
        source='products',
        target='customers',
        state='shelved',
        into=Sink(),
        amount=ProductElement.restock,
        legs=[(Source(), 'freshness', ProductElement.sell)],
        pairing='cell',
        form='agent',
        schedule='member',
        actor='target',
        partners=1,
        phase=PHASE,
    )
    satisfaction_update = ReinforcementInteraction(
        'satisfaction',
        target='customers',
        scores=('s1', 's2'),
        payoff=CustomerElement.payoff,
        rate=1 - alpha,
        chosen='shop',
        schedule='member',
        form='agent',
        phase=PHASE,
    )
    return System(
        elements=[
            GroupElement(
                'customers',
                element=CustomerElement,
                size=customers,
                values={'s1': satisfaction[0], 's2': satisfaction[1], 'T': T},
            ),
            GroupElement(
                'products',
                element=ProductElement,
                size=N_SHOPS * shelf,
                values={'tau_a': tau_a},
                location=np.repeat(np.arange(N_SHOPS), shelf).tolist(),
            ),
        ],
        interactions=[choice, movement, purchase, satisfaction_update],
        space=NetworkSpace(nx.empty_graph(N_SHOPS)),
        time=DiscreteTime(
            schedule=PhasesScheduleSpec([PhaseSpec(PHASE, activation='visits', groups=('customers',))]),
        ),
    )

produce_market(n_shoppers=50, n_stalls=9, *, n_items_stocked=6, random_state=0)

Return Hamill and Gilbert's produce market, where shoppers with lists of produce visit stalls at random.

There are twelve kinds of fruit and vegetables. Every stall sells a random selection of them at its own markup of 1 to 30 percent on the wholesale prices, and every shopper wants between one and eight of them.

Parameters:

Name	Type	Description	Default
n_shoppers	int	The number of shoppers.	50
n_stalls	int	The number of stalls.	9
n_items_stocked	int	The number of kinds of produce that every stall sells.	6
random_state	int | Generator | None	The seed or generator of the stocks, the markups, and the shopping lists.	0

Returns:

Name	Type	Description
system	System	The system, with the groups stalls and shoppers, the mean number of items left on the lists items_left, and the mean money spent spent.

Examples:

>>> from skcomplex.systems.economics.markets import produce_market
>>> [group.size for group in produce_market(10, 3).elements]
[3, 10]

Source code in src/skcomplex/systems/economics/markets/_produce.py

@register_system(
    field='economics/markets',
    tags=('market', 'agents', 'heterogeneity'),
    references=('Hamill and Gilbert 2016',),
    defaults={'until': 10},
    observables=lambda system: [
        ObservableSpec('items_left', group='shoppers', variable='wanted', reduce=_items_left),
        ObservableSpec('spent', group='shoppers', variable='spent'),
    ],
)
def produce_market(
    n_shoppers: int = 50,
    n_stalls: int = 9,
    *,
    n_items_stocked: int = 6,
    random_state: int | np.random.Generator | None = 0,
) -> System:
    """Return Hamill and Gilbert's produce market, where shoppers with lists of produce visit stalls at random.

    There are twelve kinds of fruit and vegetables. Every stall sells a random selection of them at its own markup of 1
    to 30 percent on the wholesale prices, and every shopper wants between one and eight of them.

    Args:
        n_shoppers:
            The number of shoppers.

        n_stalls:
            The number of stalls.

        n_items_stocked:
            The number of kinds of produce that every stall sells.

        random_state:
            The seed or generator of the stocks, the markups, and the shopping lists.

    Returns:
        system:
            The system, with the groups `stalls` and `shoppers`, the mean number of items left on the lists
            `items_left`, and the mean money spent `spent`.

    Examples:
        >>> from skcomplex.systems.economics.markets import produce_market
        >>> [group.size for group in produce_market(10, 3).elements]
        [3, 10]
    """
    rng = np.random.default_rng(random_state)
    stocks = _random_masks(np.full(n_stalls, n_items_stocked), rng)
    markups = rng.integers(1, MAX_MARKUP + 1, n_stalls) / 100
    lists = _random_masks(rng.integers(1, MAX_LIST_LENGTH + 1, n_shoppers), rng)
    stalls = GroupElement(
        'stalls',
        element=StallElement,
        size=n_stalls,
        values={'stall': np.arange(n_stalls, dtype=float), 'stock': stocks, 'markup': markups},
    )
    shoppers = GroupElement('shoppers', element=ShopperElement, size=n_shoppers, values={'wanted': lists})
    return System(
        elements=[stalls, shoppers],
    )

riccati_prices(price=1.0, *, k=1.0, beta=10.0, lam=1.0, delta=1.0, alpha=2.0, gamma=-0.1)

Return a price that moves with excess demand under linear demand and saturating quadratic supply.

The price follows the Riccati equation dP/dt = k (D - S) with demand D = beta - lam P and supply S = delta + alpha P + gamma P^2. When (lam + alpha)^2 > 4 gamma (delta - beta) there are two equilibrium prices, and prices below the larger one converge to the smaller one. Otherwise the price blows up in finite time.

Parameters:

Name	Type	Description	Default
price	float	The initial price.	1.0
k	float	The speed of the price adjustment.	1.0
beta	float	The demand at price zero.	10.0
lam	float	The slope of the demand.	1.0
delta	float	The supply at price zero.	1.0
alpha	float	The linear intensity of the supply.	2.0
gamma	float	The quadratic intensity of the supply, negative for saturation.	-0.1

Returns:

Name	Type	Description
system	System	The system, with the price market.P.

Examples:

>>> from skcomplex.systems.economics.markets import riccati_prices
>>> riccati_prices(gamma=-0.2).elements[0].gamma
-0.2

Source code in src/skcomplex/systems/economics/markets/_prices.py

@register_system(
    field='economics/markets',
    tags=('market', 'riccati', 'ode'),
    references=('Forrest 2018',),
    defaults={'until': 10.0, 'dt': 0.1},
)
def riccati_prices(
    price: float = 1.0,
    *,
    k: float = 1.0,
    beta: float = 10.0,
    lam: float = 1.0,
    delta: float = 1.0,
    alpha: float = 2.0,
    gamma: float = -0.1,
) -> System:
    """Return a price that moves with excess demand under linear demand and saturating quadratic supply.

    The price follows the Riccati equation `dP/dt = k (D - S)` with demand `D = beta - lam P` and supply
    `S = delta + alpha P + gamma P^2`. When `(lam + alpha)^2 > 4 gamma (delta - beta)` there are two equilibrium prices,
    and prices below the larger one converge to the smaller one. Otherwise the price blows up in finite time.

    Args:
        price:
            The initial price.

        k:
            The speed of the price adjustment.

        beta:
            The demand at price zero.

        lam:
            The slope of the demand.

        delta:
            The supply at price zero.

        alpha:
            The linear intensity of the supply.

        gamma:
            The quadratic intensity of the supply, negative for saturation.

    Returns:
        system:
            The system, with the price `market.P`.

    Examples:
        >>> from skcomplex.systems.economics.markets import riccati_prices
        >>> riccati_prices(gamma=-0.2).elements[0].gamma
        -0.2
    """
    market = QuadraticSupplyMarketElement(
        'market',
        P=price,
        k=k,
        beta=beta,
        lam=lam,
        delta=delta,
        alpha=alpha,
        gamma=gamma,
    )
    adjustment = RateInteraction('adjustment', target='market', state='P', rate=_excess_demand)
    return System(elements=[market], interactions=[adjustment])

shops(n_shops=16, *, width=20, height=20, online=0.0, distance_cost=1.0, capacity=1000.0, fixed_cost=20.0, value=20.0, random_state=0)

Return shops on a grid that compete for the shoppers on every cell, in person or in a digital market.

Every step every shopper wants one unit and shops online with probability online, and in person otherwise, when it pays a travel cost for the distance to the shop. The shoppers choose in random order the shops they prefer by price plus travel cost, within the shops' capacities, a greedy matching that pays the shop's ask for a unit. Shops search for the price of the largest profit by trial and error, pay a fixed cost every step, and close when their money runs out. Distance gives every shop a local market, so many shops survive with high prices. In the digital market the cheapest shop takes every shopper, the others undercut it, prices fall toward the unit cost, and most shops close (Hamill and Gilbert 2016 sec. 6.3-6.4).

Parameters:

Name	Type	Description	Default
n_shops	int	The initial number of shops.	16
width	int	The number of columns of the grid, with one shopper per cell.	20
height	int	The number of rows of the grid.	20
online	float	The probability that a shopper shops online in a step.	0.0
distance_cost	float	The travel cost per cell of shopping in person.	1.0
capacity	float	The units every shop can sell in a step.	1000.0
fixed_cost	float	The cost of running a shop for a step.	20.0
value	float	The highest price a shopper pays.	20.0
random_state	int | Generator | None	The seed or generator of the cells and initial prices of the shops.	0

Returns:

Name	Type	Description
system	System	The system, with the groups shoppers and shops and the market shopping, whose reporters are the shopping.value, shopping.volume, and shopping.unmatched of the purchases.

Examples:

>>> from skcomplex.systems.economics.markets import shops
>>> [group.size for group in shops(4, width=5, height=5).elements]
[25, 4]

Source code in src/skcomplex/systems/economics/markets/_shops.py

@register_system(
    field='economics/markets',
    tags=('market', 'agents', 'retail', 'digital market', 'grid'),
    references=('Hamill and Gilbert 2016 sec. 6.3-6.4',),
    defaults={'until': 100},
    observables=_shops_observables,
)
def shops(
    n_shops: int = 16,
    *,
    width: int = 20,
    height: int = 20,
    online: float = 0.0,
    distance_cost: float = 1.0,
    capacity: float = 1000.0,
    fixed_cost: float = 20.0,
    value: float = 20.0,
    random_state: int | np.random.Generator | None = 0,
) -> System:
    """Return shops on a grid that compete for the shoppers on every cell, in person or in a digital market.

    Every step every shopper wants one unit and shops online with probability `online`, and in person otherwise, when it
    pays a travel cost for the distance to the shop. The shoppers choose in random order the shops they prefer by price
    plus travel cost, within the shops' capacities, a greedy matching that pays the shop's ask for a unit. Shops search
    for the price of the largest profit by trial and error, pay a fixed cost every step, and close when their money runs
    out. Distance gives every shop a local market, so many shops survive with high prices. In the digital market the
    cheapest shop takes every shopper, the others undercut it, prices fall toward the unit cost, and most shops close
    (Hamill and Gilbert 2016 sec. 6.3-6.4).

    Args:
        n_shops:
            The initial number of shops.

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

        height:
            The number of rows of the grid.

        online:
            The probability that a shopper shops online in a step.

        distance_cost:
            The travel cost per cell of shopping in person.

        capacity:
            The units every shop can sell in a step.

        fixed_cost:
            The cost of running a shop for a step.

        value:
            The highest price a shopper pays.

        random_state:
            The seed or generator of the cells and initial prices of the shops.

    Returns:
        system:
            The system, with the groups `shoppers` and `shops` and the market `shopping`, whose reporters are the
            `shopping.value`, `shopping.volume`, and `shopping.unmatched` of the purchases.

    Examples:
        >>> from skcomplex.systems.economics.markets import shops
        >>> [group.size for group in shops(4, width=5, height=5).elements]
        [25, 4]
    """
    rng = np.random.default_rng(random_state)
    cells = [(x, y) for x in range(width) for y in range(height)]
    places = rng.choice(len(cells), size=min(n_shops, len(cells)), replace=False)
    shoppers = GroupElement(
        'shoppers',
        element=ResidentElement,
        size=len(cells),
        values={'bid': value, 'online': online, 'distance_cost': distance_cost},
        location=cells,
    )
    stores = GroupElement(
        'shops',
        element=StoreElement,
        size=len(places),
        values={
            'price': rng.uniform(0.25 * value, 0.75 * value, len(places)),
            'capacity': capacity,
            'fixed_cost': fixed_cost,
        },
        variable=True,
        location=[cells[index] for index in places],
    )
    plan = [
        MapInteraction(
            'demand',
            target='shoppers',
            state='demand',
            function=lambda agent, system: 1.0,
            form='agent',
            phase='market',
        ),
        MapInteraction(
            'travel',
            target='shoppers',
            state='travel',
            function=ResidentElement.get_travel,
            form='agent',
            phase='market',
        ),
        *(
            MapInteraction(
                f'opening_{name}',
                target='shops',
                state=name,
                function=function,
                form='agent',
                phase='market',
            )
            for name, function in StoreElement.get_opening()
        ),
    ]
    settlement = [
        *(
            MapInteraction(
                f'settlement_{name}',
                target='shops',
                state=name,
                function=function,
                form='agent',
                phase='accounts',
            )
            for name, function in StoreElement.get_settlement()
        ),
        SignAdjustmentInteraction(
            'settlement_price',
            target='shops',
            state='price',
            signal=lambda agent, system: agent.direction,
            step=lambda agent, system: agent.adjustment,
            lower=lambda agent, system: agent.unit_cost,
            form='agent',
            phase='accounts',
        ),
        LifeCycleInteraction('closure', target='shops', death=lambda agent, system: agent.money < 0, phase='accounts'),
    ]
    shopping = MatchingInteraction[SystemView](
        'shopping',
        requesters='shoppers',
        providers='shops',
        rule=PartnerSpec(choice='score', score=ResidentElement.get_preference, clearing='provider'),
        limit=('bid', 'ask'),
        quantity=('demand', 'supply'),
        transfers=(('money', -1),),
        phase='market',
        on_match=StoreElement.record_sale,
    )
    return System(
        elements=[shoppers, stores],
        interactions=[*plan, shopping, *settlement],
        space=LatticeSpace((width, height), boundary='fixed', neighborhood='moore'),
        time=DiscreteTime(schedule=PhasesScheduleSpec([PhaseSpec('market'), PhaseSpec('accounts')])),
    )

sugarscape(width=50, height=50, *, n_foragers=400, max_vision=6, max_metabolism=4.0, low_sugar=5.0, high_sugar=25.0, growback=1.0, random_state=0)

Return the Sugarscape of Epstein and Axtell, foragers of random vision and metabolism on two sugar peaks.

Every step the foragers, in random order, move to the richest free cell in sight, harvest it, and burn their metabolism, and those whose wealth falls below zero die. Then the sugar grows back. Foragers of poor vision or high metabolism starve, so the population falls to the carrying capacity of the landscape, and the wealth of the survivors becomes skewed, with a Gini coefficient that rises from its initial value.

Parameters:

Name	Type	Description	Default
width	int	The number of columns.	50
height	int	The number of rows.	50
n_foragers	int	The initial number of foragers.	400
max_vision	int	The largest vision, every forager seeing from 1 to max_vision cells.	6
max_metabolism	float	The largest metabolism, every forager burning from 1 to max_metabolism sugar a step.	4.0
low_sugar	float	The lowest initial wealth.	5.0
high_sugar	float	The highest initial wealth.	25.0
growback	float	The sugar that grows back on every patch in a step.	1.0
random_state	int | Generator | None	The seed or generator of the foragers' cells, visions, metabolisms, and wealth.	0

Returns:

Name	Type	Description
system	System	The system, with the groups patches and foragers, whose catalog observables are the population population and the Gini coefficient of wealth gini.

Examples:

>>> from skcomplex.systems.economics.markets import sugarscape
>>> [group.size for group in sugarscape(10, 10, n_foragers=20).elements]
[100, 20]

Source code in src/skcomplex/systems/economics/markets/_sugarscape.py

@register_system(
    field='economics/markets',
    tags=('sugarscape', 'wealth', 'agents', 'grid'),
    references=('Argun et al. 2021 sec. 16.2', 'Downey 2018 ch. 9'),
    defaults={'until': 100},
    observables=lambda system: [
        ObservableSpec('population', group='foragers', variable='sugar', reduce='count'),
        ObservableSpec('gini', group='foragers', variable='sugar', reduce=_gini),
    ],
)
def sugarscape(
    width: int = 50,
    height: int = 50,
    *,
    n_foragers: int = 400,
    max_vision: int = 6,
    max_metabolism: float = 4.0,
    low_sugar: float = 5.0,
    high_sugar: float = 25.0,
    growback: float = 1.0,
    random_state: int | np.random.Generator | None = 0,
) -> System:
    """Return the Sugarscape of Epstein and Axtell, foragers of random vision and metabolism on two sugar peaks.

    Every step the foragers, in random order, move to the richest free cell in sight, harvest it, and burn their
    metabolism, and those whose wealth falls below zero die. Then the sugar grows back. Foragers of poor vision or
    high metabolism starve, so the population falls to the carrying capacity of the landscape, and the wealth of the
    survivors becomes skewed, with a Gini coefficient that rises from its initial value.

    Args:
        width:
            The number of columns.

        height:
            The number of rows.

        n_foragers:
            The initial number of foragers.

        max_vision:
            The largest vision, every forager seeing from 1 to `max_vision` cells.

        max_metabolism:
            The largest metabolism, every forager burning from 1 to `max_metabolism` sugar a step.

        low_sugar:
            The lowest initial wealth.

        high_sugar:
            The highest initial wealth.

        growback:
            The sugar that grows back on every patch in a step.

        random_state:
            The seed or generator of the foragers' cells, visions, metabolisms, and wealth.

    Returns:
        system:
            The system, with the groups `patches` and `foragers`, whose catalog observables are the population
            `population` and the Gini coefficient of wealth `gini`.

    Examples:
        >>> from skcomplex.systems.economics.markets import sugarscape
        >>> [group.size for group in sugarscape(10, 10, n_foragers=20).elements]
        [100, 20]
    """
    rng = np.random.default_rng(random_state)
    capacity = landscape(width, height).ravel()
    cells = [(x, y) for x in range(width) for y in range(height)]
    patches = GroupElement(
        'patches',
        element=SugarPatchElement,
        size=len(cells),
        values={'sugar': capacity, 'capacity': capacity, 'growback': growback},
        location=cells,
    )
    occupied = rng.choice(len(cells), size=min(n_foragers, len(cells)), replace=False)
    foragers = GroupElement(
        'foragers',
        element=ForagerElement,
        size=len(occupied),
        values={
            'sugar': rng.uniform(low_sugar, high_sugar, len(occupied)),
            'vision': rng.integers(1, max_vision + 1, len(occupied)).astype(float),
            'metabolism': rng.uniform(1.0, max_metabolism, len(occupied)),
        },
        variable=True,
        location=[cells[index] for index in occupied],
    )
    return System(
        elements=[patches, foragers],
        interactions=[
            *_get_foraging(),
            RegrowthInteraction('growback', target='patches', state='sugar', growth='growback', phase='growth'),
        ],
        space=LatticeSpace((width, height), boundary='fixed', neighborhood='von_neumann'),
        time=DiscreteTime(schedule=PhasesScheduleSpec([PhaseSpec('foraging'), PhaseSpec('growth')])),
    )

toy_trader(n_traders=400, *, sight=1.0, mutation=0.0, imitation=0.1, fitness='none', n_goods=1, size=51.0, random_state=0)

Return Gooding's Toy Trader, traders with money and toys who buy the cheapest toy they can see.

Four hundred traders on a 51 by 51 torus start with 100 money, 5 toys, and a toy price of 10. Every iteration every trader buys one toy from the cheapest trader within its market sight, if it can pay, and moves one unit in a random direction. A seller keeps its stock at 3 toys by lowering its price by the factor 0.99 above it and raising it by 1.0101 below it (Gooding 2019, Section 8.2 and the appendix). Money and toys are conserved and supply exceeds the demand for stocks, yet the price depends on the market sight: it falls when traders see only their neighbours and rises when they see everyone, so supply and demand alone do not set it (Section 10.1). Mutating pricing factors make the traders heterogeneous and raise prices and inequality, and the imitation of the richest 5 percent by the poorest half under a fitness test of money or of assets evolves the price patterns of speculation and asset markets (Sections 10.2 and 11). With n_goods=3 the traders also trade crayons and blackboards, starting with 30 and 15, which keep the fixed factors and the target stocks of 10 to 15 and 5 to 8 of the Three-Goods Trader (Section 11.1).

Parameters:

Name	Type	Description	Default
n_traders	int	The number of traders.	400
sight	float	The radius within which a trader sees prices.	1.0
mutation	float	The probability per iteration that a factor of the toy price mutates, such as 1 / 5000.	0.0
imitation	float	The probability per iteration that a poor trader imitates a rich one under a fitness test.	0.1
fitness	str	The fitness test of imitation, 'none', 'money', or 'assets'.	'none'
n_goods	int	The goods traded, 1 for toys only or 3 for toys, crayons, and blackboards.	1
size	float	The side of the torus.	51.0
random_state	int | Generator | None	The seed or generator of the places of the traders.	0

Returns:

Name	Type	Description
system	System	The system, with the group traders and the market market, whose reporters are the average price of the toys traded market.average_price, the toys traded market.trades, and the Gini coefficients of money market.money_gini and of assets market.asset_gini.

Examples:

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

Source code in src/skcomplex/systems/economics/markets/_toy_trader.py

@register_system(
    field='economics/markets',
    tags=('money', 'prices', 'agents', 'inequality', 'evolution', 'imitation', 'market sight'),
    references=('Gooding 2019',),
    defaults={'until': 300},
)
def toy_trader(
    n_traders: int = 400,
    *,
    sight: float = 1.0,
    mutation: float = 0.0,
    imitation: float = 0.1,
    fitness: str = 'none',
    n_goods: int = 1,
    size: float = 51.0,
    random_state: int | np.random.Generator | None = 0,
) -> System:
    """Return Gooding's Toy Trader, traders with money and toys who buy the cheapest toy they can see.

    Four hundred traders on a 51 by 51 torus start with 100 money, 5 toys, and a toy price of 10. Every iteration every
    trader buys one toy from the cheapest trader within its market sight, if it can pay, and moves one unit in a random
    direction. A seller keeps its stock at 3 toys by lowering its price by the factor 0.99 above it and raising it by
    1.0101 below it (Gooding 2019, Section 8.2 and the appendix). Money and toys are conserved and supply exceeds the
    demand for stocks, yet the price depends on the market sight: it falls when traders see only their neighbours and
    rises when they see everyone, so supply and demand alone do not set it (Section 10.1). Mutating pricing factors
    make the traders heterogeneous and raise prices and inequality, and the imitation of the richest 5 percent by the
    poorest half under a fitness test of money or of assets evolves the price patterns of speculation and asset markets
    (Sections 10.2 and 11). With `n_goods=3` the traders also trade crayons and blackboards, starting with 30 and 15,
    which keep the fixed factors and the target stocks of 10 to 15 and 5 to 8 of the Three-Goods Trader (Section 11.1).

    Args:
        n_traders:
            The number of traders.

        sight:
            The radius within which a trader sees prices.

        mutation:
            The probability per iteration that a factor of the toy price mutates, such as 1 / 5000.

        imitation:
            The probability per iteration that a poor trader imitates a rich one under a fitness test.

        fitness:
            The fitness test of imitation, `'none'`, `'money'`, or `'assets'`.

        n_goods:
            The goods traded, 1 for toys only or 3 for toys, crayons, and blackboards.

        size:
            The side of the torus.

        random_state:
            The seed or generator of the places of the traders.

    Returns:
        system:
            The system, with the group `traders` and the market `market`, whose reporters are the average price of the
            toys traded `market.average_price`, the toys traded `market.trades`, and the Gini coefficients of money
            `market.money_gini` and of assets `market.asset_gini`.

    Examples:
        >>> from skcomplex.systems.economics.markets import toy_trader
        >>> toy_trader(10).elements[0].prepare().size_
        10
    """
    rng = np.random.default_rng(random_state)
    places = rng.uniform(0, size, (n_traders, 2))
    stocks = dict(zip(GOODS, STOCKS[:n_goods] + (0.0,) * (len(GOODS) - n_goods), strict=True))
    traders = GroupElement(
        'traders',
        element=ToyTraderElement,
        size=n_traders,
        values={'x': places[:, 0], 'y': places[:, 1], **stocks},
    )
    market = ToyMarketInteraction(
        'market',
        traders='traders',
        sight=sight,
        mutation=mutation,
        imitation=imitation,
        size=size,
        n_goods=n_goods,
        fitness=fitness,
    )
    return System(elements=[traders], interactions=[market])

trade_goods(first, second, goods, weights, *, quantity=0.1, limit=1000)

Barter small quantities of goods between two agents while both gain, and return the trades.

For every pair of goods in turn, the agent that values the first good more in units of the second buys quantity of it from the other at the geometric mean of their marginal rates of substitution, as in the Sugarscape trade of Epstein and Axtell (1996). A trade happens only when it raises the Cobb-Douglas utility of both agents, and every trade is two transfers, so the total of every good is unchanged. The agents trade until a round over all pairs of goods makes no trade, or until limit trades.

Parameters:

Name	Type	Description	Default
first	object	The first agent, with a state for the holding and for the weight of every good.	required
second	object	The second agent.	required
goods	Sequence[str]	The names of the states that hold the goods.	required
weights	Sequence[str]	The names of the states that hold the weights of the goods, in the order of goods.	required
quantity	float	The quantity of the bought good in every trade.	0.1
limit	int	The largest number of trades.	1000

Returns:

Name	Type	Description
trades	list[tuple[int, int, float]]	The index of the bought good, the index of the good paid with, and the price of every trade.

Examples:

>>> from types import SimpleNamespace
>>> from skcomplex.systems.economics.markets import trade_goods
>>> first = SimpleNamespace(x=9.0, y=1.0, a=0.5, b=0.5)
>>> second = SimpleNamespace(x=1.0, y=9.0, a=0.5, b=0.5)
>>> trades = trade_goods(first, second, ('x', 'y'), ('a', 'b'), quantity=0.5)
>>> first.x + second.x, len(trades) > 0
(10.0, True)

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

def trade_goods(
    first: object,
    second: object,
    goods: Sequence[str],
    weights: Sequence[str],
    *,
    quantity: float = 0.1,
    limit: int = 1000,
) -> list[tuple[int, int, float]]:
    """Barter small quantities of goods between two agents while both gain, and return the trades.

    For every pair of goods in turn, the agent that values the first good more in units of the second buys `quantity`
    of it from the other at the geometric mean of their marginal rates of substitution, as in the Sugarscape trade of
    Epstein and Axtell (1996). A trade happens only when it raises the Cobb-Douglas utility of both agents, and every
    trade is two transfers, so the total of every good is unchanged. The agents trade until a round over all pairs of
    goods makes no trade, or until `limit` trades.

    Args:
        first:
            The first agent, with a state for the holding and for the weight of every good.

        second:
            The second agent.

        goods:
            The names of the states that hold the goods.

        weights:
            The names of the states that hold the weights of the goods, in the order of `goods`.

        quantity:
            The quantity of the bought good in every trade.

        limit:
            The largest number of trades.

    Returns:
        trades:
            The index of the bought good, the index of the good paid with, and the price of every trade.

    Examples:
        >>> from types import SimpleNamespace
        >>> from skcomplex.systems.economics.markets import trade_goods
        >>> first = SimpleNamespace(x=9.0, y=1.0, a=0.5, b=0.5)
        >>> second = SimpleNamespace(x=1.0, y=9.0, a=0.5, b=0.5)
        >>> trades = trade_goods(first, second, ('x', 'y'), ('a', 'b'), quantity=0.5)
        >>> first.x + second.x, len(trades) > 0
        (10.0, True)
    """
    trades: list[tuple[int, int, float]] = []
    pairs = list(itertools.combinations(range(len(goods)), 2))
    while len(trades) < limit:
        traded = False
        for pair in pairs:
            price = _trade_once(first, second, pair, goods=goods, weights=weights, quantity=quantity)
            if price is not None:
                trades.append((pair[0], pair[1], price))
                traded = True
        if not traded:
            break
    return trades

wilhite_trade_networks(n_agents=500, *, topology='small_world', n_groups=10, shortcuts=5, endowment=100.0, units=100, random_state=0)

Return Wilhite's bilateral barter of two goods among traders on one of four trade networks.

Traders start with random bundles of two goods and the utility good1 good2. Every step, each trader searches its neighbours and trades with the one whose marginal rate of substitution differs the most, single units at a price between their rates, while both gain. Trade drives the rates of every connected set of traders towards one price. On the complete network every trader searches every other, so it needs the most searches, and traders crowd onto the most extreme partners, so prices converge slowly. Isolated groups settle on different prices. Groups chained by shared end traders, or a small world with some long links, bring the prices together with far fewer searches.

Parameters:

Name	Type	Description	Default
n_agents	int	The number of traders.	500
topology	str	'complete', 'isolated' groups, groups 'linked' in a ring by shared end traders, or 'small_world', the linked groups with random long links.	'small_world'
n_groups	int	The number of groups of the isolated, linked, and small-world networks.	10
shortcuts	int	The number of random long links between groups of the small world.	5
endowment	float	The largest initial holding of each good, drawn uniformly from 1 to it.	100.0
units	int	The most units one pair trades in a step.	100
random_state	int | Generator | None	The seed or generator of the bundles and the long links.	0

Returns:

Name	Type	Description
system	System	The system, with the group traders on the network, the reporters barter.trades and barter.searches, and the observables dispersion, the spread of the logarithms of the rates, and utility, the mean utility.

Raises:

Type	Description
ValueError	The topology is unknown.

Examples:

>>> from skcomplex.systems.economics.markets import wilhite_trade_networks
>>> system = wilhite_trade_networks(40, topology='isolated', n_groups=4)
>>> system.elements[0].prepare().size_
40

Source code in src/skcomplex/systems/economics/markets/_barter_networks.py

@register_system(
    field='economics/markets',
    tags=('barter', 'trade', 'network', 'small world', 'agents'),
    references=('Wilhite 2001', 'Gooding 2019'),
    defaults={'until': 60},
    observables=lambda system: [
        ObservableSpec('dispersion', group='traders', variable='mrs', reduce=_log_spread),
        ObservableSpec('utility', _utility, group='traders'),
    ],
)
def wilhite_trade_networks(
    n_agents: int = 500,
    *,
    topology: str = 'small_world',
    n_groups: int = 10,
    shortcuts: int = 5,
    endowment: float = 100.0,
    units: int = 100,
    random_state: int | np.random.Generator | None = 0,
) -> System:
    """Return Wilhite's bilateral barter of two goods among traders on one of four trade networks.

    Traders start with random bundles of two goods and the utility `good1 good2`. Every step, each trader searches
    its neighbours and trades with the one whose marginal rate of substitution differs the most, single units at a
    price between their rates, while both gain. Trade drives the rates of every connected set of traders towards one
    price. On the complete network every trader searches every other, so it needs the most searches, and traders crowd
    onto the most extreme partners, so prices converge slowly. Isolated groups settle on different prices. Groups
    chained by shared end traders, or a small world with some long links, bring the prices together with far fewer
    searches.

    Args:
        n_agents:
            The number of traders.

        topology:
            `'complete'`, `'isolated'` groups, groups `'linked'` in a ring by shared end traders, or `'small_world'`,
            the linked groups with random long links.

        n_groups:
            The number of groups of the isolated, linked, and small-world networks.

        shortcuts:
            The number of random long links between groups of the small world.

        endowment:
            The largest initial holding of each good, drawn uniformly from 1 to it.

        units:
            The most units one pair trades in a step.

        random_state:
            The seed or generator of the bundles and the long links.

    Returns:
        system:
            The system, with the group `traders` on the network, the reporters `barter.trades` and `barter.searches`,
            and the observables `dispersion`, the spread of the logarithms of the rates, and `utility`, the mean
            utility.

    Raises:
        ValueError:
            The topology is unknown.

    Examples:
        >>> from skcomplex.systems.economics.markets import wilhite_trade_networks
        >>> system = wilhite_trade_networks(40, topology='isolated', n_groups=4)
        >>> system.elements[0].prepare().size_
        40
    """
    rng = np.random.default_rng(random_state)
    graph = _trade_network(topology, n_agents, n_groups, shortcuts, rng)
    goods = rng.integers(1, int(endowment) + 1, size=(2, n_agents)).astype(float)
    traders = GroupElement(
        'traders',
        element=BarterTraderElement,
        size=n_agents,
        values={
            'good1': goods[0],
            'good2': goods[1],
            'mrs': goods[1] / goods[0],
            'links': np.array([graph.degree(node) for node in range(n_agents)], dtype=float),
        },
        location=list(range(n_agents)),
    )
    return System(
        elements=[traders],
        interactions=[
            MapInteraction(
                'appraisal',
                target='traders',
                state='mrs',
                function=BarterTraderElement.get_rate,
                form='agent',
            ),
            BilateralBarterInteraction(
                'barter',
                requesters='traders',
                providers='traders',
                rule=PartnerSpec(choice='score', score=BarterTraderElement.get_rate_gap, clearing='provider'),
                quantity=('quantity', 'partners'),
                units=units,
            ),
        ],
        space=NetworkSpace(graph),
        time=DiscreteTime(),
    )