Markets

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

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

Bases: FeedbackInteraction[ModelView]

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.models.economics.markets import AuctionLedgerInteraction
>>> AuctionLedgerInteraction('ledger', phase='close').reports
('surplus', 'efficiency')

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

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

act(model)

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

Parameters:

Name	Type	Description	Default
model	ModelView	The model view scoped to the ledger.	required

Source code in src/skcomplex/models/economics/markets/_auction_ledger.py

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

    Args:
        model:
            The model view scoped to the ledger.
    """
    buyers = model.agents('buyers', ZeroIntelligenceBuyerView)
    sellers = model.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())
    model.report('surplus', surplus)
    model.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.elements import GroupElement
>>> from skcomplex.models.economics.markets import BarterTraderElement
>>> GroupElement('traders', element=BarterTraderElement, size=4).size
4

Source code in src/skcomplex/models/economics/markets/_barter_trader.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, model) staticmethod

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

Parameters:

Name	Type	Description	Default
agent	BarterTraderView	The trader.	required
model	ModelView	The model view of the appraisal.	required

Returns:

Name	Type	Description
mrs	float	The marginal rate of substitution.

Source code in src/skcomplex/models/economics/markets/_barter_trader.py

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

    Args:
        agent:
            The trader.

        model:
            The model 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/models/economics/markets/_barter_trader.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/models/economics/markets/_barter_trader.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.models.economics.markets import BartererElement
>>> GroupElement('barterers', element=BartererElement, size=2, values={'chocolate': [8.0, 2.0]}).size
2

Source code in src/skcomplex/models/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[ModelView]

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, ModelView], None] | None	A function of every match and the model 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.models.economics.markets import BilateralBarterInteraction
>>> BilateralBarterInteraction('barter', requesters='traders', providers='traders').units
100

Source code in src/skcomplex/models/economics/markets/_bilateral_barter.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, ModelView], 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(model)

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

Parameters:

Name	Type	Description	Default
model	ModelView	The model view of the interaction.	required

Source code in src/skcomplex/models/economics/markets/_bilateral_barter.py

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

    Args:
        model:
            The model view of the interaction.
    """
    model.storage['trades'] = 0.0
    super().match(model)
    model.report('trades', self._get_trades(model))
    traders = model.agents('requesters', BarterTraderView)
    model.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.models.economics.markets import BreweryInteraction
>>> BreweryInteraction('brewery', factory='factory').factory
'factory'

Source code in src/skcomplex/models/economics/markets/_brewery.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

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 model 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.models.economics.markets import CarDealershipElement
>>> CarDealershipElement('dealer', inventory=150.0).inventory
150.0

Source code in src/skcomplex/models/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.elements import GroupElement
>>> from skcomplex.models.economics.markets import CarElement
>>> GroupElement('cars', element=CarElement, size=1).size
1

Source code in src/skcomplex/models/economics/markets/_car.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.models.economics.markets import CarWashOperationInteraction
>>> CarWashOperationInteraction('wash', cars='cars', till='till', p_enter=0.01).p_enter
0.01

Source code in src/skcomplex/models/economics/markets/_car_wash_operation.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(model)

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
model	ModelView	The model.	required

Source code in src/skcomplex/models/economics/markets/_car_wash_operation.py

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

    Args:
        model:
            The model.
    """
    values = {name: float(model.parameter(name)) for name in _PARAMETERS}
    till = model.agents('till', TillView)[0]
    for car in model.agents('cars', CarView):
        car.crew = 0
        if car.work <= 0:
            self._finish(car, till, model, values)
    if model.rng.random() >= values['p_pass'] * values['p_enter']:
        return
    if sum(car.stage == WASH for car in model.agents('cars', CarView)) >= values['balk']:
        till.turned_away += 1
        return
    model.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 model, 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.models.economics.markets import CobwebInteraction
>>> CobwebInteraction('cobweb', target='wheat', a=10, b=2, c=1, d=1).b
2

Source code in src/skcomplex/models/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 model 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.models.economics.markets import CommodityMarketElement
>>> CommodityMarketElement('market', lifetime=20.0).lifetime
20.0

Source code in src/skcomplex/models/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.models.economics.markets import ConsumerElement
>>> GroupElement('consumers', element=ConsumerElement, size=4, values={'ethics': 0.5}).size
4

Source code in src/skcomplex/models/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.models.economics.markets import CournotAdjustmentInteraction
>>> CournotAdjustmentInteraction('market', target='firms', error=0.1).check().error
0.1

Source code in src/skcomplex/models/economics/markets/_cournot_adjustment.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.models.economics.markets import CournotBestResponseElement
>>> CournotBestResponseElement('duopoly', demand=8.0).demand
8.0

Source code in src/skcomplex/models/economics/markets/_cournot_best_response.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.elements import GroupElement
>>> from skcomplex.models.economics.markets import CustomerElement
>>> GroupElement('customers', element=CustomerElement, size=10, values={'s1': 0.5, 's2': 0.5}).size
10

Source code in src/skcomplex/models/economics/markets/_customer.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, model) staticmethod

Return the node of the shop the customer chose.

Parameters:

Name	Type	Description	Default
agent	AgentViewProtocol	The customer.	required
model	ModelViewProtocol	The model 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/models/economics/markets/_customer.py

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

    Args:
        agent:
            The customer.

        model:
            The model 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, model, 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
model	ModelView	The model 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/models/economics/markets/_customer.py

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

    Args:
        agent:
            The customer.

        model:
            The model 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/models/economics/markets/_customer.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, model, option) staticmethod

Return the utility S_k / T of visiting shop k.

Parameters:

Name	Type	Description	Default
agent	CustomerView	The customer.	required
model	ModelView	The model 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/models/economics/markets/_customer.py

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

    Args:
        agent:
            The customer.

        model:
            The model 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.elements import GroupElement
>>> from skcomplex.models.economics.markets import DuopolistElement
>>> GroupElement('firms', element=DuopolistElement, size=2, values={'output': [100.0, 100.0]}).size
2

Source code in src/skcomplex/models/economics/markets/_duopolist.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.models.economics.markets import EchelonElement
>>> EchelonElement('retailer', inventory=10.0).inventory
10.0

Source code in src/skcomplex/models/economics/markets/_echelon.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.elements import GroupElement
>>> from skcomplex.models.economics.markets import EmployerElement
>>> GroupElement('firms', element=EmployerElement, size=5).size
5

Source code in src/skcomplex/models/economics/markets/_employer.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, ModelView], float]], ...]	The name of every state and the function of its value, of the firm and the model view.

Source code in src/skcomplex/models/economics/markets/_employer.py

@staticmethod
def get_posting() -> tuple[tuple[str, Callable[[EmployerView, ModelView], 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 model view.
    """
    return (
        ('demand', lambda agent, model: float(model.rng.random() < agent.vacancy_rate)),
        (
            'bid',
            lambda agent, model: (
                float(model.rng.uniform(agent.low_wage, agent.high_wage)) if agent.demand else agent.bid
            ),
        ),
        ('posted', lambda agent, model: model.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.models.economics.markets import ExchangeInteraction
>>> ExchangeInteraction('exchange', source='traders', target='traders').reports
('price', 'dispersion', 'volume')

Source code in src/skcomplex/models/economics/markets/_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(model)

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

Parameters:

Name	Type	Description	Default
model	ModelViewProtocol	The model view of the exchange.	required

Source code in src/skcomplex/models/economics/markets/_exchange_interaction.py

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

    Args:
        model:
            The model view of the exchange.
    """
    trades = model.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)]
    model.report('price', math.exp(float(np.mean(logs))) if logs else math.nan)
    model.report('dispersion', float(np.std(logs)) if len(logs) > 1 else math.nan)
    model.report('volume', len(trades))

transfer(source, target, model)

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
model	ModelViewProtocol	The model view of the exchange.	required

Returns:

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

Source code in src/skcomplex/models/economics/markets/_exchange_interaction.py

def transfer(self, source: AgentViewProtocol, target: AgentViewProtocol, model: ModelViewProtocol) -> 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.

        model:
            The model 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 = model.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.models.economics.markets import FactoryElement
>>> FactoryElement('factory').production_delay
3.0

Source code in src/skcomplex/models/economics/markets/_factory.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

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.elements import GroupElement
>>> from skcomplex.models.economics.markets import ForagerElement
>>> GroupElement('foragers', element=ForagerElement, size=3, values={'vision': 2.0}).size
3

Source code in src/skcomplex/models/economics/markets/_forager.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.elements import GroupElement
>>> from skcomplex.models.economics.markets import GoodsTraderElement
>>> GroupElement('traders', element=GoodsTraderElement, size=4, values={'sugar': 5.0}).size
4

Source code in src/skcomplex/models/economics/markets/_goods_trader.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[ModelView]

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[[ModelView], 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.models.economics.markets import HeuristicPricingInteraction
>>> HeuristicPricingInteraction('pricing', traders='traders').reports
('deviation', 'trend_share', 'fitness_fundamentalist', 'fitness_trend')

Source code in src/skcomplex/models/economics/markets/_heuristic_pricing.py

def __init__(
    self,
    label: str,
    *,
    rule: Callable[[ModelView], 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(model)

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

Parameters:

Name	Type	Description	Default
model	ModelView	The model view of the institution.	required

Source code in src/skcomplex/models/economics/markets/_heuristic_pricing.py

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

    Args:
        model:
            The model view of the institution.
    """
    rate, strength = model.parameter('rate'), model.parameter('strength')
    state = model.storage.get('market')
    if not isinstance(state, _MarketStateSpec):
        state = _MarketStateSpec((model.parameter('initial'),))
    history = list(state.history)
    forecasts = np.array(
        [heuristics.fundamental(history), heuristics.fundamental(history, fundamental=0.0, reversion=strength)],
    )
    traders = model.agents('traders', HeuristicTraderView)
    share = float(np.mean([trader.predictor for trader in traders])) if traders else 0.0
    price = float(forecasts @ [1 - share, share] + model.rng.normal(0.0, model.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]) / model.parameter('risk')
    fitness -= [model.parameter('cost'), 0.0]
    model.storage['market'] = _MarketStateSpec((history[-1], price), forecasts)
    model.report('deviation', price)
    model.report('trend_share', share)
    for name, value in zip(PREDICTORS, fitness, strict=True):
        model.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.elements import GroupElement
>>> from skcomplex.models.economics.markets import HeuristicTraderElement
>>> GroupElement('traders', element=HeuristicTraderElement, size=10, values={'predictor': 1.0}).size
10

Source code in src/skcomplex/models/economics/markets/_heuristic_trader.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, model, 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
model	ModelView	The model 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/models/economics/markets/_heuristic_trader.py

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

    Args:
        agent:
            The trader.

        model:
            The model 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, model, 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
model	ModelView	The model 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/models/economics/markets/_heuristic_trader.py

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

    Args:
        agent:
            The trader.

        model:
            The model view of the reinforcement.

        index:
            The index of the heuristic.

    Returns:
        fitness:
            The last reported fitness of the heuristic.
    """
    reported = model.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.elements import GroupElement
>>> from skcomplex.models.economics.markets import JobseekerElement
>>> GroupElement('workers', element=JobseekerElement, size=5).size
5

Source code in src/skcomplex/models/economics/markets/_jobseeker.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, ModelView], float]], ...]	The name of every state and the function of its value, of the worker and the model view.

Source code in src/skcomplex/models/economics/markets/_jobseeker.py

@staticmethod
def get_search() -> tuple[tuple[str, Callable[[JobseekerView, ModelView], 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 model view.
    """
    return (
        ('supply', lambda agent, model: float(agent.employer < 0)),
        (
            'ask',
            lambda agent, model: (
                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, ModelView], float]], ...]	The name of every state and the function of its value, of the worker and the model view.

Source code in src/skcomplex/models/economics/markets/_jobseeker.py

@staticmethod
def get_spell() -> tuple[tuple[str, Callable[[JobseekerView, ModelView], 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 model view.
    """
    return (
        ('wage', lambda agent, model: 0.0 if agent.employer < 0 else agent.wage),
        (
            'duration',
            lambda agent, model: (
                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.models.economics.markets import MarketElement
>>> MarketElement('wheat', price=2.0).price
2.0

Source code in src/skcomplex/models/economics/markets/_market.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.models.economics.markets import PricePointElement
>>> PricePointElement('point', price=0.5).price
0.5

Source code in src/skcomplex/models/economics/markets/_price_point.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/models/economics/markets/_price_point.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.elements import GroupElement
>>> from skcomplex.models.economics.markets import PricingFirmElement
>>> GroupElement('firms', element=PricingFirmElement, size=2).size
2

Source code in src/skcomplex/models/economics/markets/_pricing_firm.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.models.economics.markets import PricingGameSpec
>>> PricingGameSpec(np.array([0.5, 1.0]), 0.25, 0.5).beta
0.5

payoff(agent, model, 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
model	ModelView	The model 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/models/economics/markets/_pricing_game_spec.py

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

    Args:
        agent:
            The firm.

        model:
            The model 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 model.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, model, 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
model	ModelView	The model 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/models/economics/markets/_pricing_game_spec.py

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

    Args:
        agent:
            The firm.

        model:
            The model 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.elements import GroupElement
>>> from skcomplex.models.economics.markets import ProductElement
>>> GroupElement('products', element=ProductElement, size=4, location=[0, 0, 1, 1]).size
4

Source code in src/skcomplex/models/economics/markets/_product.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, model) 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
model	ModelView	The model view of the purchase.	required

Returns:

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

Source code in src/skcomplex/models/economics/markets/_product.py

@staticmethod
def restock(product: ProductView, customer: CustomerView, model: ModelView) -> 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.

        model:
            The model view of the purchase.

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

sell(product, customer, model) 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
model	ModelView	The model view of the purchase.	required

Returns:

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

Source code in src/skcomplex/models/economics/markets/_product.py

@staticmethod
def sell(product: ProductView, customer: CustomerView, model: ModelView) -> 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.

        model:
            The model view of the purchase.

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

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.models.economics.markets import QuadraticSupplyMarketElement
>>> QuadraticSupplyMarketElement('market', P=2.0).P
2.0

Source code in src/skcomplex/models/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.elements import GroupElement
>>> from skcomplex.models.economics.markets import ResidentElement
>>> GroupElement('shoppers', element=ResidentElement, size=3, values={'online': 1.0}).size
3

Source code in src/skcomplex/models/economics/markets/_resident.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/models/economics/markets/_resident.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, model) 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
model	ModelView	The model view of the plan.	required

Returns:

Name	Type	Description
travel	float	The travel cost per cell.

Source code in src/skcomplex/models/economics/markets/_resident.py

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

    Args:
        agent:
            The shopper.

        model:
            The model view of the plan.

    Returns:
        travel:
            The travel cost per cell.
    """
    return 0.0 if model.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.models.economics.markets import RetailDemandInteraction
>>> RetailDemandInteraction('customers', retailer='retailer').retailer
'retailer'

Source code in src/skcomplex/models/economics/markets/_retail_demand.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.models.economics.markets import RetailerElement
>>> RetailerElement('retailer').demand
4.0

Source code in src/skcomplex/models/economics/markets/_retailer.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.elements import GroupElement
>>> from skcomplex.models.economics.markets import ShopperElement
>>> GroupElement('shoppers', element=ShopperElement, size=5, values={'wanted': 3}).size
5

Source code in src/skcomplex/models/economics/markets/_shopper.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, model)

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
model	ModelView	The model.	required

Source code in src/skcomplex/models/economics/markets/_shopper.py

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

    Args:
        agent:
            The shopper.

        model:
            The model.
    """
    wanted, visited = int(agent.wanted), int(agent.visited)
    if not wanted:
        return
    unvisited = [stall for stall in model.agents(self.stalls, StallView) if not visited >> int(stall.stall) & 1]
    if not unvisited:
        agent.missed, agent.wanted = wanted, 0
        return
    stall = unvisited[int(model.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.elements import GroupElement
>>> from skcomplex.models.economics.markets import StallElement
>>> GroupElement('stalls', element=StallElement, size=3, values={'stall': [0, 1, 2]}).size
3

Source code in src/skcomplex/models/economics/markets/_stall.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.elements import GroupElement
>>> from skcomplex.models.economics.markets import StoreElement
>>> GroupElement('shops', element=StoreElement, size=4, values={'price': 10.0}).size
4

Source code in src/skcomplex/models/economics/markets/_store.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, ModelView], float]], ...]	The name of every state and the function of its value, of the shop and the model view.

Source code in src/skcomplex/models/economics/markets/_store.py

@staticmethod
def get_opening() -> tuple[tuple[str, Callable[[StoreView, ModelView], 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 model view.
    """
    return (
        ('supply', lambda agent, model: agent.capacity),
        ('ask', lambda agent, model: agent.price),
        ('sales', lambda agent, model: 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/models/economics/markets/_store.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 model then steps the price by the sign of the direction.

Returns:

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

Source code in src/skcomplex/models/economics/markets/_store.py

@staticmethod
def get_settlement() -> tuple[tuple[str, Callable[[StoreView, ModelView], 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 model 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 model view.
    """
    profit = StoreElement.get_profit
    return (
        ('direction', lambda agent, model: -agent.direction if profit(agent) < agent.profit else agent.direction),
        ('profit', lambda agent, model: profit(agent)),
        ('money', lambda agent, model: agent.money - agent.sales * agent.unit_cost - agent.fixed_cost),
    )

record_sale(match, model) 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
model	ModelView	The model view of the market.	required

Source code in src/skcomplex/models/economics/markets/_store.py

@staticmethod
def record_sale(match: MatchSpec, model: ModelView) -> 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.

        model:
            The model 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.elements import GroupElement
>>> from skcomplex.models.economics.markets import SugarPatchElement
>>> GroupElement('patches', element=SugarPatchElement, size=4, values={'capacity': 2.0}).size
4

Source code in src/skcomplex/models/economics/markets/_sugar_patch.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.models.economics.markets import SupplyLinkInteraction
>>> SupplyLinkInteraction('link', downstream='retailer', upstream='wholesaler').upstream
'wholesaler'

Source code in src/skcomplex/models/economics/markets/_supply_link.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.elements import GroupElement
>>> from skcomplex.models.economics.markets import TillElement
>>> GroupElement('till', element=TillElement, size=1).size
1

Source code in src/skcomplex/models/economics/markets/_till.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

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.elements import GroupElement
>>> from skcomplex.models.economics.markets import WasherElement
>>> GroupElement('workers', element=WasherElement, size=4).size
4

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

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

work(agent, model)

Work one second on the front car of an area.

Parameters:

Name	Type	Description	Default
agent	AgentView	The worker.	required
model	ModelView	The model.	required

Source code in src/skcomplex/models/economics/markets/_washer.py

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

    Args:
        agent:
            The worker.

        model:
            The model.
    """
    cars = model.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(model.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.elements import GroupElement
>>> from skcomplex.models.economics.markets import ZeroIntelligenceBuyerElement
>>> GroupElement('buyers', element=ZeroIntelligenceBuyerElement, size=5, values={'value': 100.0}).size
5

Source code in src/skcomplex/models/economics/markets/_zero_intelligence_buyer.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, model) 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
model	ModelView	The model view of the quotes.	required

Returns:

Name	Type	Description
bid	float	The bid of the round.

Source code in src/skcomplex/models/economics/markets/_zero_intelligence_buyer.py

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

    Args:
        agent:
            The buyer.

        model:
            The model view of the quotes.

    Returns:
        bid:
            The bid of the round.
    """
    return float(model.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.elements import GroupElement
>>> from skcomplex.models.economics.markets import ZeroIntelligenceSellerElement
>>> GroupElement('sellers', element=ZeroIntelligenceSellerElement, size=5, values={'value': 50.0}).size
5

Source code in src/skcomplex/models/economics/markets/_zero_intelligence_seller.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, model) 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
model	ModelView	The model view of the quotes.	required

Returns:

Name	Type	Description
ask	float	The ask of the round.

Source code in src/skcomplex/models/economics/markets/_zero_intelligence_seller.py

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

    Args:
        agent:
            The seller.

        model:
            The model view of the quotes.

    Returns:
        ask:
            The ask of the round.
    """
    bottom = agent.value if agent.constrained else 0.0
    return float(model.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 model, 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.models.economics.markets import barter
>>> barter().elements[0].check().prepare().size_
2

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

@register_model(
    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 model, 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.models.economics.markets import barter
        >>> barter().elements[0].check().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 model 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 model, 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 model are the orders <echelon>.orders and the shipments <echelon>.shipments of every echelon, and the orders of the customers customers.orders.

Examples:

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

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

@register_model(
    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 model 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 model, 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 model are the orders `<echelon>.orders` and the shipments
            `<echelon>.shipments` of every echelon, and the orders of the customers `customers.orders`.

    Examples:
        >>> from skcomplex.models.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 model, 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.models.economics.markets import brock_hommes
>>> brock_hommes(100).elements[0].check().prepare().size_
100

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

@register_model(
    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 model, 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.models.economics.markets import brock_hommes
        >>> brock_hommes(100).elements[0].check().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 model, with the cars on the lot dealer.inventory, and the flows dealer.deliveries and dealer.sales. The observables of the model are dealer.perceived for the perceived sales, dealer.desired_inventory, and dealer.orders.

Examples:

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

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

@register_model(
    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 model, with the cars on the lot `dealer.inventory`, and the flows `dealer.deliveries` and
            `dealer.sales`. The observables of the model are `dealer.perceived` for the perceived sales,
            `dealer.desired_inventory`, and `dealer.orders`.

    Examples:
        >>> from skcomplex.models.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 model, with the groups workers, cars, and till.

Examples:

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

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

@register_model(
    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 model, with the groups `workers`, `cars`, and `till`.

    Examples:
        >>> from skcomplex.models.economics.markets import car_wash
        >>> car_wash(5).elements[0].check().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.models.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/models/economics/markets/_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.models.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 model, with the market market.

Examples:

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

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

@register_model(
    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 model, with the market `market`.

    Examples:
        >>> from skcomplex.models.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 model, 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 model are market.coverage, market.price, market.expected_price, and market.capacity_orders.

Examples:

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

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

@register_model(
    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 model, 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 model are `market.coverage`, `market.price`, `market.expected_price`, and `market.capacity_orders`.

    Examples:
        >>> from skcomplex.models.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.models.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/models/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.models.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 model, with the group firms, the total output total, and the price price.

Examples:

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

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

@register_model(
    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 model, with the group `firms`, the total output `total`, and the price `price`.

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

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 model, 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.models.economics.markets import double_auction
>>> [group.size for group in double_auction(10, 8).elements]
[10, 8]

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

@register_model(
    field='economics/markets',
    tags=('market', 'agents', 'double auction', 'zero intelligence'),
    references=('Gooding 2019 ch. 8-12', '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.

    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 model, 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.models.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 Chesney's 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, as in the choice of Ajani. The share of fair food rises with the weight of ethics and falls with the premium, and social influence raises it.

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 model, with the group consumers, their logit choice shopping, and the share of fair food fair_share.

Examples:

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

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

@register_model(
    field='economics/markets',
    tags=('consumer choice', 'social influence', 'agents', 'grid'),
    references=('Chesney 2021 ch. 3', 'Chesney 2021 ch. 7'),
    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 Chesney's 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, as in the choice of Ajani. The
    share of fair food rises with the weight of ethics and falls with the premium, and social influence raises it.

    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 model, with the group `consumers`, their logit choice `shopping`, and the share of fair food
            `fair_share`.

    Examples:
        >>> from skcomplex.models.economics.markets import fair_food
        >>> fair_food(5, 4).elements[0].check().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(),
    )

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 toy and three-goods traders of Gooding (2019).

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 model, 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.models.economics.markets import goods_traders
>>> goods_traders(20, goods=2).elements[0].check().prepare().size_
20

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

@register_model(
    field='economics/markets',
    tags=('market', 'agents', 'barter', 'price convergence', 'grid'),
    references=('Gooding 2019 ch. 8-12', '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 toy and three-goods traders of Gooding (2019).

    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 model, 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.models.economics.markets import goods_traders
        >>> goods_traders(20, goods=2).elements[0].check().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, model)

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
model	ModelView	The model view of the market.	required

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

def hire(trade: MatchSpec, model: ModelView) -> 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.

        model:
            The model 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 = model.time, firm.posted

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 model, 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.models.economics.markets import labour_market
>>> [group.size for group in labour_market(20, 5).elements]
[5, 20]

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

@register_model(
    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 model, 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.models.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, model: 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.models.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/models/economics/markets/_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.models.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.models.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/models/economics/markets/_auction_ledger.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.models.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 model 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 model, 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 model, 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.engine.runtime import Simulation
>>> from skcomplex.models.economics.markets import oligopoly_equilibria
>>> trajectory = Simulation(oligopoly_equilibria()).check().prepare().run(1.0).trajectory_
>>> round(float(trajectory.values('cournot.q_i')[0]), 6)
3.333333

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

@register_model(
    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 model 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 model, 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 model, 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.engine.runtime import Simulation
        >>> from skcomplex.models.economics.markets import oligopoly_equilibria
        >>> trajectory = Simulation(oligopoly_equilibria()).check().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 model, with the groups customers and products, and the asymmetry m of the probabilities of visiting the two shops.

Examples:

>>> from skcomplex.models.economics.markets import perishable_goods_duopoly
>>> perishable_goods_duopoly(20).elements[0].check().prepare().size_
20

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

@register_model(
    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 model, with the groups `customers` and `products`, and the asymmetry `m` of the probabilities of
            visiting the two shops.

    Examples:
        >>> from skcomplex.models.economics.markets import perishable_goods_duopoly
        >>> perishable_goods_duopoly(20).elements[0].check().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 model, 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.models.economics.markets import produce_market
>>> [group.size for group in produce_market(10, 3).elements]
[3, 10]

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

@register_model(
    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 model, 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.models.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 model, with the price market.P.

Examples:

>>> from skcomplex.models.economics.markets import riccati_prices
>>> riccati_prices(gamma=-0.2).elements[0].gamma
-0.2

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

@register_model(
    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 model, with the price `market.P`.

    Examples:
        >>> from skcomplex.models.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 model, 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.models.economics.markets import shops
>>> [group.size for group in shops(4, width=5, height=5).elements]
[25, 4]

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

@register_model(
    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 model, 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.models.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, model: 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, model: agent.direction,
            step=lambda agent, model: agent.adjustment,
            lower=lambda agent, model: agent.unit_cost,
            form='agent',
            phase='accounts',
        ),
        LifeCycleInteraction('closure', target='shops', death=lambda agent, model: agent.money < 0, phase='accounts'),
    ]
    shopping = MatchingInteraction[ModelView](
        '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 model, with the groups patches and foragers, whose catalog observables are the population population and the Gini coefficient of wealth gini.

Examples:

>>> from skcomplex.models.economics.markets import sugarscape
>>> [group.size for group in sugarscape(10, 10, n_foragers=20).elements]
[100, 20]

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

@register_model(
    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 model, with the groups `patches` and `foragers`, whose catalog observables are the population
            `population` and the Gini coefficient of wealth `gini`.

    Examples:
        >>> from skcomplex.models.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')])),
    )

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.models.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/models/economics/markets/_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.models.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 model, 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.models.economics.markets import wilhite_trade_networks
>>> system = wilhite_trade_networks(40, topology='isolated', n_groups=4)
>>> system.elements[0].check().prepare().size_
40

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

@register_model(
    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 model, 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.models.economics.markets import wilhite_trade_networks
        >>> system = wilhite_trade_networks(40, topology='isolated', n_groups=4)
        >>> system.elements[0].check().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(),
    )