Bilateral two-good trade on complete, isolated, and small-world networks
Wilhite's bilateral barter of two goods among traders on one of four trade networks.
- Name:
wilhite_trade_networks - Field:
economics/markets - Formalism: agent-based model
- References: Wilhite 2001; Gooding 2019
Factory
@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(),
)
Components
class BarterTraderElement(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.
Args:
label:
The label of the agent or group.
visual:
How the element is drawn, or `None` for `default_visual_`.
good1:
The holding of the first good.
good2:
The holding of the second good.
mrs:
The marginal rate of substitution of the bundle.
links:
The number of trading partners the trader can search.
partners:
The number of partners that can match with the trader in a step, without limit.
Examples:
>>> from skcomplex.core.elements import GroupElement
>>> from skcomplex.models.economics.markets import BarterTraderElement
>>> GroupElement('traders', element=BarterTraderElement, size=4).size
4
"""
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
@classmethod
def _get_view(cls) -> type[BarterTraderView]:
"""Return the view of a trader, with its holdings, rate, and links."""
return BarterTraderView
@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()
@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))
@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
class MapInteraction(Interaction):
"""Map of a state of discrete time, whose next value is a function, or the state plus it.
In the mode `'replace'` the next value of the state is `f`, and in the mode `'increment'` it is the state plus `f`.
Under the `'synchronous'` schedule every member of a bound group updates at once, from the states at the start of
the step. Under the `'sequential'` schedule the members update one after the other, in their order, and under the
`'random-sequential'` schedule in a new random order every step, each member reading the values the members before
it set. Under the `'member'` schedule of the agent form, every member updates in its own activation, together with
the rules of its element and the other member interactions, in the order the system lists them.
In the `'symbolic'` form, `function(namespace)` returns a symbolic expression, or a vector of them for a vector
state, where `namespace.target` holds the states and parameters of the target, `namespace.sources` the namespaces
of the sources, and `namespace.t` the time, and may hold draws such as `normal('noise')`. In the `'agent'` form,
the target is an agent group and `function(agent, model)` returns the value as a number, or as an array of the
shape of a vector state, so it can read the agent, its neighbours, and the history and storage of the model view.
Args:
label:
The label of the interaction, unique among the parts of its system.
target:
The path of the element or group whose state the map sets.
state:
The name of the state of the target.
function:
The function `f(namespace)` in the symbolic form, or `f(agent, model)` in the agent form.
sources:
The paths of the elements, or members of groups such as `cells[0]`, that the function reads.
mode:
`'replace'` or `'increment'`.
schedule:
`'synchronous'`, `'sequential'`, `'random-sequential'`, or `'member'` for the agent form.
form:
`'symbolic'` for update laws of the states of the target, or `'agent'` for a rule over an agent group.
phase:
The phase of the step the agent form runs in, or `None` for the default phase.
Raises:
ValidationError:
The mode, the schedule, or the form is unknown, the symbolic form has the schedule `'member'`, the
function is not callable, or the agent form returns a value that does not match the state, raised by
`run` of the simulation.
Examples:
>>> from skcomplex.core.elements import Element
>>> from skcomplex.core.interactions import MapInteraction
>>> from skcomplex.core.spaces import DiscreteTime
>>> from skcomplex.core.system import System
>>> from skcomplex.engine.runtime import Simulation
>>> from skcomplex.specs import StateSpec
>>> from typing import Annotated, SupportsFloat
>>> class ValueElement(Element):
... def __init__(self, label, x: Annotated[float, StateSpec()] = 0.5):
... super().__init__(label)
... self.x = x
>>> logistic = MapInteraction(
... 'map', target='value', state='x', function=lambda s: 3 * s.target.x * (1 - s.target.x)
... )
>>> system = System(elements=[ValueElement('value')], interactions=[logistic], time=DiscreteTime())
>>> Simulation(system).check().prepare().run(until=2).trajectory_.values('value.x').tolist()
[0.5, 0.75, 0.5625]
>>> agents = MapInteraction('map', target='walkers', state='x', function=lambda a, m: a.x + 1, form='agent')
>>> type(agents.check().prepare().run(SimpleNamespace()).laws_[0]).__name__
'RuleLaw'
>>> from skcomplex.core.elements import GroupElement
>>> class WeightsElement(Element):
... def __init__(self, label, visual=None, *, w: Annotated[object, StateSpec(vector=True)] = 0.0):
... super().__init__(label, visual=visual)
... self.w = w
>>> group = GroupElement('weights', element=WeightsElement, size=1, values={'w': [1.0, 2.0]})
>>> doubling = MapInteraction(
... 'double', target='weights', state='w', function=lambda a, m: [2 * x for x in a.w], form='agent'
... )
>>> trajectory = Simulation(System(elements=[group], interactions=[doubling])).check().prepare().run(until=2)
>>> trajectory.trajectory_.agents('weights', 2).values['w'].tolist()
[[4.0, 8.0]]
"""
def __init__(
self,
label: str,
*,
target: Annotated[str, RoleSpec()],
state: str,
function: Callable[..., object],
sources: Annotated[Sequence[str], RoleSpec(variadic=True)] = (),
mode: str = 'replace',
schedule: str = 'synchronous',
form: str = 'symbolic',
phase: str | None = None,
) -> None:
super().__init__(label)
self.target = target
self.state = state
self.function = function
self.sources = sources
self.mode = mode
self.schedule = schedule
self.form = form
self.phase = phase
def _check(self) -> None:
"""Raise an error for an unknown mode, schedule, or form, or a function that cannot be called.
Raises:
ValidationError:
The mode, the schedule, or the form is unknown, the symbolic form has the schedule `'member'`, or the
function is not callable.
"""
super()._check()
self._check_variant('mode', self.mode, MODES, (BaseSpec,))
for name, value, variants in (
('schedule', self.schedule, SCHEDULES),
('form', self.form, FORMS),
):
if value not in variants:
error_msg = f'The {name} of map {self.label!r} is {value!r}. Use one of {", ".join(variants)}.'
raise ValidationError(error_msg)
if self.schedule == 'member' and self.form != 'agent':
error_msg = (
f'The map {self.label!r} updates every member in its activation, which only the agent form does.'
)
raise ValidationError(error_msg)
if not callable(self.function):
error_msg = f'The function of map {self.label!r} must be callable, but it is {self.function!r}.'
raise ValidationError(error_msg)
def _apply_mode(self, current: Value, value: Value) -> Value:
"""Return the next value of a state from its current value and the value of the function, by the mode."""
if self.mode == 'increment':
return current + value
if self.mode == 'replace':
return value
raise self._get_variant_error(self.mode, 'mode', '_apply_mode')
def _get_vector(self, agent: AgentViewProtocol, model: ModelViewProtocol) -> Value:
"""Return the next value of a vector state of an agent."""
current = np.asarray(getattr(agent, self.state), dtype=np.float64)
value = np.asarray(self.function(agent, model), dtype=np.float64)
if value.shape != current.shape:
error_msg = (
f'The function of map {self.label!r} returns the shape {value.shape} for the state {self.state!r} of '
f'the shape {current.shape}. Return an array of the shape of the state.'
)
raise ValidationError(error_msg)
return self._apply_mode(current, value)
def _get_value(self, agent: AgentViewProtocol, model: ModelViewProtocol) -> Value:
"""Return the next value of the state of an agent, a number or the array of a vector state."""
if np.ndim(getattr(agent, self.state)) > 0:
return self._get_vector(agent, model)
found = self.function(agent, model)
if not isinstance(found, SupportsFloat):
error_msg = f'The function of map {self.label!r} returns {found!r} for an agent. Return a number.'
raise ValidationError(error_msg)
value = float(found)
return self._apply_mode(float(getattr(agent, self.state)), value)
def _set_value(self, agent: AgentViewProtocol, value: Value) -> None:
"""Set the state of an agent to its next value."""
setattr(agent, self.state, value)
def _update_member(self, agent: AgentViewProtocol, model: ModelViewProtocol) -> None:
"""Set the state of one member to its next value."""
if agent.alive:
self._set_value(agent, self._get_value(agent, model))
def update(self, model: ModelViewProtocol) -> None:
"""Set the state of every agent of the target to its next value, by the schedule.
Args:
model:
The model view of the interaction.
"""
self._update_agents(model, self.schedule, self._get_value, self._set_value)
def _get_order(self) -> object:
"""Return the order of the members of the symbolic form: none, their own order, or a uniform draw."""
if self.schedule == 'synchronous':
return None
if self.schedule == 'sequential':
return SR(0)
return uniform(re.sub(r'\W', '_', f'order_{self.label}'))
def _get_laws(self, namespace: SimpleNamespace) -> list[BasePart]:
"""Return the update of the state to its next value, or the rule of the agent form.
Args:
namespace:
The namespace of the symbols the laws can use, with `namespace.target` and `namespace.sources`.
Returns:
laws:
The update law of the state, or the rule that runs the map of the agents.
"""
if self.form == 'agent' and self.schedule == 'member':
return [RuleLaw(self._update_member, phase=self.phase, members=self._get_role('target'))]
if self.form == 'agent':
return [RuleLaw(self.update, phase=self.phase)]
value = getattr(self._get_member(namespace, 'target'), self.state)
mapped = cast(Value, self.function(namespace))
next_value = self._apply_mode(value, mapped)
return [UpdateLaw(value, next_value, increment=self.mode == 'increment', order=self._get_order())]
class BilateralBarterInteraction(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.
Args:
label:
The label of the interaction.
requesters:
The path of the group of traders that search for a partner.
providers:
The path of the group of traders that can be found, the same group.
rule:
The spec of the algorithm that pairs the traders, such as `PartnerSpec('local', clearing='provider')`.
limit:
The state with the limit of both sides, missing for traders, which then accept any partner.
quantity:
The states of the partners a requester wants and a provider takes, one and without limit for traders.
allowance:
The state of a requester with the most value it can give, missing for traders.
transfers:
The states every match moves, none for traders, whose trades move the goods.
phase:
The phase the barter runs in, or `None` for the default phase.
on_match:
A function of every match and the model view, run after the trade, or `None`.
units:
The most units one pair trades in a step.
Examples:
>>> from skcomplex.models.economics.markets import BilateralBarterInteraction
>>> BilateralBarterInteraction('barter', requesters='traders', providers='traders').units
100
"""
reports: ClassVar[tuple[str, ...]] = (*MatchingInteraction.reports, 'trades', 'searches')
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
@staticmethod
def _barter(first: BarterTraderView, second: BarterTraderView, units: int) -> int:
"""Trade single units of the first good at the bundle price while both traders gain, and return the units."""
buyer, seller = (first, second) if first.good2 * second.good1 > second.good2 * first.good1 else (second, first)
price = (buyer.good2 + seller.good2) / (buyer.good1 + seller.good1)
traded = 0
while traded < units:
buyer_gains = (buyer.good1 + 1) * (buyer.good2 - price) > buyer.good1 * buyer.good2
seller_gains = (seller.good1 - 1) * (seller.good2 + price) > seller.good1 * seller.good2
if not (buyer_gains and seller_gains):
break
transfer(seller, buyer, 1.0, 'good1')
transfer(buyer, seller, price, 'good2')
traded += 1
return traded
def _settle(self, match: MatchSpec, model: ModelView) -> None:
"""Trade along a pair of partners, count the units traded, and settle the match."""
first, second = match.requester, match.provider
if not isinstance(first, BarterTraderView) or not isinstance(second, BarterTraderView):
super()._settle(match, model)
return
if BarterTraderElement.is_partner(first, second):
traded = self._barter(first, second, int(model.parameter('units')))
model.storage['trades'] = self._get_trades(model) + traded
super()._settle(match, model)
@staticmethod
def _get_trades(model: ModelView) -> float:
"""Return the units traded so far in the step, kept in the storage of the barter."""
trades = model.storage.get('trades')
return float(trades) if isinstance(trades, int | float) else 0.0
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)))