TAP growth with random catalysis and autocatalytic sets
The TAP growth of goods from combinations of goods, with random catalysis of their production functions.
- Name:
tap_catalysis - Field:
economics/innovation - Formalism: jump process
- References: Hordijk and Kauffman 2022
Select this system
from skcomplex.systems import create_simulation, create_system, describe
system = create_system('tap_catalysis')
entry = describe('tap_catalysis')
trajectory = create_simulation('tap_catalysis').prepare().run().trajectory_
Factory
@register_system(
field='economics/innovation',
tags=('innovation', 'adjacent possible', 'autocatalytic sets', 'hypergraph', 'jump process'),
references=('Hordijk and Kauffman 2022',),
defaults={'until': 400},
)
def tap_catalysis(
m0: int = 10,
*,
parents: int = 4,
alpha: float = 0.01,
goods: int = 1250,
pc: float = 0.0044,
) -> System:
"""Return the TAP growth of goods from combinations of goods, with random catalysis of their production functions.
The TAP model with catalysis of Hordijk and Kauffman (2022, Algorithm 1), in continuous time. From `m0` goods,
new goods made from `i` parents drawn uniformly among the `M` goods appear at the rate `alpha^i C(M, i)`, for
every `i` up to `parents`, until there are `goods` goods. The production function of a new good is a hyperedge of
its parents and the good, its last member. A new good catalyses the production of every other good, and every
other good catalyses its production, each with the probability `pc`, as directed edges from the catalyst to the
good. Algorithm 1 draws a Poisson number of goods of every order per step from the rates at its start, and the
process here draws them one at a time from the rates at that time. The initial goods are the food set, and
`AutocatalyticSetAnalysis` finds the largest autocatalytic set of the production functions, which exists in about
half the runs at `pc = 0.0044` and then tends to hold most of the network.
Args:
m0:
The number of initial goods, the food set.
parents:
The largest number of parents of a good, `K`.
alpha:
The base of the probabilities `alpha^i` that a combination of `i` goods makes a new one.
goods:
The number of goods at which the growth stops, `M`.
pc:
The catalysis probability.
Returns:
system:
The system, with the goods `goods`, the catalysis `catalysis`, and the production functions `functions`.
Examples:
>>> from skcomplex.systems.economics.innovation import tap_catalysis
>>> [element.label for element in tap_catalysis(goods=50).elements]
['goods', 'catalysis', 'functions']
"""
growth = [
TopologyInteraction(
f'order_{order}',
target='goods',
action='add-node',
rate=_combination_rate(order, alpha, goods),
hyperedge=order + 1,
independent=True,
direction='both',
toward=lambda node: pc,
)
for order in range(1, parents + 1)
]
links = int(3 * pc * goods * goods) + goods
return System(
elements=[
GroupElement('goods', element=Element, on='sites', variable=True, capacity=goods),
GroupElement('catalysis', element=Element, on='edges', variable=True, capacity=links),
GroupElement(
'functions',
element=Element,
on='hyperedges',
variable=True,
size=0,
capacity=goods,
rank=parents + 1,
),
],
interactions=growth,
space=NetworkSpace(nx.empty_graph(m0, create_using=nx.DiGraph)),
)
Components
Elements:
Element
Interactions:
TopologyInteraction
class Element(BaseElement):
"""Base class of entities that carry state, such as particles, populations, agents, or firms.
A subclass takes its states, its parameters, and `visual` as arguments of its own `__init__`, annotating each state
with a `StateSpec` and each parameter with a `ParameterSpec` inside `Annotated`, and returns its intrinsic dynamics
from `_get_laws`, which returns no laws by default. A subclass drawn in its own way gives `visual` a default in its
signature.
An element whose laws have the agent trait, such as rules, alone or in a group, is an agent: under discrete time it
runs on the agent stepper, whose rules receive every member as an agent with its states as attributes. Its states
then hold any value, not only numbers. A member is removed only when its states tagged with a quantity are settled
to zero, and its states and parameters cannot take the names of the agent view, such as `location` or `alive`, or
of the attributes Mesa reads, such as `position` or `system`.
A subclass that connects to other elements declares its ports in the class variable `ports`, a mapping from the name
of a port to its `ConnectorSpec`, and takes one algebraic state per variable of a port, named `<port>_<variable>`.
A flow is positive into the element.
Args:
label:
The label of the element, unique among the parts of its system.
visual:
How the element is drawn, or `None` for `default_visual_`.
Attributes:
default_visual_:
The visual of the element when it sets none.
Examples:
>>> from typing import Annotated
>>> from skcomplex.core.components.elements import Element
>>> from skcomplex.core.specs import StateSpec
>>> from skcomplex.core.specs import VisualSpec
>>> from sage.all__sagemath_symbolics import oo
>>> from sage.sets.real_set import RealSet
>>> class PopulationElement(Element):
... def __init__(
... self,
... label,
... n: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
... visual=VisualSpec('line'),
... ):
... super().__init__(label, visual=visual)
... self.n = n
>>> PopulationElement('rabbits', n=40)
PopulationElement(label='rabbits', n=40)
"""
ports: ClassVar[Mapping[str, ConnectorSpec]] = {}
@classmethod
def _get_view(cls) -> type | None:
"""Return the class of the views of the element's agents, or `None` for the engine's plain view.
An agent element names a subclass of the engine's `AgentView` that annotates its states and parameters, such as
`class WordView(AgentView): count: float`, so its rules read them with checked types.
"""
return None
def _get_laws(self, namespace: SimpleNamespace) -> Sequence[BasePart]:
"""Return the laws of the element for a namespace of symbols, none by default.
Args:
namespace:
The namespace of the symbols the laws can use.
Returns:
laws:
The laws of the element.
"""
return []
class TopologyInteraction(NeighborInteraction):
"""Change of the topology of a network: nodes and edges that are added or removed, and edges that are rewired.
The action runs for every node of the group a role binds, `'add-node'`, `'remove-node'`, `'replace-node'`, which
removes the node and adds a new one, and `'add-edge'`, or for every edge of its network, `'remove-edge'` and
`'rewire'`. Without a target, `'add-node'` runs once per step, adding one node from outside the network to the one
network whose groups are variable, as in growing networks. It fires at a `rate`, with a `probability` per step, or
with probability one, and only where the `condition` holds. The rate, the probability, and the condition are numbers
or functions of the namespace of an edge, whose ends are `namespace.source` and `namespace.target`, and an action of
the nodes reads the node as the source. Without a target, the namespace of `'add-node'` holds the values drawn for
the node being added as `namespace.source`, which the condition and a probability read and a rate does not. A new
node joins `edges` other nodes, a node adding edges joins `edges` other nodes, and a rewiring keeps the end `keep`
of its edge. New ends are drawn with probability proportional to `toward(node)`, a function of the namespace of a
candidate node that also holds the node the new edge starts from as `node.source`, or uniformly. That node is the
new node of `'add-node'`, the adding node of `'add-edge'`, and the kept end of `'rewire'`. Where fewer candidates
have a positive weight than the edges to draw, the new ends are drawn uniformly. A new node or edge takes the
`values`, each a value or a function of the random number generator that draws it. The groups on the nodes and the
edges of the network are variable. With `hyperedge`, a new node instead forms a hyperedge of that size with
`hyperedge - 1` existing nodes and joins each of them by an edge: the first two are the ends of an edge drawn
uniformly, so a size of three attaches a triangle along an edge, and further members are drawn by `toward`. The
new hyperedge is a new member of the variable group on the hyperedges of the network. With `edges=0` or
`independent`, every other member is drawn by `toward`, and the new node draws its edges apart. The action
`'add-hyperedge'` runs for every node of the group on the nodes, which forms a hyperedge of size `hyperedge` with
nodes drawn by `toward` and is its last member, and `'remove-hyperedge'` runs for every hyperedge of the variable
group on the hyperedges that the target binds, whose namespace reads the hyperedge member as the source. With
`independent`, every candidate becomes a new end on its own with probability `toward`, so a node joins a random
number of nodes, and on a directed layer `direction` sends the new edges out of the node, into it, or both ways.
`'close-triangle'` joins two neighbours of the node that are not yet joined, or a node without such a pair to
`edges` nodes, `'contract-edge'` merges a neighbour drawn by `toward` into the node, which keeps the
union of their links, and `'duplicate-node'` adds a copy of the node that inherits each of its links with
probability `inherit` and joins it with probability `parent`. A replacement removes the node and adds the new one
in one request. With `candidates`, `'add-edge'` adds for every new edge the candidate of the smallest `score`, and
without a target it runs once per step between two nodes drawn uniformly. With `one`, the action runs at every step
for one member drawn uniformly among those present, firing with its probability there.
Args:
label:
The label of the interaction, unique among the parts of its system.
target:
The path of the group on the nodes of the network, of the variable group on its hyperedges for
`'remove-hyperedge'`, or `None` for an addition of one node or of edges between drawn nodes per step.
action:
The change, `'add-node'`, `'remove-node'`, `'replace-node'`, `'add-edge'`, `'remove-edge'`, `'rewire'`,
`'add-hyperedge'`, `'remove-hyperedge'`, `'close-triangle'`, `'contract-edge'`, or `'duplicate-node'`.
rate:
The rate of the change, a number or a function, or `None` for a probability per step.
probability:
The probability of the change per step, a number or a function, or `None` for probability one.
condition:
A function of the namespace that returns where the change may fire, or `None` everywhere.
toward:
A function of the namespace of a candidate node that returns its weight as a new end, reading the node
the new edge starts from as `namespace.source`, or `None` for equal weights.
edges:
The number of edges a new node or a node adding edges joins.
keep:
The end of an edge a rewiring keeps, `'source'` or `'target'`.
values:
The initial states of a new node or edge by name, each a value or a function of the random number
generator that draws it, which default to the defaults of the element type.
layer:
The layer of the network whose edges change, or `None` for the graph of the network.
hyperedge:
The size of the hyperedge every new node or every node of `'add-hyperedge'` forms with existing nodes, or
`None` for none.
independent:
Whether every candidate becomes a new end on its own with the probability `toward`, in place of `edges`
ends drawn by weight.
direction:
The direction of new edges on a directed layer, `'out'` of the node, `'in'` to it, or `'both'`.
candidates:
The number of candidate edges of `'add-edge'` for every edge it adds.
score:
A function of the namespace of a candidate end that returns the score of its edge, reading the other end
as `namespace.source`, whose smallest value is added, or `None` for the first candidate.
one:
Whether the action runs at every step for one member drawn uniformly among those present.
inherit:
The probability that the copy of `'duplicate-node'` inherits each link of the node.
parent:
The probability that the copy of `'duplicate-node'` joins the node.
Raises:
ValidationError:
The action, the kept end, or the direction is unknown, both a rate and a probability are given, the
number of edges is not a whole number, an action other than `'add-node'` and `'add-edge'` has no target, a
hyperedge is given to an action other than `'add-node'` and `'add-hyperedge'` or has fewer than two
members, `'add-hyperedge'` has no hyperedge, candidates or a score belong to another action than
`'add-edge'`, or a probability of duplication lies outside `[0, 1]`, raised by `prepare`.
Examples:
>>> import networkx as nx
>>> from skcomplex.core.components.elements import Element, GroupElement
>>> from skcomplex.core.components.interactions import TopologyInteraction
>>> from skcomplex.core.components.spaces import NetworkSpace
>>> from skcomplex.core.components.system import System
>>> from skcomplex.core.engine.runtime import Simulation
>>> nodes = GroupElement('nodes', element=Element, on='sites', variable=True, capacity=20)
>>> links = GroupElement('links', element=Element, on='edges', variable=True, capacity=20)
>>> expiry = TopologyInteraction('expiry', target='nodes', action='remove-edge')
>>> system = System(elements=[nodes, links], interactions=[expiry], space=NetworkSpace(nx.cycle_graph(5)))
>>> trajectory = Simulation(system, random_state=0).prepare().run(until=1).trajectory_
>>> len(trajectory.network(0).sources), len(trajectory.network(1).sources)
(5, 0)
A network that grows by triangles from one triangle:
>>> import xgi
>>> from skcomplex.core.components.spaces import DiscreteTime
>>> triangles = GroupElement('triangles', element=Element, on='hyperedges', variable=True, capacity=8)
>>> growth = TopologyInteraction('growth', action='add-node', hyperedge=3)
>>> space = NetworkSpace(nx.complete_graph(3), hyperedges=[[0, 1, 2]])
>>> system = System(
... elements=[nodes, links, triangles], interactions=[growth], space=space, time=DiscreteTime()
... )
>>> trajectory = Simulation(system, random_state=0).prepare().run(until=5).trajectory_
>>> trajectory.network(-1).hyperedges.shape
(6, 3)
"""
law_classes: ClassVar[tuple[type[BaseLaw], ...]] = (JumpLaw, UpdateLaw)
def __init__(
self,
label: str,
*,
target: Annotated[str | None, RoleSpec(optional=True, elements=(ElementKind.ANY,))] = None,
action: str = 'add-edge',
rate: Chance | None = None,
probability: Chance | None = None,
condition: Callable[[SimpleNamespace], object] | None = None,
toward: Callable[[SimpleNamespace], object] | None = None,
edges: int = 1,
keep: str = 'source',
values: Mapping[str, object] | None = None,
layer: str | None = None,
hyperedge: int | None = None,
independent: bool = False,
direction: str = 'out',
candidates: int = 1,
score: Callable[[SimpleNamespace], object] | None = None,
one: bool = False,
inherit: float = 1.0,
parent: float = 0.0,
) -> None:
super().__init__(label, layer=layer)
self.target = target
self.action = action
self.rate = rate
self.probability = probability
self.condition = condition
self.toward = toward
self.edges = edges
self.keep = keep
self.values = values
self.hyperedge = hyperedge
self.independent = independent
self.direction = direction
self.candidates = candidates
self.score = score
self.one = one
self.inherit = inherit
self.parent = parent
def _check(self) -> None:
"""Raise an error for an unknown action or kept end, a rate next to a probability, or a bad number of edges.
Raises:
ValidationError:
The action or the kept end is unknown, both a rate and a probability are given, a condition, a
weight, or a score cannot be called, the number of edges is not a whole number of at least 0, an
action other than `'add-node'` and `'add-edge'` has no target, a hyperedge is given to an action other
than `'add-node'` and `'add-hyperedge'`, or `'add-hyperedge'` has none, or the request is invalid.
"""
super()._check()
self._check_variant('action', self.action, ACTIONS, (BaseSpec,))
if self.rate is not None and self.probability is not None:
error_msg = f'{self.label!r} fires at a rate and with a probability. Give one of them.'
raise ValidationError(error_msg)
if self.target is None and self.action not in {'add-node', 'add-edge'}:
error_msg = f'{self.label!r} runs {self.action!r} without a target. Give the group on the nodes.'
raise ValidationError(error_msg)
if self.hyperedge is not None and self.action not in {'add-node', 'add-hyperedge'}:
error_msg = (
f'{self.label!r} runs {self.action!r} with a hyperedge. Only new nodes and new hyperedges take a size.'
)
raise ValidationError(error_msg)
for name in ('condition', 'toward', 'score'):
value = getattr(self, name)
if value is not None and not callable(value):
error_msg = f'The {name} of {self.label!r} is {value!r}. Give a function of the namespace, or None.'
raise ValidationError(error_msg)
for request in self._get_requests():
request._validated()
def _get_requests(self) -> list[TopologyRequestSpec]:
"""Return the request of the action."""
if self.action not in ACTIONS:
raise self._get_variant_error(self.action, 'action', '_get_requests')
return [
TopologyRequestSpec(
self.action,
self.edges,
self.toward,
self.keep,
self.layer,
tuple((self.values or {}).items()),
self.hyperedge if self.action in {'add-node', 'add-hyperedge'} else None,
self.independent,
self.direction,
self.candidates,
self.score,
self.one,
self.inherit,
self.parent,
),
]
@staticmethod
def _evaluate(value: Chance, namespace: SimpleNamespace) -> object:
"""Return a number, or the value of a function of the namespace."""
return value(namespace) if callable(value) else value
def _get_laws(self, namespace: SimpleNamespace) -> list[BaseLaw]:
"""Return the requests of the action, fired at the rate or with the probability where the condition holds.
Args:
namespace:
The namespace of an edge, with `namespace.source` and `namespace.target`.
Returns:
laws:
A jump law of the request at a rate, or else an update law of the request with its probability.
"""
chance: object = 1
if self.rate is not None:
chance = self._evaluate(self.rate, namespace)
elif self.probability is not None:
chance = self._evaluate(self.probability, namespace)
if self.condition is not None:
chance = where(self.condition(namespace), chance, 0)
if self.rate is not None:
return [JumpLaw(chance, {request: 1}) for request in self._get_requests()]
return [UpdateLaw(request, chance) for request in self._get_requests()]