Skip to content

Stochastic processes

Stochastic processes: random walks, Markov sources, counting, urn, branching, and sample-space-reducing processes.

AgeingWalkerElement(label, visual=VisualSpec('agents', {'color_by': 'x'}), *, x=0.0, ups=0.0, p=0.5, nu=1.0)

Bases: Element

Walker that steps up with a probability that decays with the number of its earlier up steps.

After k up steps, the next step is up with probability p / (1 + k)^nu and down otherwise. The walker remembers its number of up steps at the end of the last step, from which its step follows.

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': 'x'})
x Annotated[float, StateSpec(domain=ZZ, role=position)]

The initial position.

 0.0
ups Annotated[float, StateSpec(domain=NN, lags=1)]

The initial number of up steps.

 0.0
p Annotated[float, ParameterSpec(domain=closed_open(0, oo))]

The probability of the first up step.

 0.5
nu Annotated[float, ParameterSpec(domain=closed_open(0, oo))]

The exponent of the decay of the up probability with the number of up steps.

 1.0

Examples:

>>> from skcomplex.core.elements import GroupElement
>>> from skcomplex.models.mathematics.stochastic_processes import AgeingWalkerElement
>>> group = GroupElement('walkers', element=AgeingWalkerElement, size=10, values={'nu': 0.5})
>>> group.label, group.element.__name__, group.size
('walkers', 'AgeingWalkerElement', 10)
Source code in src/skcomplex/models/mathematics/stochastic_processes/_ageing_walker.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
def __init__(
    self,
    label: str,
    visual: VisualSpec | None = VisualSpec('agents', {'color_by': 'x'}),
    *,
    x: Annotated[float, StateSpec(domain=ZZ, role='position')] = 0.0,
    ups: Annotated[float, StateSpec(domain=NN, lags=1)] = 0.0,
    p: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.5,
    nu: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
) -> None:
    super().__init__(label, visual=visual)
    self.x = x
    self.ups = ups
    self.p = p
    self.nu = nu

BernoulliTrialElement(label, visual=VisualSpec('histograms', {'color_by': 'successes'}), *, outcome=0.0, successes=0.0, p=0.5)

Bases: Element

Sequence of independent trials, each a success with the same probability, and the count of its successes.

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('histograms', {'color_by': 'successes'})
outcome Annotated[float, StateSpec(domain=FiniteEnumeratedSet([0, 1]))]

The initial outcome, 1 for a success and 0 for a failure.

 0.0
successes Annotated[float, StateSpec(domain=NN)]

The initial number of successes.

 0.0
p Annotated[float, ParameterSpec(domain=closed_open(0, oo))]

The probability of a success.

 0.5

Examples:

>>> from skcomplex.core.elements import GroupElement
>>> from skcomplex.models.mathematics.stochastic_processes import BernoulliTrialElement
>>> group = GroupElement('trials', element=BernoulliTrialElement, size=4, values={'p': 0.2})
>>> group.label, group.element.__name__, group.size
('trials', 'BernoulliTrialElement', 4)
Source code in src/skcomplex/models/mathematics/stochastic_processes/_bernoulli_trial.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
def __init__(
    self,
    label: str,
    visual: VisualSpec | None = VisualSpec('histograms', {'color_by': 'successes'}),
    *,
    outcome: Annotated[float, StateSpec(domain=FiniteEnumeratedSet([0, 1]))] = 0.0,
    successes: Annotated[float, StateSpec(domain=NN)] = 0.0,
    p: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.5,
) -> None:
    super().__init__(label, visual=visual)
    self.outcome = outcome
    self.successes = successes
    self.p = p

FragmentElement(label, visual=VisualSpec('agents', {'color_by': 'length'}), *, length=1000.0, rate=0.5)

Bases: Element

Fragment of integer size, which the breaking of the model cuts at a uniformly random point into two fragments.

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': 'length'})
length Annotated[float, StateSpec(domain=open(0, oo))]

The initial size.

 1000.0
rate Annotated[float, ParameterSpec(domain=closed_open(0, oo))]

The probability that a fragment larger than 1 breaks in a step.

 0.5

Examples:

>>> from skcomplex.core.elements import GroupElement
>>> from skcomplex.models.mathematics.stochastic_processes import FragmentElement
>>> group = GroupElement('fragments', element=FragmentElement, size=1, values={'length': 100.0})
>>> group.label, group.element.__name__, group.size
('fragments', 'FragmentElement', 1)
Source code in src/skcomplex/models/mathematics/stochastic_processes/_fragment.py 
41
42
43
44
45
46
47
48
49
50
51
def __init__(
    self,
    label: str,
    visual: VisualSpec | None = VisualSpec('agents', {'color_by': 'length'}),
    *,
    length: Annotated[float, StateSpec(domain=RealSet.open(0, oo))] = 1000.0,
    rate: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.5,
) -> None:
    super().__init__(label, visual=visual)
    self.length = length
    self.rate = rate

breaks(agent, model) staticmethod

Return the probability that a fragment breaks in a step, zero at size 1.

Parameters:

Name Type Description Default
agent FragmentView

The fragment.

 required
model ModelView

The model.

 required

Returns:

Name Type Description
probability float

The breaking probability, or zero for a fragment of size 1.
Source code in src/skcomplex/models/mathematics/stochastic_processes/_fragment.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
@staticmethod
def breaks(agent: FragmentView, model: ModelView) -> float:
    """Return the probability that a fragment breaks in a step, zero at size 1.

    Args:
        agent:
            The fragment.

        model:
            The model.

    Returns:
        probability:
            The breaking probability, or zero for a fragment of size 1.
    """
    return agent.rate if round(agent.length) > 1 else 0.0

cut(agent, model) staticmethod

Return the share of a breaking fragment that stays with it, cut at a uniformly random whole point.

Parameters:

Name Type Description Default
agent FragmentView

The fragment.

 required
model ModelView

The model.

 required

Returns:

Name Type Description
share float

The share of the size left of the cut.
Source code in src/skcomplex/models/mathematics/stochastic_processes/_fragment.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
@staticmethod
def cut(agent: FragmentView, model: ModelView) -> float:
    """Return the share of a breaking fragment that stays with it, cut at a uniformly random whole point.

    Args:
        agent:
            The fragment.

        model:
            The model.

    Returns:
        share:
            The share of the size left of the cut.
    """
    length = round(agent.length)
    return int(model.rng.integers(1, length)) / length

HiddenMarkovSourceElement(label, visual=None, *, hidden=0, symbol=0)

Bases: Element

Source of symbols emitted from the states of a hidden Markov chain, one symbol per step.

Parameters:

Name Type Description Default
label str

The label of the source.

 required
visual VisualSpec | None

How the element is drawn, or None for default_visual_.

 None
hidden Annotated[float, StateSpec(domain=NN)]

The initial hidden state.

 0
symbol Annotated[float, StateSpec(domain=NN)]

The initial symbol.

 0

Examples:

>>> from skcomplex.models.mathematics.stochastic_processes import HiddenMarkovSourceElement
>>> HiddenMarkovSourceElement('source', hidden=1).hidden
1
Source code in src/skcomplex/models/mathematics/stochastic_processes/_hidden_markov_source.py 
36
37
38
39
40
41
42
43
44
45
46
def __init__(
    self,
    label: str,
    visual: VisualSpec | None = None,
    *,
    hidden: Annotated[float, StateSpec(domain=NN)] = 0,
    symbol: Annotated[float, StateSpec(domain=NN)] = 0,
) -> None:
    super().__init__(label, visual=visual)
    self.hidden = hidden
    self.symbol = symbol

IndividualElement(label, visual=VisualSpec('agents', {'color_by': 'generation'}), *, generation=0.0, m=1.0)

Bases: Element

Individual of a Galton-Watson process, which the life cycle of the model replaces by its offspring.

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': 'generation'})
generation Annotated[float, StateSpec(domain=NN)]

The initial generation.

 0.0
m Annotated[float, ParameterSpec(domain=closed_open(0, oo))]

The mean number of offspring.

 1.0

Examples:

>>> from skcomplex.models.mathematics.stochastic_processes import IndividualElement
>>> group = GroupElement('population', element=IndividualElement, size=10, values={'m': 1.5})
>>> group.label, group.element.__name__, group.size
('population', 'IndividualElement', 10)
Source code in src/skcomplex/models/mathematics/stochastic_processes/_branching.py 
43
44
45
46
47
48
49
50
51
52
53
def __init__(
    self,
    label: str,
    visual: VisualSpec | None = VisualSpec('agents', {'color_by': 'generation'}),
    *,
    generation: Annotated[float, StateSpec(domain=NN)] = 0.0,
    m: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
) -> None:
    super().__init__(label, visual=visual)
    self.generation = generation
    self.m = m

IntermittentVariableElement(label, visual=None, *, x=1.0, epsilon=0.0, D=0.5)

Bases: Element

Positive variable driven across the bifurcation at epsilon = 0 by multiplicative noise of intensity D.

Parameters:

Name Type Description Default
label str

The label of the variable or the group.

 required
visual VisualSpec | None

How the element is drawn, or None for default_visual_.

 None
x Annotated[float, StateSpec(domain=open(0, oo))]

The initial value.

 1.0
epsilon Annotated[float, ParameterSpec()]

The control parameter, the distance from the bifurcation.

 0.0
D Annotated[float, ParameterSpec(domain=open(0, oo))]

The noise intensity.

 0.5

Examples:

>>> from skcomplex.models.mathematics.stochastic_processes import IntermittentVariableElement
>>> IntermittentVariableElement('variable', D=0.2).D
0.2
Source code in src/skcomplex/models/mathematics/stochastic_processes/_intermittency.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
def __init__(
    self,
    label: str,
    visual: VisualSpec | None = None,
    *,
    x: Annotated[float, StateSpec(domain=RealSet.open(0, oo))] = 1.0,
    epsilon: Annotated[float, ParameterSpec()] = 0.0,
    D: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 0.5,
) -> None:
    super().__init__(label, visual=visual)
    self.x = x
    self.epsilon = epsilon
    self.D = D

MarkovSourceElement(label, visual=None, *, symbol=0)

Bases: Element

Source of symbols 0 to n - 1, whose symbol a transition matrix steps in discrete time.

Parameters:

Name Type Description Default
label str

The label of the source.

 required
visual VisualSpec | None

How the element is drawn, or None for default_visual_.

 None
symbol Annotated[float, StateSpec(domain=NN)]

The initial symbol.

 0

Examples:

>>> from skcomplex.models.mathematics.stochastic_processes import MarkovSourceElement
>>> MarkovSourceElement('source', symbol=1).symbol
1
Source code in src/skcomplex/models/mathematics/stochastic_processes/_markov_source.py 
33
34
35
36
37
38
39
40
41
def __init__(
    self,
    label: str,
    visual: VisualSpec | None = None,
    *,
    symbol: Annotated[float, StateSpec(domain=NN)] = 0,
) -> None:
    super().__init__(label, visual=visual)
    self.symbol = symbol

MemoryWalkerElement(label, visual=None, *, x=0.0, v=0.0, alpha=0.5, sigma=1.0)

Bases: Element

Walker on the real line whose velocity is an autoregressive process of memory alpha and noise sigma.

Parameters:

Name Type Description Default
label str

The label of the walker or the group.

 required
visual VisualSpec | None

How the element is drawn, or None for default_visual_.

 None
x Annotated[float, StateSpec(role=position)]

The initial position.

 0.0
v Annotated[float, StateSpec(role=velocity)]

The initial velocity.

 0.0
alpha Annotated[float, ParameterSpec(domain=closed_open(0, oo))]

The memory of the velocity.

 0.5
sigma Annotated[float, ParameterSpec(domain=closed_open(0, oo))]

The deviation of the noise of the velocity.

 1.0

Examples:

>>> from skcomplex.models.mathematics.stochastic_processes import MemoryWalkerElement
>>> MemoryWalkerElement('walker', alpha=0.3).alpha
0.3
Source code in src/skcomplex/models/mathematics/stochastic_processes/_memory_walker.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def __init__(
    self,
    label: str,
    visual: VisualSpec | None = None,
    *,
    x: Annotated[float, StateSpec(role='position')] = 0.0,
    v: Annotated[float, StateSpec(role='velocity')] = 0.0,
    alpha: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.5,
    sigma: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
) -> None:
    super().__init__(label, visual=visual)
    self.x = x
    self.v = v
    self.alpha = alpha
    self.sigma = sigma

PersistentWalkerElement(label, visual=None, *, d=1.0, x=0.0, persistence=0.8)

Bases: Element

Walker on the integers with a direction d = ±1, which it keeps with the probability persistence.

Parameters:

Name Type Description Default
label str

The label of the walker or the group.

 required
visual VisualSpec | None

How the element is drawn, or None for default_visual_.

 None
d Annotated[float, StateSpec(domain=ZZ)]

The initial direction, 1 or -1.

 1.0
x Annotated[float, StateSpec(domain=ZZ, role=position)]

The initial position.

 0.0
persistence Annotated[float, ParameterSpec(domain=closed_open(0, oo))]

The probability of keeping the direction at a step.

 0.8

Examples:

>>> from skcomplex.models.mathematics.stochastic_processes import PersistentWalkerElement
>>> PersistentWalkerElement('walker', persistence=0.9).persistence
0.9
Source code in src/skcomplex/models/mathematics/stochastic_processes/_persistent_walker.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
def __init__(
    self,
    label: str,
    visual: VisualSpec | None = None,
    *,
    d: Annotated[float, StateSpec(domain=ZZ)] = 1.0,
    x: Annotated[float, StateSpec(domain=ZZ, role='position')] = 0.0,
    persistence: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.8,
) -> None:
    super().__init__(label, visual=visual)
    self.d = d
    self.x = x
    self.persistence = persistence

PoissonCounterElement(label, visual=None, *, count=0.0, rate=1.0)

Bases: Element

Counter of the events of a Poisson process, with the rate of events it is counted at.

Parameters:

Name Type Description Default
label str

The label of the counter or the group.

 required
visual VisualSpec | None

How the element is drawn, or None for default_visual_.

 None
count Annotated[float, StateSpec(domain=NN)]

The initial number of events.

 0.0
rate Annotated[float, ParameterSpec(domain=closed_open(0, oo))]

The rate of events, which the jump of the counter reads.

 1.0

Examples:

>>> from skcomplex.models.mathematics.stochastic_processes import PoissonCounterElement
>>> PoissonCounterElement('counter', rate=2.0).rate
2.0
Source code in src/skcomplex/models/mathematics/stochastic_processes/_poisson_counter.py 
37
38
39
40
41
42
43
44
45
46
47
def __init__(
    self,
    label: str,
    visual: VisualSpec | None = None,
    *,
    count: Annotated[float, StateSpec(domain=NN)] = 0.0,
    rate: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
) -> None:
    super().__init__(label, visual=visual)
    self.count = count
    self.rate = rate

PolyaUrnElement(label, visual=None, *, red=1.0, blue=1.0, added=1)

Bases: Element

Urn of red and blue balls, from which a drawn ball returns with added extra balls of its color.

The jumps of the model draw at rate 1, and a draw picks a color with probability proportional to its number of balls, so the sequence of draws is that of Pólya's urn in discrete time.

Parameters:

Name Type Description Default
label str

The label of the urn or the group.

 required
visual VisualSpec | None

How the element is drawn, or None for default_visual_.

 None
red Annotated[float, StateSpec(domain=NN)]

The initial number of red balls.

 1.0
blue Annotated[float, StateSpec(domain=NN)]

The initial number of blue balls.

 1.0
added int

The number of balls of the drawn color added after every draw.

 1

Examples:

>>> from skcomplex.models.mathematics.stochastic_processes import PolyaUrnElement
>>> PolyaUrnElement('urn', added=2).added
2
Source code in src/skcomplex/models/mathematics/stochastic_processes/_urns.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
def __init__(
    self,
    label: str,
    visual: VisualSpec | None = None,
    *,
    red: Annotated[float, StateSpec(domain=NN)] = 1.0,
    blue: Annotated[float, StateSpec(domain=NN)] = 1.0,
    added: int = 1,
) -> None:
    super().__init__(label, visual=visual)
    self.red = red
    self.blue = blue
    self.added = added

QueueElement(label, visual=None, *, n=0.0, arrival_rate=0.5, service_rate=1.0)

Bases: Element

Queue at a desk, the number of customers waiting, with the rates of their arrivals and departures.

The jumps of the model read the rates: customers arrive at rate arrival_rate and leave at rate service_rate while the queue is not empty.

Parameters:

Name Type Description Default
label str

The label of the queue or the group.

 required
visual VisualSpec | None

How the element is drawn, or None for default_visual_.

 None
n Annotated[float, StateSpec(domain=NN)]

The initial number of customers.

 0.0
arrival_rate Annotated[float, ParameterSpec(domain=closed_open(0, oo))]

The rate of arrivals.

 0.5
service_rate Annotated[float, ParameterSpec(domain=closed_open(0, oo))]

The rate of departures from a queue that is not empty.

 1.0

Examples:

>>> from skcomplex.models.mathematics.stochastic_processes import QueueElement
>>> QueueElement('desk', arrival_rate=0.8).arrival_rate
0.8
Source code in src/skcomplex/models/mathematics/stochastic_processes/_queues.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
def __init__(
    self,
    label: str,
    visual: VisualSpec | None = None,
    *,
    n: Annotated[float, StateSpec(domain=NN)] = 0.0,
    arrival_rate: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.5,
    service_rate: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
) -> None:
    super().__init__(label, visual=visual)
    self.n = n
    self.arrival_rate = arrival_rate
    self.service_rate = service_rate

RandomWalkerElement(label, visual=None, *, x=0.0, rate=1.0, bias=0.5)

Bases: Element

Walker on the integers that steps at the rate rate, up with the probability bias.

Parameters:

Name Type Description Default
label str

The label of the walker or the group.

 required
visual VisualSpec | None

How the element is drawn, or None for default_visual_.

 None
x Annotated[float, StateSpec(domain=ZZ, role=position)]

The initial position.

 0.0
rate Annotated[float, ParameterSpec(domain=closed_open(0, oo))]

The rate of steps.

 1.0
bias Annotated[float, ParameterSpec(domain=closed_open(0, oo))]

The probability that a step goes up.

 0.5

Examples:

>>> from skcomplex.models.mathematics.stochastic_processes import RandomWalkerElement
>>> RandomWalkerElement('walker', bias=0.7).bias
0.7
Source code in src/skcomplex/models/mathematics/stochastic_processes/_random_walker.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
def __init__(
    self,
    label: str,
    visual: VisualSpec | None = None,
    *,
    x: Annotated[float, StateSpec(domain=ZZ, role='position')] = 0.0,
    rate: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
    bias: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.5,
) -> None:
    super().__init__(label, visual=visual)
    self.x = x
    self.rate = rate
    self.bias = bias

ReducingBallElement(label, visual=VisualSpec('agents', {'color_by': 'level'}), *, level=10.0, n_states=10.0, lam=1.0)

Bases: Element

Ball of the sample-space-reducing process, whose state the reducing jump moves to a random lower state.

With probability lam the jump takes the ball uniformly to a state below its own, and otherwise uniformly to any state, and the ball restarts from the top state after it reaches state 1.

Parameters:

Name Type Description Default
label str

The label of the agent or group.

 required
visual VisualSpec | None

How the element is drawn, or None for default_visual_.

 VisualSpec('agents', {'color_by': 'level'})
level Annotated[float, StateSpec(domain=NN)]

The initial state, from 1 to n_states.

 10.0
n_states Annotated[float, ParameterSpec(domain=open(0, oo))]

The number of states.

 10.0
lam Annotated[float, ParameterSpec(domain=closed_open(0, oo))]

The probability of a reducing jump.

 1.0

Examples:

>>> from skcomplex.core.elements import GroupElement
>>> from skcomplex.models.mathematics.stochastic_processes import ReducingBallElement
>>> group = GroupElement('balls', element=ReducingBallElement, size=3, values={'n_states': 10.0})
>>> group.label, group.element.__name__, group.size
('balls', 'ReducingBallElement', 3)
Source code in src/skcomplex/models/mathematics/stochastic_processes/_reducing_ball.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
def __init__(
    self,
    label: str,
    visual: VisualSpec | None = VisualSpec('agents', {'color_by': 'level'}),
    *,
    level: Annotated[float, StateSpec(domain=NN)] = 10.0,
    n_states: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 10.0,
    lam: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
) -> None:
    super().__init__(label, visual=visual)
    self.level = level
    self.n_states = n_states
    self.lam = lam

ReducingJumpInteraction(label, *, target)

Bases: MapInteraction

Jump of the sample-space-reducing process, to a random lower state, to any state, or back to the top state.

A ball at state 1 restarts from the top state n_states. Otherwise it jumps with probability lam uniformly to a state below its own, and else uniformly to any state, so the states are visited with probabilities proportional to i^(-lam).

Parameters:

Name Type Description Default
label str

The label of the interaction.

 required
target Annotated[str, RoleSpec()]

The path of the group of balls.

 required

Examples:

>>> from skcomplex.models.mathematics.stochastic_processes import ReducingJumpInteraction
>>> ReducingJumpInteraction('jump', target='balls').check().state
'level'
Source code in src/skcomplex/models/mathematics/stochastic_processes/_reducing_jump.py 
34
35
def __init__(self, label: str, *, target: Annotated[str, RoleSpec()]) -> None:
    super().__init__(label, target=target, state='level', function=self._get_level, form='agent')

WeakeningParticleElement(label, visual=None, *, x=0.0, k=1.0, ell=1.0, D=0.25)

Bases: Element

Brownian particle on the real line held by a restoring force of stiffness k that weakens beyond ell.

Parameters:

Name Type Description Default
label str

The label of the particle or the group.

 required
visual VisualSpec | None

How the element is drawn, or None for default_visual_.

 None
x Annotated[float, StateSpec(role=position)]

The initial position.

 0.0
k Annotated[float, ParameterSpec(domain=open(0, oo))]

The stiffness of the force near the origin.

 1.0
ell Annotated[float, ParameterSpec(domain=open(0, oo))]

The distance where the force starts to weaken.

 1.0
D Annotated[float, ParameterSpec(domain=open(0, oo))]

The diffusion coefficient.

 0.25

Examples:

>>> from skcomplex.models.mathematics.stochastic_processes import WeakeningParticleElement
>>> WeakeningParticleElement('particle', D=0.5).D
0.5
Source code in src/skcomplex/models/mathematics/stochastic_processes/_weakening_particle.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def __init__(
    self,
    label: str,
    visual: VisualSpec | None = None,
    *,
    x: Annotated[float, StateSpec(role='position')] = 0.0,
    k: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 1.0,
    ell: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 1.0,
    D: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 0.25,
) -> None:
    super().__init__(label, visual=visual)
    self.x = x
    self.k = k
    self.ell = ell
    self.D = D

ageing_random_walk(n_walkers=50, *, p=0.5, nu=1.0)

Return ageing random walks, whose up steps become rarer the more of them a walker has taken.

After k up steps a walker steps up with probability p / (1 + k)^nu, so the walk is history dependent and, for nu > 0, its number of up steps grows as n^(1 / (1 + nu)) after n steps.

Parameters:

Name Type Description Default
n_walkers int

The number of independent walkers, all started at 0.

 50
p float

The probability of the first up step.

 0.5
nu float

The exponent of the decay of the up probability.

 1.0

Returns:

Name Type Description
system System

The model, with the group walkers.

Examples:

>>> from skcomplex.models.mathematics.stochastic_processes import ageing_random_walk
>>> ageing_random_walk(5).elements[0].check().prepare().size_
5
Source code in src/skcomplex/models/mathematics/stochastic_processes/_walks.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
@register_model(
    field='mathematics/stochastic_processes',
    tags=('random walk', 'agents', 'history dependence', 'ageing'),
    references=('Hanel, Thurner, and Gell-Mann 2014', 'Thurner et al. 2018'),
    defaults={'until': 100},
)
def ageing_random_walk(n_walkers: int = 50, *, p: float = 0.5, nu: float = 1.0) -> System:
    """Return ageing random walks, whose up steps become rarer the more of them a walker has taken.

    After `k` up steps a walker steps up with probability `p / (1 + k)^nu`, so the walk is history dependent and,
    for `nu > 0`, its number of up steps grows as `n^(1 / (1 + nu))` after `n` steps.

    Args:
        n_walkers:
            The number of independent walkers, all started at 0.

        p:
            The probability of the first up step.

        nu:
            The exponent of the decay of the up probability.

    Returns:
        system:
            The model, with the group `walkers`.

    Examples:
        >>> from skcomplex.models.mathematics.stochastic_processes import ageing_random_walk
        >>> ageing_random_walk(5).elements[0].check().prepare().size_
        5
    """
    ageing = MapInteraction(
        'ageing',
        target='walkers',
        state='ups',
        function=lambda agent, model: float(model.rng.random() < agent.p / (1 + agent.ups) ** agent.nu),
        mode='increment',
        form='agent',
    )
    step = MapInteraction(
        'step',
        target='walkers',
        state='x',
        function=lambda agent, model: 2 * (agent.ups - agent.lags('ups')[-1]) - 1,
        mode='increment',
        form='agent',
    )
    return System(
        elements=[GroupElement('walkers', element=AgeingWalkerElement, size=n_walkers, values={'p': p, 'nu': nu})],
        interactions=[ageing, step],
    )

bernoulli_process(n_sequences=20, *, p=0.5)

Return independent Bernoulli processes, one success or failure per step, whose success counts are binomial.

Parameters:

Name Type Description Default
n_sequences int

The number of independent sequences of trials.

 20
p float

The probability of a success.

 0.5

Returns:

Name Type Description
system System

The model, with the group trials of states outcome and successes.

Examples:

>>> from skcomplex.models.mathematics.stochastic_processes import bernoulli_process
>>> bernoulli_process(3).elements[0].check().prepare().size_
3
Source code in src/skcomplex/models/mathematics/stochastic_processes/_counting.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
@register_model(
    field='mathematics/stochastic_processes',
    tags=('counting process', 'agents', 'independent trials'),
    references=('Thurner et al. 2018',),
    defaults={'until': 100},
)
def bernoulli_process(n_sequences: int = 20, *, p: float = 0.5) -> System:
    """Return independent Bernoulli processes, one success or failure per step, whose success counts are binomial.

    Args:
        n_sequences:
            The number of independent sequences of trials.

        p:
            The probability of a success.

    Returns:
        system:
            The model, with the group `trials` of states `outcome` and `successes`.

    Examples:
        >>> from skcomplex.models.mathematics.stochastic_processes import bernoulli_process
        >>> bernoulli_process(3).elements[0].check().prepare().size_
        3
    """
    trial = MapInteraction(
        'trial',
        target='trials',
        state='outcome',
        function=lambda agent, model: float(model.rng.random() < agent.p),
        form='agent',
    )
    count = MapInteraction(
        'count',
        target='trials',
        state='successes',
        function=lambda agent, model: agent.outcome,
        mode='increment',
        form='agent',
    )
    return System(
        elements=[GroupElement('trials', element=BernoulliTrialElement, size=n_sequences, values={'p': p})],
        interactions=[trial, count],
    )

birth_death_queue(n_queues=100, *, arrival_rate=0.5, service_rate=1.0, n=0.0)

Return independent M/M/1 queues, birth-death processes with constant arrival and service rates.

Customers arrive as a Poisson process of rate lambda and are served one at a time in exponential times of rate mu. For the load rho = lambda / mu < 1 the length of an unbounded queue reaches the stationary geometric distribution (1 - rho) rho^n, of mean rho / (1 - rho) and variance rho / (1 - rho)^2, and for rho > 1 it grows without bound.

Parameters:

Name Type Description Default
n_queues int

The number of independent queues.

 100
arrival_rate float

The rate lambda of arrivals.

 0.5
service_rate float

The rate mu of departures.

 1.0
n float

The initial number of customers of every queue.

 0.0

Returns:

Name Type Description
system System

The model, with the lengths queues.n.

Examples:

>>> from skcomplex.models.mathematics.stochastic_processes import birth_death_queue
>>> birth_death_queue(5).elements[0].check().prepare().size_
5
Source code in src/skcomplex/models/mathematics/stochastic_processes/_queues.py 
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
@register_model(
    field='mathematics/stochastic_processes',
    tags=('jump process', 'birth-death process', 'queueing'),
    references=('Bertin 2021', 'Kendall 1953', 'Feller 1968'),
    defaults={'until': 200.0, 'dt': 0.5},
)
def birth_death_queue(
    n_queues: int = 100,
    *,
    arrival_rate: float = 0.5,
    service_rate: float = 1.0,
    n: float = 0.0,
) -> System:
    """Return independent M/M/1 queues, birth-death processes with constant arrival and service rates.

    Customers arrive as a Poisson process of rate `lambda` and are served one at a time in exponential times of
    rate `mu`. For the load `rho = lambda / mu < 1` the length of an unbounded queue reaches the stationary
    geometric distribution `(1 - rho) rho^n`, of mean `rho / (1 - rho)` and variance `rho / (1 - rho)^2`, and for
    `rho > 1` it grows without bound.

    Args:
        n_queues:
            The number of independent queues.

        arrival_rate:
            The rate `lambda` of arrivals.

        service_rate:
            The rate `mu` of departures.

        n:
            The initial number of customers of every queue.

    Returns:
        system:
            The model, with the lengths `queues.n`.

    Examples:
        >>> from skcomplex.models.mathematics.stochastic_processes import birth_death_queue
        >>> birth_death_queue(5).elements[0].check().prepare().size_
        5
    """
    queue = GroupElement(
        'queues',
        element=QueueElement,
        size=n_queues,
        values={'n': n, 'arrival_rate': arrival_rate, 'service_rate': service_rate},
    )
    arrivals = JumpInteraction('arrivals', target='queues', changes={'n': 1}, rate=lambda s: s.target.arrival_rate)
    departures = JumpInteraction(
        'departures',
        target='queues',
        changes={'n': -1},
        rate=lambda s: where(s.target.n > 0, s.target.service_rate, 0),
    )
    return System(elements=[queue], interactions=[arrivals, departures])

fragmentation(size=1000, *, rate=0.5)

Return the random fragmentation of an object of integer size, whose fragments keep breaking into smaller ones.

The total size is conserved, and every fragment ends at size 1.

Parameters:

Name Type Description Default
size int

The size of the initial object.

 1000
rate float

The probability that a fragment larger than 1 breaks in a step.

 0.5

Returns:

Name Type Description
system System

The model, with the group fragments of state length.

Examples:

>>> from skcomplex.models.mathematics.stochastic_processes import fragmentation
>>> fragmentation(50).elements[0].check().prepare().size_
1
Source code in src/skcomplex/models/mathematics/stochastic_processes/_reducing.py 
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
@register_model(
    field='mathematics/stochastic_processes',
    tags=('fragmentation', 'agents', 'sample space reduction'),
    references=('Thurner et al. 2018',),
    defaults={'until': 30},
)
def fragmentation(size: int = 1000, *, rate: float = 0.5) -> System:
    """Return the random fragmentation of an object of integer size, whose fragments keep breaking into smaller ones.

    The total size is conserved, and every fragment ends at size 1.

    Args:
        size:
            The size of the initial object.

        rate:
            The probability that a fragment larger than 1 breaks in a step.

    Returns:
        system:
            The model, with the group `fragments` of state `length`.

    Examples:
        >>> from skcomplex.models.mathematics.stochastic_processes import fragmentation
        >>> fragmentation(50).elements[0].check().prepare().size_
        1
    """
    return System(
        elements=[
            GroupElement(
                'fragments',
                element=FragmentElement,
                size=1,
                values={'length': float(size), 'rate': rate},
                variable=True,
            ),
        ],
        interactions=[
            LifeCycleInteraction(
                'breaking',
                target='fragments',
                birth=FragmentElement.breaks,
                split=('length',),
                keep=FragmentElement.cut,
            ),
        ],
    )

galton_watson(n_individuals=10, *, m=1.0)

Return a Galton-Watson branching process with Poisson offspring, one generation per step.

The mean size of generation n is n_individuals m^n. The process dies out with certainty for m <= 1, and survives with a positive probability for m > 1.

Parameters:

Name Type Description Default
n_individuals int

The size of the first generation.

 10
m float

The mean number of offspring of an individual.

 1.0

Returns:

Name Type Description
system System

The model, with the group population, whose count is the size of the current generation.

Examples:

>>> from skcomplex.models.mathematics.stochastic_processes import galton_watson
>>> galton_watson(5, m=1.2).elements[0].check().prepare().size_
5
Source code in src/skcomplex/models/mathematics/stochastic_processes/_branching.py 
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
@register_model(
    field='mathematics/stochastic_processes',
    tags=('branching process', 'agents', 'extinction', 'criticality'),
    references=('Watson and Galton 1875', 'Gros 2015', 'Jensen 2023'),
    defaults={'until': 20},
)
def galton_watson(n_individuals: int = 10, *, m: float = 1.0) -> System:
    """Return a Galton-Watson branching process with Poisson offspring, one generation per step.

    The mean size of generation `n` is `n_individuals m^n`. The process dies out with certainty for `m <= 1`, and
    survives with a positive probability for `m > 1`.

    Args:
        n_individuals:
            The size of the first generation.

        m:
            The mean number of offspring of an individual.

    Returns:
        system:
            The model, with the group `population`, whose count is the size of the current generation.

    Examples:
        >>> from skcomplex.models.mathematics.stochastic_processes import galton_watson
        >>> galton_watson(5, m=1.2).elements[0].check().prepare().size_
        5
    """
    return System(
        elements=[
            GroupElement('population', element=IndividualElement, size=n_individuals, values={'m': m}, variable=True),
        ],
        interactions=[
            LifeCycleInteraction(
                'reproduction',
                target='population',
                birth=1.0,
                offspring=lambda agent, model: int(model.rng.poisson(agent.m)),
                values=lambda agent, model: {'generation': agent.generation + 1},
                replace=True,
            ),
        ],
    )

hidden_markov_source(transition=CHAIN, emission=EMISSION)

Return a hidden Markov source, whose symbols are emitted at random from the states of a hidden Markov chain.

The hidden state follows the transition matrix P, and in hidden state i the source emits symbol j with probability E_ij. The frequency of symbol j is Σ_i π_i E_ij for the stationary distribution π of the hidden chain, and the symbols alone are not a Markov chain of any finite order in general.

Parameters:

Name Type Description Default
transition Matrix

The transition matrix P of the hidden chain.

 CHAIN
emission Matrix

The emission matrix E, one row per hidden state and one column per symbol.

 EMISSION

Returns:

Name Type Description
system System

The model, with the hidden states source.hidden and the symbols source.symbol.

Examples:

>>> from skcomplex.models.mathematics.stochastic_processes import hidden_markov_source
>>> hidden_markov_source().interactions[0].observed
'symbol'
Source code in src/skcomplex/models/mathematics/stochastic_processes/_sources.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
@register_model(
    field='mathematics/stochastic_processes',
    tags=('hidden markov model', 'information theory', 'markov chain', 'discrete map'),
    references=('Lindgren 2024',),
    defaults={'until': 300, 'dt': 1},
)
def hidden_markov_source(transition: Matrix = CHAIN, emission: Matrix = EMISSION) -> System:
    """Return a hidden Markov source, whose symbols are emitted at random from the states of a hidden Markov chain.

    The hidden state follows the transition matrix `P`, and in hidden state `i` the source emits symbol `j` with
    probability `E_ij`. The frequency of symbol `j` is `Σ_i π_i E_ij` for the stationary distribution `π` of the hidden
    chain, and the symbols alone are not a Markov chain of any finite order in general.

    Args:
        transition:
            The transition matrix `P` of the hidden chain.

        emission:
            The emission matrix `E`, one row per hidden state and one column per symbol.

    Returns:
        system:
            The model, with the hidden states `source.hidden` and the symbols `source.symbol`.

    Examples:
        >>> from skcomplex.models.mathematics.stochastic_processes import hidden_markov_source
        >>> hidden_markov_source().interactions[0].observed
        'symbol'
    """
    chain = TransitionInteraction(
        'chain',
        target='source',
        state='hidden',
        matrix=lambda s: transition,
        emission=lambda s: emission,
        observed='symbol',
    )
    return System(elements=[HiddenMarkovSourceElement('source')], interactions=[chain])

markov_source(transition=CHAIN, *, symbol=0)

Return a Markov source, a sequence of symbols in which each symbol depends on the one before it.

The symbol after i is j with probability P_ij. The entropy rate of the source is h = -Σ_i π_i Σ_j P_ij log2 P_ij bits per symbol, where π is the stationary distribution of the chain, and the block entropies of the sequence approach it from above.

Parameters:

Name Type Description Default
transition Matrix

The transition matrix P, one row and one column per symbol, with rows that sum to 1.

 CHAIN
symbol int

The initial symbol.

 0

Returns:

Name Type Description
system System

The model, with the symbols source.symbol.

Examples:

>>> from skcomplex.models.mathematics.stochastic_processes import markov_source
>>> markov_source().interactions[0].state
'symbol'
Source code in src/skcomplex/models/mathematics/stochastic_processes/_sources.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
@register_model(
    field='mathematics/stochastic_processes',
    tags=('markov chain', 'information theory', 'entropy rate', 'discrete map'),
    references=('Lindgren 2024',),
    defaults={'until': 300, 'dt': 1},
)
def markov_source(transition: Matrix = CHAIN, *, symbol: int = 0) -> System:
    """Return a Markov source, a sequence of symbols in which each symbol depends on the one before it.

    The symbol after `i` is `j` with probability `P_ij`. The entropy rate of the source is
    `h = -Σ_i π_i Σ_j P_ij log2 P_ij` bits per symbol, where `π` is the stationary distribution of the chain, and the
    block entropies of the sequence approach it from above.

    Args:
        transition:
            The transition matrix `P`, one row and one column per symbol, with rows that sum to 1.

        symbol:
            The initial symbol.

    Returns:
        system:
            The model, with the symbols `source.symbol`.

    Examples:
        >>> from skcomplex.models.mathematics.stochastic_processes import markov_source
        >>> markov_source().interactions[0].state
        'symbol'
    """
    chain = TransitionInteraction('chain', target='source', state='symbol', matrix=lambda s: transition)
    return System(elements=[MarkovSourceElement('source', symbol=symbol)], interactions=[chain])

on_off_intermittency(n_realizations=20, *, epsilon=0.0, D=0.5, x=1.0)

Return stochastic on-off intermittency, dx/dt = (epsilon + xi) x - x^3 with Stratonovich white noise xi.

The noise has the correlation <xi(t) xi(t')> = 2 D delta(t - t'). The variable y = ln x diffuses freely while x is small and is pushed back by -exp(2 y) once x grows, so near the bifurcation epsilon = 0 the variable rests close to 0 for long laminar phases, whose durations are distributed as tau^(-3/2), between bursts where x is of order 1. For epsilon > 0 the power law is cut off beyond durations of order D / epsilon^2.

Parameters:

Name Type Description Default
n_realizations int

The number of independent realizations.

 20
epsilon float

The control parameter, the distance from the bifurcation.

 0.0
D float

The noise intensity.

 0.5
x float

The initial value of every realization.

 1.0

Returns:

Name Type Description
system System

The model, with the values variables.x.

Examples:

>>> from skcomplex.models.mathematics.stochastic_processes import on_off_intermittency
>>> on_off_intermittency(5).elements[0].size
5
Source code in src/skcomplex/models/mathematics/stochastic_processes/_intermittency.py 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
@register_model(
    field='mathematics/stochastic_processes',
    tags=('noise', 'sde', 'multiplicative noise', 'intermittency', 'bifurcation'),
    references=('Bertin 2021', 'Platt, Spiegel, and Tresser 1993', 'Heagy, Platt, and Hammel 1994'),
    defaults={'until': 500.0, 'dt': 0.1, 'solver': HeunSolver(dt=0.01)},
)
def on_off_intermittency(
    n_realizations: int = 20,
    *,
    epsilon: float = 0.0,
    D: float = 0.5,
    x: float = 1.0,
) -> System:
    """Return stochastic on-off intermittency, `dx/dt = (epsilon + xi) x - x^3` with Stratonovich white noise `xi`.

    The noise has the correlation `<xi(t) xi(t')> = 2 D delta(t - t')`. The variable `y = ln x` diffuses freely while
    `x` is small and is pushed back by `-exp(2 y)` once `x` grows, so near the bifurcation `epsilon = 0` the variable
    rests close to 0 for long laminar phases, whose durations are distributed as `tau^(-3/2)`, between bursts where
    `x` is of order 1. For `epsilon > 0` the power law is cut off beyond durations of order `D / epsilon^2`.

    Args:
        n_realizations:
            The number of independent realizations.

        epsilon:
            The control parameter, the distance from the bifurcation.

        D:
            The noise intensity.

        x:
            The initial value of every realization.

    Returns:
        system:
            The model, with the values `variables.x`.

    Examples:
        >>> from skcomplex.models.mathematics.stochastic_processes import on_off_intermittency
        >>> on_off_intermittency(5).elements[0].size
        5
    """
    variables = GroupElement(
        'variables',
        element=IntermittentVariableElement,
        size=n_realizations,
        values={'x': x, 'epsilon': epsilon, 'D': D},
    )
    drift = RateInteraction(
        'drift',
        target='variables',
        state='x',
        rate=lambda s: s.target.epsilon * s.target.x - s.target.x**3,
        noise=lambda s: sqrt(2 * s.target.D) * s.target.x,
        convention='stratonovich',
    )
    return System(elements=[variables], interactions=[drift])

persistent_random_walk(n_walkers=2000, *, persistence=0.8, random_state=0)

Return walkers on the integers that keep the direction of their last step with a probability.

At every step a walker moves by its direction d = ±1, and keeps the direction with probability q and reverses it otherwise. Successive steps are correlated with rho = 2 q - 1, so the walk diffuses faster than a simple random walk for q > 1/2, with a mean squared displacement that grows as t (1 + rho) / (1 - rho) at long times.

Parameters:

Name Type Description Default
n_walkers int

The number of independent walkers, all started at 0.

 2000
persistence float

The probability q of keeping the direction.

 0.8
random_state int | Generator | None

The seed or generator of the initial directions.

 0

Returns:

Name Type Description
system System

The model, with the directions walkers.d and the positions walkers.x.

Examples:

>>> from skcomplex.models.mathematics.stochastic_processes import persistent_random_walk
>>> persistent_random_walk(10).elements[0].check().prepare().size_
10
Source code in src/skcomplex/models/mathematics/stochastic_processes/_walks.py 
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
@register_model(
    field='mathematics/stochastic_processes',
    tags=('random walk', 'discrete map', 'persistence', 'diffusion'),
    references=('Jensen 2023',),
    defaults={'until': 200, 'dt': 1},
)
def persistent_random_walk(
    n_walkers: int = 2000,
    *,
    persistence: float = 0.8,
    random_state: int | np.random.Generator | None = 0,
) -> System:
    """Return walkers on the integers that keep the direction of their last step with a probability.

    At every step a walker moves by its direction `d = ±1`, and keeps the direction with probability `q` and reverses
    it otherwise. Successive steps are correlated with `rho = 2 q - 1`, so the walk diffuses faster than a simple
    random walk for `q > 1/2`, with a mean squared displacement that grows as `t (1 + rho) / (1 - rho)` at long times.

    Args:
        n_walkers:
            The number of independent walkers, all started at 0.

        persistence:
            The probability `q` of keeping the direction.

        random_state:
            The seed or generator of the initial directions.

    Returns:
        system:
            The model, with the directions `walkers.d` and the positions `walkers.x`.

    Examples:
        >>> from skcomplex.models.mathematics.stochastic_processes import persistent_random_walk
        >>> persistent_random_walk(10).elements[0].check().prepare().size_
        10
    """
    directions = np.random.default_rng(random_state).choice([-1.0, 1.0], size=n_walkers)
    walkers = GroupElement(
        'walkers',
        element=PersistentWalkerElement,
        size=n_walkers,
        values={'d': directions, 'persistence': persistence},
    )
    turn = MapInteraction(
        'turn',
        target='walkers',
        state='d',
        function=lambda s: s.target.d * (1 - 2 * bernoulli('turn', 1 - s.target.persistence)),
    )
    step = MapInteraction(
        'step',
        target='walkers',
        state='x',
        function=lambda s: s.target.d,
        mode='increment',
    )
    return System(elements=[walkers], interactions=[turn, step])

poisson_process(n_processes=20, *, rate=1.0)

Return independent Poisson processes, counters that grow by one at exponentially distributed waiting times.

Parameters:

Name Type Description Default
n_processes int

The number of independent counters.

 20
rate float

The rate of events of a counter.

 1.0

Returns:

Name Type Description
system System

The model, with the counts counters.count.

Examples:

>>> from skcomplex.models.mathematics.stochastic_processes import poisson_process
>>> poisson_process(5).elements[0].check().prepare().size_
5
Source code in src/skcomplex/models/mathematics/stochastic_processes/_counting.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
@register_model(
    field='mathematics/stochastic_processes',
    tags=('counting process', 'jump process', 'waiting times'),
    references=('Thurner et al. 2018',),
    defaults={'until': 50.0, 'dt': 1.0},
)
def poisson_process(n_processes: int = 20, *, rate: float = 1.0) -> System:
    """Return independent Poisson processes, counters that grow by one at exponentially distributed waiting times.

    Args:
        n_processes:
            The number of independent counters.

        rate:
            The rate of events of a counter.

    Returns:
        system:
            The model, with the counts `counters.count`.

    Examples:
        >>> from skcomplex.models.mathematics.stochastic_processes import poisson_process
        >>> poisson_process(5).elements[0].check().prepare().size_
        5
    """
    counters = GroupElement('counters', element=PoissonCounterElement, size=n_processes, values={'rate': rate})
    events = JumpInteraction('events', target='counters', changes={'count': 1}, rate=lambda s: s.target.rate)
    return System(elements=[counters], interactions=[events])

polya_urn(n_urns=10, *, red=1, blue=1, added=1)

Return independent Pólya urns, whose share of red balls converges to a random limit with a beta distribution.

The limit follows Beta(red / added, blue / added), uniform for one ball of each color and one added ball.

Parameters:

Name Type Description Default
n_urns int

The number of independent urns.

 10
red int

The initial number of red balls.

 1
blue int

The initial number of blue balls.

 1
added int

The number of balls of the drawn color added after every draw.

 1

Returns:

Name Type Description
system System

The model, with the states urns.red and urns.blue.

Examples:

>>> from skcomplex.models.mathematics.stochastic_processes import polya_urn
>>> polya_urn(3).elements[0].check().prepare().size_
3
Source code in src/skcomplex/models/mathematics/stochastic_processes/_urns.py 
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
@register_model(
    field='mathematics/stochastic_processes',
    tags=('urn', 'reinforcement', 'jump process', 'history dependence'),
    references=('Eggenberger and Pólya 1923', 'Thurner et al. 2018'),
    defaults={'until': 100.0, 'dt': 1.0},
)
def polya_urn(n_urns: int = 10, *, red: int = 1, blue: int = 1, added: int = 1) -> System:
    """Return independent Pólya urns, whose share of red balls converges to a random limit with a beta distribution.

    The limit follows `Beta(red / added, blue / added)`, uniform for one ball of each color and one added ball.

    Args:
        n_urns:
            The number of independent urns.

        red:
            The initial number of red balls.

        blue:
            The initial number of blue balls.

        added:
            The number of balls of the drawn color added after every draw.

    Returns:
        system:
            The model, with the states `urns.red` and `urns.blue`.

    Examples:
        >>> from skcomplex.models.mathematics.stochastic_processes import polya_urn
        >>> polya_urn(3).elements[0].check().prepare().size_
        3
    """
    urns = GroupElement(
        'urns',
        element=PolyaUrnElement,
        size=n_urns,
        values={'red': float(red), 'blue': float(blue), 'added': added},
    )
    red_draws = JumpInteraction(
        'red_draws',
        target='urns',
        changes={'red': added},
        rate=lambda s: s.target.red / (s.target.red + s.target.blue),
    )
    blue_draws = JumpInteraction(
        'blue_draws',
        target='urns',
        changes={'blue': added},
        rate=lambda s: s.target.blue / (s.target.red + s.target.blue),
    )
    return System(elements=[urns], interactions=[red_draws, blue_draws])

random_walk(n_walkers=100, *, rate=1.0, bias=0.5)

Return independent walkers on the integers that step at random times, a continuous-time random walk.

Every walker steps up at rate rate * bias and down at rate rate * (1 - bias), so its mean moves with velocity rate (2 bias - 1) and its variance grows as 4 rate bias (1 - bias) t.

Parameters:

Name Type Description Default
n_walkers int

The number of walkers.

 100
rate float

The rate of steps of a walker.

 1.0
bias float

The probability that a step goes up.

 0.5

Returns:

Name Type Description
system System

The model, with the positions walkers.x.

Examples:

>>> from skcomplex.models.mathematics.stochastic_processes import random_walk
>>> random_walk(20).elements[0].check().prepare().size_
20
Source code in src/skcomplex/models/mathematics/stochastic_processes/_walks.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
@register_model(
    field='mathematics/stochastic_processes',
    tags=('random walk', 'jump process', 'diffusion'),
    references=('Pearson 1905', 'Argun et al. 2021', 'Charbonneau 2017', 'Jensen 2023'),
    defaults={'until': 100.0, 'dt': 1.0},
)
def random_walk(n_walkers: int = 100, *, rate: float = 1.0, bias: float = 0.5) -> System:
    """Return independent walkers on the integers that step at random times, a continuous-time random walk.

    Every walker steps up at rate `rate * bias` and down at rate `rate * (1 - bias)`, so its mean moves with velocity
    `rate (2 bias - 1)` and its variance grows as `4 rate bias (1 - bias) t`.

    Args:
        n_walkers:
            The number of walkers.

        rate:
            The rate of steps of a walker.

        bias:
            The probability that a step goes up.

    Returns:
        system:
            The model, with the positions `walkers.x`.

    Examples:
        >>> from skcomplex.models.mathematics.stochastic_processes import random_walk
        >>> random_walk(20).elements[0].check().prepare().size_
        20
    """
    walkers = GroupElement('walkers', element=RandomWalkerElement, size=n_walkers, values={'rate': rate, 'bias': bias})
    up = JumpInteraction('up', target='walkers', changes={'x': 1}, rate=lambda s: s.target.rate * s.target.bias)
    down = JumpInteraction(
        'down',
        target='walkers',
        changes={'x': -1},
        rate=lambda s: s.target.rate * (1 - s.target.bias),
    )
    return System(elements=[walkers], interactions=[up, down])

random_walk_with_memory(n_walkers=2000, *, alpha=0.5, sigma=1.0)

Return random walks whose velocity is an autoregressive process, x ← x + v and v ← alpha v + xi.

The noise xi is normal with mean 0 and deviation sigma, and the walks start from x = v = 0. The pair (x, v) is Markovian while the displacements are correlated, so the mean squared displacement grows as sigma² t / (1 - alpha)² at long times instead of sigma² t / (1 - alpha²) for uncorrelated steps of the same variance.

Parameters:

Name Type Description Default
n_walkers int

The number of independent walkers.

 2000
alpha float

The memory alpha of the velocity, between 0 and 1.

 0.5
sigma float

The deviation sigma of the noise.

 1.0

Returns:

Name Type Description
system System

The model, with the positions walkers.x and the velocities walkers.v.

Examples:

>>> from skcomplex.models.mathematics.stochastic_processes import random_walk_with_memory
>>> random_walk_with_memory(10).elements[0].check().prepare().size_
10
Source code in src/skcomplex/models/mathematics/stochastic_processes/_walks.py 
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
@register_model(
    field='mathematics/stochastic_processes',
    tags=('random walk', 'discrete map', 'memory', 'autoregressive'),
    references=('Bertin 2021',),
    defaults={'until': 200, 'dt': 1},
)
def random_walk_with_memory(n_walkers: int = 2000, *, alpha: float = 0.5, sigma: float = 1.0) -> System:
    """Return random walks whose velocity is an autoregressive process, `x ← x + v` and `v ← alpha v + xi`.

    The noise `xi` is normal with mean 0 and deviation `sigma`, and the walks start from `x = v = 0`. The pair
    `(x, v)` is Markovian while the displacements are correlated, so the mean squared displacement grows as
    `sigma² t / (1 - alpha)²` at long times instead of `sigma² t / (1 - alpha²)` for uncorrelated steps of the same
    variance.

    Args:
        n_walkers:
            The number of independent walkers.

        alpha:
            The memory `alpha` of the velocity, between 0 and 1.

        sigma:
            The deviation `sigma` of the noise.

    Returns:
        system:
            The model, with the positions `walkers.x` and the velocities `walkers.v`.

    Examples:
        >>> from skcomplex.models.mathematics.stochastic_processes import random_walk_with_memory
        >>> random_walk_with_memory(10).elements[0].check().prepare().size_
        10
    """
    walkers = GroupElement(
        'walkers',
        element=MemoryWalkerElement,
        size=n_walkers,
        values={'alpha': alpha, 'sigma': sigma},
    )
    step = MapInteraction('step', target='walkers', state='x', function=lambda s: s.target.v, mode='increment')
    velocity = MapInteraction(
        'velocity',
        target='walkers',
        state='v',
        function=lambda s: s.target.alpha * s.target.v + normal('xi', 0, s.target.sigma),
    )
    return System(elements=[walkers], interactions=[step, velocity])

sample_space_reducing(n_balls=20, *, n_states=10, lam=1.0)

Return sample-space-reducing processes, whose state visits follow Zipf's law 1 / i for lam = 1.

Parameters:

Name Type Description Default
n_balls int

The number of independent balls, all started at the top state.

 20
n_states int

The number of states.

 10
lam float

The probability of a reducing jump, the exponent of the visiting distribution.

 1.0

Returns:

Name Type Description
system System

The model, with the group balls of state level.

Examples:

>>> from skcomplex.models.mathematics.stochastic_processes import sample_space_reducing
>>> sample_space_reducing(4).elements[0].check().prepare().size_
4
Source code in src/skcomplex/models/mathematics/stochastic_processes/_reducing.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
@register_model(
    field='mathematics/stochastic_processes',
    tags=('sample space reduction', 'agents', 'zipf', 'power law'),
    references=('Corominas-Murtra, Hanel, and Thurner 2015', 'Thurner et al. 2018'),
    defaults={'until': 100},
)
def sample_space_reducing(n_balls: int = 20, *, n_states: int = 10, lam: float = 1.0) -> System:
    """Return sample-space-reducing processes, whose state visits follow Zipf's law `1 / i` for `lam = 1`.

    Args:
        n_balls:
            The number of independent balls, all started at the top state.

        n_states:
            The number of states.

        lam:
            The probability of a reducing jump, the exponent of the visiting distribution.

    Returns:
        system:
            The model, with the group `balls` of state `level`.

    Examples:
        >>> from skcomplex.models.mathematics.stochastic_processes import sample_space_reducing
        >>> sample_space_reducing(4).elements[0].check().prepare().size_
        4
    """
    return System(
        elements=[
            GroupElement(
                'balls',
                element=ReducingBallElement,
                size=n_balls,
                values={'level': float(n_states), 'n_states': float(n_states), 'lam': lam},
                variable=True,
            ),
        ],
        interactions=[ReducingJumpInteraction('jump', target='balls')],
    )

weakening_ornstein_uhlenbeck(n_particles=200, *, k=1.0, ell=1.0, D=0.25)

Return Brownian particles held by a restoring force that weakens with distance.

The particles follow dx = -k x / (1 + x²/ℓ²) dt + √(2D) dW. Near the origin the force is that of an Ornstein-Uhlenbeck process, and far from it the force decays as 1 / x, so the stationary density (1 + x²/ℓ²)^(-k ℓ² / 2D) has power-law tails.

Parameters:

Name Type Description Default
n_particles int

The number of independent particles, all started at the origin.

 200
k float

The stiffness of the force near the origin.

 1.0
ell float

The distance where the force starts to weaken.

 1.0
D float

The diffusion coefficient.

 0.25

Returns:

Name Type Description
system System

The model, with the positions particles.x.

Examples:

>>> from skcomplex.models.mathematics.stochastic_processes import weakening_ornstein_uhlenbeck
>>> weakening_ornstein_uhlenbeck(10).elements[0].check().prepare().size_
10
Source code in src/skcomplex/models/mathematics/stochastic_processes/_walks.py 
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
@register_model(
    field='mathematics/stochastic_processes',
    tags=('random walk', 'sde', 'heavy tails', 'power law'),
    references=('Jensen 2023',),
    defaults={'until': 50.0, 'dt': 0.05},
)
def weakening_ornstein_uhlenbeck(
    n_particles: int = 200,
    *,
    k: float = 1.0,
    ell: float = 1.0,
    D: float = 0.25,
) -> System:
    """Return Brownian particles held by a restoring force that weakens with distance.

    The particles follow `dx = -k x / (1 + x²/ℓ²) dt + √(2D) dW`. Near the origin the force is that of an
    Ornstein-Uhlenbeck process, and far from it the force decays as `1 / x`, so the stationary density
    `(1 + x²/ℓ²)^(-k ℓ² / 2D)` has power-law tails.

    Args:
        n_particles:
            The number of independent particles, all started at the origin.

        k:
            The stiffness of the force near the origin.

        ell:
            The distance where the force starts to weaken.

        D:
            The diffusion coefficient.

    Returns:
        system:
            The model, with the positions `particles.x`.

    Examples:
        >>> from skcomplex.models.mathematics.stochastic_processes import weakening_ornstein_uhlenbeck
        >>> weakening_ornstein_uhlenbeck(10).elements[0].check().prepare().size_
        10
    """
    particles = GroupElement(
        'particles',
        element=WeakeningParticleElement,
        size=n_particles,
        values={'k': k, 'ell': ell, 'D': D},
    )
    restoring = RateInteraction(
        'restoring',
        target='particles',
        state='x',
        rate=lambda s: -s.target.k * s.target.x / (1 + (s.target.x / s.target.ell) ** 2),
        noise=lambda s: sqrt(2 * s.target.D),
    )
    return System(elements=[particles], interactions=[restoring])