Stock-flow consistent monetary economy
A stock-flow consistent economy, a closed circuit of money between households, firms, and a bank.
- Name:
stock_flow_consistent_economy - Field:
economics/macroeconomics - Formalism: discrete map
- References: Godley and Lavoie 2007; Caiani et al. 2016
Factory
@register_model(
field='economics/macroeconomics',
tags=('stocks and flows', 'stock-flow consistent', 'money', 'discrete map'),
references=('Godley and Lavoie 2007', 'Caiani et al. 2016'),
defaults={'until': 100, 'dt': 1},
)
def stock_flow_consistent_economy(
*,
household_deposits: float = 60.0,
firm_deposits: float = 20.0,
bank_deposits: float = 10.0,
loans: float = 50.0,
sales: float = 40.0,
alpha_income: float = 0.6,
alpha_wealth: float = 0.4,
wage_share: float = 0.7,
interest: float = 0.03,
payout: float = 0.5,
distribution: float = 0.3,
credit: float = 0.1,
repayment: float = 0.08,
) -> System:
"""Return a stock-flow consistent economy, a closed circuit of money between households, firms, and a bank.
Money is the deposits of the three sectors, tagged as the quantity `money`, and every transaction of a period is
a flow, so each payment is one sector's outflow and another's inflow. Households consume a share `alpha_income` of
their wages and a share `alpha_wealth` of their deposits, firms pay the wage share of last period's sales and the
interest on their loans, and distribute a share of their deposits as profits, and the bank pays out a share of its
funds as dividends. Lending creates money and
repayment destroys it, the only flows with the environment, so the total money changes exactly by lending minus
repayment, in the spirit of the models of Godley and Lavoie (2007) and the balance sheets of Caiani et al. (2016).
Loans approach `credit * sales / repayment` and sales settle where the stocks stop growing. The circuit moves the
money of a period, and the maps `loans` and `sales` set the loans and the sales of the firms.
Args:
household_deposits:
The initial deposits of the households.
firm_deposits:
The initial deposits of the firms.
bank_deposits:
The initial own funds of the bank.
loans:
The initial loans of the firms.
sales:
The sales of the previous period.
alpha_income:
The propensity to consume out of wages.
alpha_wealth:
The propensity to consume out of deposits.
wage_share:
The share of sales paid as wages.
interest:
The interest rate on loans per period.
payout:
The share of the bank's funds paid out as dividends per period.
distribution:
The share of the firms' deposits distributed to households as profits per period.
credit:
The new loans per unit of sales.
repayment:
The share of loans repaid per period.
Returns:
system:
The model, with the deposits `households.deposits`, `firms.deposits`, and `bank.deposits`, the loans
`firms.loans`, the sales `firms.sales`, the flows `circuit.consumption`, `circuit.wages`,
`circuit.interest`, `circuit.dividends`, `circuit.profits`, `circuit.lending`, and `circuit.repayments`,
the maps `loans` and `sales`, and the net money created `__boundary.money`.
Examples:
>>> from skcomplex.models.economics.macroeconomics import stock_flow_consistent_economy
>>> [element.label for element in stock_flow_consistent_economy().elements]
['households', 'firms', 'bank']
"""
return System(
elements=[
HouseholdsElement(
'households',
deposits=household_deposits,
alpha_income=alpha_income,
alpha_wealth=alpha_wealth,
),
FirmsElement(
'firms',
deposits=firm_deposits,
loans=loans,
sales=sales,
wage_share=wage_share,
distribution=distribution,
repayment=repayment,
),
MonetaryBankElement('bank', deposits=bank_deposits, interest=interest, payout=payout, credit=credit),
],
interactions=[
CircuitInteraction('circuit', households='households', firms='firms', bank='bank'),
MapInteraction('loans', target='firms', state='loans', function=_next_loans, sources=['bank']),
MapInteraction('sales', target='firms', state='sales', function=_next_sales, sources=['households']),
],
)
Components
class HouseholdsElement(Element):
"""Households sector, whose deposits are money, and which consumes out of its wages and its deposits.
Args:
label:
The label of the sector.
visual:
How the element is drawn, or `None` for `default_visual_`.
deposits:
The deposits of the households.
alpha_income:
The propensity to consume out of wages.
alpha_wealth:
The propensity to consume out of deposits.
Examples:
>>> from skcomplex.models.economics.macroeconomics import HouseholdsElement
>>> HouseholdsElement('households').deposits
60.0
"""
def __init__(
self,
label: str,
visual: VisualSpec | None = None,
*,
deposits: Annotated[object, StateSpec(domain=RealSet.closed_open(0, oo), quantity='money')] = 60.0,
alpha_income: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.6,
alpha_wealth: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.4,
) -> None:
super().__init__(label, visual=visual)
self.deposits = deposits
self.alpha_income = alpha_income
self.alpha_wealth = alpha_wealth
class FirmsElement(Element):
"""Firms sector, with its deposits, its loans, and the sales of the last period, which pays wages and profits.
Args:
label:
The label of the sector.
visual:
How the element is drawn, or `None` for `default_visual_`.
deposits:
The deposits of the firms, which are money.
loans:
The loans of the firms.
sales:
The sales of the last period.
wage_share:
The share of sales paid as wages.
distribution:
The share of the firms' deposits distributed as profits per period.
repayment:
The share of loans repaid per period.
Examples:
>>> from skcomplex.models.economics.macroeconomics import FirmsElement
>>> FirmsElement('firms').loans
50.0
"""
def __init__(
self,
label: str,
visual: VisualSpec | None = None,
*,
deposits: Annotated[object, StateSpec(domain=RealSet.closed_open(0, oo), quantity='money')] = 20.0,
loans: Annotated[object, StateSpec(domain=RealSet.closed_open(0, oo))] = 50.0,
sales: Annotated[object, StateSpec(domain=RealSet.closed_open(0, oo))] = 40.0,
wage_share: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.7,
distribution: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.3,
repayment: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.08,
) -> None:
super().__init__(label, visual=visual)
self.deposits = deposits
self.loans = loans
self.sales = sales
self.wage_share = wage_share
self.distribution = distribution
self.repayment = repayment
class MonetaryBankElement(Element):
"""Bank of a stock-flow consistent economy, whose own funds are money, and which lends and charges interest.
Args:
label:
The label of the bank.
visual:
How the element is drawn, or `None` for `default_visual_`.
deposits:
The own funds of the bank.
interest:
The interest rate on loans per period.
payout:
The share of the bank's funds paid out as dividends per period.
credit:
The new loans per unit of sales.
Examples:
>>> from skcomplex.models.economics.macroeconomics import MonetaryBankElement
>>> MonetaryBankElement('bank').deposits
10.0
"""
def __init__(
self,
label: str,
visual: VisualSpec | None = None,
*,
deposits: Annotated[object, StateSpec(domain=RealSet.closed_open(0, oo), quantity='money')] = 10.0,
interest: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.03,
payout: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.5,
credit: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.1,
) -> None:
super().__init__(label, visual=visual)
self.deposits = deposits
self.interest = interest
self.payout = payout
self.credit = credit
class CircuitInteraction(TransferInteraction):
"""Circuit of money between households, firms, and a bank, one period per step.
The flows of a period are the legs of one transfer between the deposits of the sectors, consumption, wages,
interest, dividends, and profits, lending from the environment, and repayments to it, each reading the parameters
of the sectors. Every leg is named by its flow, so the reporters `<label>.<flow>` record it.
Args:
label:
The label of the circuit.
households:
The path of the households.
firms:
The path of the firms.
bank:
The path of the bank.
Examples:
>>> from skcomplex.models.economics.macroeconomics import CircuitInteraction
>>> CircuitInteraction('circuit', households='households', firms='firms', bank='bank').bank
'bank'
"""
def __init__(
self,
label: str,
*,
households: Annotated[str, RoleSpec()],
firms: Annotated[str, RoleSpec()],
bank: Annotated[str, RoleSpec()],
) -> None:
super().__init__(
label,
source=households,
target=firms,
state='households.deposits',
into='firms.deposits',
amount=lambda s: CircuitInteraction._consumption(s.households, s.firms),
flow='consumption',
legs=[
('firms.deposits', 'households.deposits', lambda s: s.firms.wage_share * s.firms.sales, 'wages'),
('firms.deposits', 'bank.deposits', lambda s: s.bank.interest * s.firms.loans, 'interest'),
('bank.deposits', 'households.deposits', lambda s: s.bank.payout * s.bank.deposits, 'dividends'),
(
'firms.deposits',
'households.deposits',
lambda s: s.firms.distribution * s.firms.deposits,
'profits',
),
(
Source('lending'),
'firms.deposits',
lambda s: CircuitInteraction._lending(s.firms, s.bank),
'lending',
),
('firms.deposits', Sink('repayment'), lambda s: s.firms.repayment * s.firms.loans, 'repayments'),
],
)
self.households = households
self.firms = firms
self.bank = bank
def _get_holders(self, namespace: SimpleNamespace) -> list[tuple[SimpleNamespace, SimpleNamespace]]:
"""Return the households and the firms, the first pair of the circuit, whose other legs name their roles."""
return [(namespace.households, namespace.firms)]
@staticmethod
def _consumption(households: SimpleNamespace, firms: SimpleNamespace) -> Expression:
"""Return the consumption of a period, a share of the wages and a share of the deposits of the households."""
return households.alpha_income * firms.wage_share * firms.sales + households.alpha_wealth * households.deposits
@staticmethod
def _lending(firms: SimpleNamespace, bank: SimpleNamespace) -> Expression:
"""Return the new loans of a period, in proportion to the sales of the firms."""
return bank.credit * firms.sales
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())]
Equations
households.deposits: deposits_households(t + 1) == -alpha_income_households*wage_share_firms*sales_firms(t) + payout_bank*deposits_bank(t) + distribution_firms*deposits_firms(t) - alpha_wealth_households*deposits_households(t) + wage_share_firms*sales_firms(t) + deposits_households(t)
firms.deposits: deposits_firms(t + 1) == alpha_income_households*wage_share_firms*sales_firms(t) - distribution_firms*deposits_firms(t) + alpha_wealth_households*deposits_households(t) - interest_bank*loans_firms(t) - repayment_firms*loans_firms(t) + credit_bank*sales_firms(t) - wage_share_firms*sales_firms(t) + deposits_firms(t)
firms.loans: loans_firms(t + 1) == -repayment_firms*loans_firms(t) + credit_bank*sales_firms(t) + loans_firms(t)
firms.sales: sales_firms(t + 1) == alpha_income_households*wage_share_firms*sales_firms(t) + alpha_wealth_households*deposits_households(t)
bank.deposits: deposits_bank(t + 1) == -payout_bank*deposits_bank(t) + interest_bank*loans_firms(t) + deposits_bank(t)