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.components.elements import GroupElement
>>> from skcomplex.systems.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/systems/mathematics/stochastic_processes/_walks.py

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.components.elements import GroupElement
>>> from skcomplex.systems.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/systems/mathematics/stochastic_processes/_counting.py

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

CascadingBallElement(label, visual=VisualSpec('agents', {'color_by': 'level'}), *, level=100.0, mu=1.5)

Bases: Element

Ball of a cascading sample-space-reducing process, which multiplies into mu balls at every state it hits.

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.	100.0
mu	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The mean number of balls a ball becomes at every jump, the multiplication factor.	1.5

Examples:

>>> from skcomplex.core.components.elements import GroupElement
>>> from skcomplex.systems.mathematics.stochastic_processes import CascadingBallElement
>>> group = GroupElement('balls', element=CascadingBallElement, size=5, values={'mu': 2.0})
>>> group.label, group.element.__name__, group.size
('balls', 'CascadingBallElement', 5)

Source code in src/skcomplex/systems/mathematics/stochastic_processes/_reducing.py

def __init__(
    self,
    label: str,
    visual: VisualSpec | None = VisualSpec('agents', {'color_by': 'level'}),
    *,
    level: Annotated[float, StateSpec(domain=NN)] = 100.0,
    mu: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.5,
) -> None:
    super().__init__(label, visual=visual)
    self.level = level
    self.mu = mu

jumps(agent, system) staticmethod

Return the state of a new ball, drawn uniformly below the state of its parent.

Parameters:

Name	Type	Description	Default
agent	CascadingBallView	The parent ball.	required
system	SystemView	The system.	required

Returns:

Name	Type	Description
values	dict[str, float]	The state of the new ball.

Source code in src/skcomplex/systems/mathematics/stochastic_processes/_reducing.py

@staticmethod
def jumps(agent: CascadingBallView, system: SystemView) -> dict[str, float]:
    """Return the state of a new ball, drawn uniformly below the state of its parent.

    Args:
        agent:
            The parent ball.

        system:
            The system.

    Returns:
        values:
            The state of the new ball.
    """
    return {'level': float(system.rng.integers(1, round(agent.level)))}

multiplies(agent, system) staticmethod

Return the number of balls a ball becomes, floor(mu) and one more with the probability of the rest.

Parameters:

Name	Type	Description	Default
agent	CascadingBallView	The ball.	required
system	SystemView	The system.	required

Returns:

Name	Type	Description
count	int	The number of new balls, of mean mu.

Source code in src/skcomplex/systems/mathematics/stochastic_processes/_reducing.py

@staticmethod
def multiplies(agent: CascadingBallView, system: SystemView) -> int:
    """Return the number of balls a ball becomes, `floor(mu)` and one more with the probability of the rest.

    Args:
        agent:
            The ball.

        system:
            The system.

    Returns:
        count:
            The number of new balls, of mean `mu`.
    """
    whole = math.floor(agent.mu)
    return whole + int(system.rng.random() < agent.mu - whole)

stops(agent, system) staticmethod

Return whether a ball has reached the ground state 1, where its cascade ends.

Parameters:

Name	Type	Description	Default
agent	CascadingBallView	The ball.	required
system	SystemView	The system.	required

Returns:

Name	Type	Description
stopped	bool	Whether the ball is at state 1.

Source code in src/skcomplex/systems/mathematics/stochastic_processes/_reducing.py

@staticmethod
def stops(agent: CascadingBallView, system: SystemView) -> bool:
    """Return whether a ball has reached the ground state 1, where its cascade ends.

    Args:
        agent:
            The ball.

        system:
            The system.

    Returns:
        stopped:
            Whether the ball is at state 1.
    """
    return agent.level <= 1

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 system 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.components.elements import GroupElement
>>> from skcomplex.systems.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/systems/mathematics/stochastic_processes/_reducing.py

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, system) 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
system	SystemView	The system.	required

Returns:

Name	Type	Description
probability	float	The breaking probability, or zero for a fragment of size 1.

Source code in src/skcomplex/systems/mathematics/stochastic_processes/_reducing.py

@staticmethod
def breaks(agent: FragmentView, system: SystemView) -> float:
    """Return the probability that a fragment breaks in a step, zero at size 1.

    Args:
        agent:
            The fragment.

        system:
            The system.

    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, system) 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
system	SystemView	The system.	required

Returns:

Name	Type	Description
share	float	The share of the size left of the cut.

Source code in src/skcomplex/systems/mathematics/stochastic_processes/_reducing.py

@staticmethod
def cut(agent: FragmentView, system: SystemView) -> float:
    """Return the share of a breaking fragment that stays with it, cut at a uniformly random whole point.

    Args:
        agent:
            The fragment.

        system:
            The system.

    Returns:
        share:
            The share of the size left of the cut.
    """
    length = round(agent.length)
    return int(system.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.systems.mathematics.stochastic_processes import HiddenMarkovSourceElement
>>> HiddenMarkovSourceElement('source', hidden=1).hidden
1

Source code in src/skcomplex/systems/mathematics/stochastic_processes/_sources.py

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 system 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.systems.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/systems/mathematics/stochastic_processes/_branching.py

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.systems.mathematics.stochastic_processes import IntermittentVariableElement
>>> IntermittentVariableElement('variable', D=0.2).D
0.2

Source code in src/skcomplex/systems/mathematics/stochastic_processes/_intermittency.py

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.systems.mathematics.stochastic_processes import MarkovSourceElement
>>> MarkovSourceElement('source', symbol=1).symbol
1

Source code in src/skcomplex/systems/mathematics/stochastic_processes/_sources.py

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.systems.mathematics.stochastic_processes import MemoryWalkerElement
>>> MemoryWalkerElement('walker', alpha=0.3).alpha
0.3

Source code in src/skcomplex/systems/mathematics/stochastic_processes/_walks.py

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

ObservedGrowthElement(label, visual=None, *, x=1.0, x0=1.0, mu=1.0, lam=1.5)

Bases: Element

Quantity that grows exponentially and is observed and restarted at exponentially distributed times.

The quantity follows x' = mu x from x0, and after a waiting time drawn from the exponential distribution of rate lam it is observed and starts again from x0.

Parameters:

Name	Type	Description	Default
label	str	The label of the element 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
x0	Annotated[float, ParameterSpec(domain=open(0, oo))]	The value every growth starts from.	1.0
mu	Annotated[float, ParameterSpec(domain=open(0, oo))]	The growth rate.	1.0
lam	Annotated[float, ParameterSpec(domain=open(0, oo))]	The rate of the observation times.	1.5

Examples:

>>> from skcomplex.systems.mathematics.stochastic_processes import ObservedGrowthElement
>>> ObservedGrowthElement('growth', lam=2.0).lam
2.0

Source code in src/skcomplex/systems/mathematics/stochastic_processes/_power_laws.py

def __init__(
    self,
    label: str,
    visual: VisualSpec | None = None,
    *,
    x: Annotated[float, StateSpec(domain=RealSet.open(0, oo))] = 1.0,
    x0: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 1.0,
    mu: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 1.0,
    lam: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 1.5,
) -> None:
    super().__init__(label, visual=visual)
    self.x = x
    self.x0 = x0
    self.mu = mu
    self.lam = lam

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.systems.mathematics.stochastic_processes import PersistentWalkerElement
>>> PersistentWalkerElement('walker', persistence=0.9).persistence
0.9

Source code in src/skcomplex/systems/mathematics/stochastic_processes/_walks.py

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.systems.mathematics.stochastic_processes import PoissonCounterElement
>>> PoissonCounterElement('counter', rate=2.0).rate
2.0

Source code in src/skcomplex/systems/mathematics/stochastic_processes/_counting.py

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 system 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.systems.mathematics.stochastic_processes import PolyaUrnElement
>>> PolyaUrnElement('urn', added=2).added
2

Source code in src/skcomplex/systems/mathematics/stochastic_processes/_urns.py

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 system 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.systems.mathematics.stochastic_processes import QueueElement
>>> QueueElement('desk', arrival_rate=0.8).arrival_rate
0.8

Source code in src/skcomplex/systems/mathematics/stochastic_processes/_queues.py

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

RandomBranchMapElement(label, visual=None, *, x=0.5, p1=0.5, mu=1.0)

Bases: Element

Value of a random interval map, which applies (1 + x) / 2 with probability p1 and mu x² otherwise.

Parameters:

Name	Type	Description	Default
label	str	The label of the element or the group.	required
visual	VisualSpec | None	How the element is drawn, or None for default_visual_.	None
x	Annotated[float, StateSpec(domain=closed(0, 1))]	The initial value, between 0 and 1.	0.5
p1	Annotated[float, ParameterSpec(domain=closed(0, 1))]	The probability of the branch (1 + x) / 2.	0.5
mu	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The coefficient of the branch mu x², between 0 and 1.	1.0

Examples:

>>> from skcomplex.systems.mathematics.stochastic_processes import RandomBranchMapElement
>>> RandomBranchMapElement('map', p1=0.3).p1
0.3

Source code in src/skcomplex/systems/mathematics/stochastic_processes/_random_maps.py

def __init__(
    self,
    label: str,
    visual: VisualSpec | None = None,
    *,
    x: Annotated[float, StateSpec(domain=RealSet.closed(0, 1))] = 0.5,
    p1: Annotated[float, ParameterSpec(domain=RealSet.closed(0, 1))] = 0.5,
    mu: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
) -> None:
    super().__init__(label, visual=visual)
    self.x = x
    self.p1 = p1
    self.mu = mu

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.systems.mathematics.stochastic_processes import RandomWalkerElement
>>> RandomWalkerElement('walker', bias=0.7).bias
0.7

Source code in src/skcomplex/systems/mathematics/stochastic_processes/_walks.py

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.components.elements import GroupElement
>>> from skcomplex.systems.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/systems/mathematics/stochastic_processes/_reducing.py

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.systems.mathematics.stochastic_processes import ReducingJumpInteraction
>>> ReducingJumpInteraction('jump', target='balls').state
'level'

Source code in src/skcomplex/systems/mathematics/stochastic_processes/_reducing.py

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.systems.mathematics.stochastic_processes import WeakeningParticleElement
>>> WeakeningParticleElement('particle', D=0.5).D
0.5

Source code in src/skcomplex/systems/mathematics/stochastic_processes/_walks.py

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 system, with the group walkers.

Examples:

>>> from skcomplex.systems.mathematics.stochastic_processes import ageing_random_walk
>>> ageing_random_walk(5).elements[0].prepare().size_
5

Source code in src/skcomplex/systems/mathematics/stochastic_processes/_walks.py

@register_system(
    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 system, with the group `walkers`.

    Examples:
        >>> from skcomplex.systems.mathematics.stochastic_processes import ageing_random_walk
        >>> ageing_random_walk(5).elements[0].prepare().size_
        5
    """
    ageing = MapInteraction(
        'ageing',
        target='walkers',
        state='ups',
        function=lambda agent, system: float(system.rng.random() < agent.p / (1 + agent.ups) ** agent.nu),
        mode='increment',
        form='agent',
    )
    step = MapInteraction(
        'step',
        target='walkers',
        state='x',
        function=lambda agent, system: 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 system, with the group trials of states outcome and successes.

Examples:

>>> from skcomplex.systems.mathematics.stochastic_processes import bernoulli_process
>>> bernoulli_process(3).elements[0].prepare().size_
3

Source code in src/skcomplex/systems/mathematics/stochastic_processes/_counting.py

@register_system(
    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 system, with the group `trials` of states `outcome` and `successes`.

    Examples:
        >>> from skcomplex.systems.mathematics.stochastic_processes import bernoulli_process
        >>> bernoulli_process(3).elements[0].prepare().size_
        3
    """
    trial = MapInteraction(
        'trial',
        target='trials',
        state='outcome',
        function=lambda agent, system: float(system.rng.random() < agent.p),
        form='agent',
    )
    count = MapInteraction(
        'count',
        target='trials',
        state='successes',
        function=lambda agent, system: 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 system, with the lengths queues.n.

Examples:

>>> from skcomplex.systems.mathematics.stochastic_processes import birth_death_queue
>>> birth_death_queue(5).elements[0].prepare().size_
5

Source code in src/skcomplex/systems/mathematics/stochastic_processes/_queues.py

@register_system(
    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 system, with the lengths `queues.n`.

    Examples:
        >>> from skcomplex.systems.mathematics.stochastic_processes import birth_death_queue
        >>> birth_death_queue(5).elements[0].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])

cascading_sample_space_reducing(n_cascades=100, *, n_states=100, mu=1.5)

Return cascading sample-space-reducing processes, whose visits follow the power law i^(-mu) for any mu.

Every cascade starts with one ball at the top state n_states. At every step each ball above state 1 jumps to a state drawn uniformly below its own and becomes mu balls there, floor(mu) and one more with the probability of the rest, and a ball at state 1 leaves the cascade. Every ball stays one step, so the states of the balls of all frames are the visits, distributed as i^(-mu) (Thurner et al. 2018, section 3.3.5.4). For mu < 1 balls are lost on the way, 1 - mu being the probability of a removal. With a conservation law that makes the states of the new balls sum to the state of their parent, the exponent is 2 whatever mu, the case of fragmentation.

Parameters:

Name	Type	Description	Default
n_cascades	int	The number of independent cascades, run together.	100
n_states	int	The number of states.	100
mu	float	The multiplication factor, the mean number of balls a ball becomes at every jump.	1.5

Returns:

Name	Type	Description
system	System	The system, with the group balls of state level.

Examples:

>>> from skcomplex.systems.mathematics.stochastic_processes import cascading_sample_space_reducing
>>> cascading_sample_space_reducing(4).elements[0].prepare().size_
4

Source code in src/skcomplex/systems/mathematics/stochastic_processes/_reducing.py

@register_system(
    field='mathematics/stochastic_processes',
    tags=('sample space reduction', 'agents', 'cascade', 'power law'),
    references=('Corominas-Murtra, Hanel, and Thurner 2017', 'Thurner et al. 2018'),
    defaults={'until': 100},
)
def cascading_sample_space_reducing(n_cascades: int = 100, *, n_states: int = 100, mu: float = 1.5) -> System:
    """Return cascading sample-space-reducing processes, whose visits follow the power law `i^(-mu)` for any `mu`.

    Every cascade starts with one ball at the top state `n_states`. At every step each ball above state 1 jumps to a
    state drawn uniformly below its own and becomes `mu` balls there, `floor(mu)` and one more with the probability of
    the rest, and a ball at state 1 leaves the cascade. Every ball stays one step, so the states of the balls of all
    frames are the visits, distributed as `i^(-mu)` (Thurner et al. 2018, section 3.3.5.4). For `mu < 1` balls are
    lost on the way, `1 - mu` being the probability of a removal. With a conservation law that makes the states of
    the new balls sum to the state of their parent, the exponent is 2 whatever `mu`, the case of `fragmentation`.

    Args:
        n_cascades:
            The number of independent cascades, run together.

        n_states:
            The number of states.

        mu:
            The multiplication factor, the mean number of balls a ball becomes at every jump.

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

    Examples:
        >>> from skcomplex.systems.mathematics.stochastic_processes import cascading_sample_space_reducing
        >>> cascading_sample_space_reducing(4).elements[0].prepare().size_
        4
    """
    return System(
        elements=[
            GroupElement(
                'balls',
                element=CascadingBallElement,
                size=n_cascades,
                values={'level': float(n_states), 'mu': mu},
                variable=True,
            ),
        ],
        interactions=[
            LifeCycleInteraction(
                'cascade',
                target='balls',
                death=CascadingBallElement.stops,
                birth=1.0,
                offspring=CascadingBallElement.multiplies,
                inherit=('mu',),
                values=CascadingBallElement.jumps,
                replace=True,
            ),
        ],
    )

exponential_growth_at_random_times(n_samples=2000, *, mu=1.0, lam=1.5, x0=1.0)

Return exponential growth observed at exponentially distributed times, whose observed values follow a power law.

Every sample grows as x0 exp(mu t) and is observed and restarted after a waiting time drawn from the exponential distribution of rate lam. The observed values, and the values of the samples at any later time, follow the power law p(x) ~ x^(-lam / mu - 1) (Thurner et al. 2018, section 3.3.6.1, after Reed and Hughes 2002).

Parameters:

Name	Type	Description	Default
n_samples	int	The number of independent samples.	2000
mu	float	The growth rate.	1.0
lam	float	The rate of the observation times.	1.5
x0	float	The value every growth starts from.	1.0

Returns:

Name	Type	Description
system	System	The system, with the group samples of state x and the observations samples.observations.

Examples:

>>> from skcomplex.systems.mathematics.stochastic_processes import exponential_growth_at_random_times
>>> exponential_growth_at_random_times(10).elements[0].prepare().size_
10

Source code in src/skcomplex/systems/mathematics/stochastic_processes/_power_laws.py

@register_system(
    field='mathematics/stochastic_processes',
    tags=('power law', 'renewal', 'exponential growth', 'random observation'),
    references=('Reed and Hughes 2002', 'Thurner et al. 2018'),
    defaults={'until': 20.0, 'dt': 1.0, 'solver': RK4Solver(dt=0.01)},
)
def exponential_growth_at_random_times(
    n_samples: int = 2000,
    *,
    mu: float = 1.0,
    lam: float = 1.5,
    x0: float = 1.0,
) -> System:
    """Return exponential growth observed at exponentially distributed times, whose observed values follow a power law.

    Every sample grows as `x0 exp(mu t)` and is observed and restarted after a waiting time drawn from the
    exponential distribution of rate `lam`. The observed values, and the values of the samples at any later time,
    follow the power law `p(x) ~ x^(-lam / mu - 1)` (Thurner et al. 2018, section 3.3.6.1, after Reed and Hughes 2002).

    Args:
        n_samples:
            The number of independent samples.

        mu:
            The growth rate.

        lam:
            The rate of the observation times.

        x0:
            The value every growth starts from.

    Returns:
        system:
            The system, with the group `samples` of state `x` and the observations `samples.observations`.

    Examples:
        >>> from skcomplex.systems.mathematics.stochastic_processes import exponential_growth_at_random_times
        >>> exponential_growth_at_random_times(10).elements[0].prepare().size_
        10
    """
    return System(
        elements=[
            GroupElement(
                'samples',
                element=ObservedGrowthElement,
                size=n_samples,
                values={'x': x0, 'x0': x0, 'mu': mu, 'lam': lam},
            ),
        ],
    )

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 system, with the group fragments of state length.

Examples:

>>> from skcomplex.systems.mathematics.stochastic_processes import fragmentation
>>> fragmentation(50).elements[0].prepare().size_
1

Source code in src/skcomplex/systems/mathematics/stochastic_processes/_reducing.py

@register_system(
    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 system, with the group `fragments` of state `length`.

    Examples:
        >>> from skcomplex.systems.mathematics.stochastic_processes import fragmentation
        >>> fragmentation(50).elements[0].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 system, with the group population, whose count is the size of the current generation.

Examples:

>>> from skcomplex.systems.mathematics.stochastic_processes import galton_watson
>>> galton_watson(5, m=1.2).elements[0].prepare().size_
5

Source code in src/skcomplex/systems/mathematics/stochastic_processes/_branching.py

@register_system(
    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 system, with the group `population`, whose count is the size of the current generation.

    Examples:
        >>> from skcomplex.systems.mathematics.stochastic_processes import galton_watson
        >>> galton_watson(5, m=1.2).elements[0].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, system: int(system.rng.poisson(agent.m)),
                values=lambda agent, system: {'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 system, with the hidden states source.hidden and the symbols source.symbol.

Examples:

>>> from skcomplex.systems.mathematics.stochastic_processes import hidden_markov_source
>>> hidden_markov_source().interactions[0].observed
'symbol'

Source code in src/skcomplex/systems/mathematics/stochastic_processes/_sources.py

@register_system(
    field='mathematics/stochastic_processes',
    tags=('hidden markov system', '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 system, with the hidden states `source.hidden` and the symbols `source.symbol`.

    Examples:
        >>> from skcomplex.systems.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 system, with the symbols source.symbol.

Examples:

>>> from skcomplex.systems.mathematics.stochastic_processes import markov_source
>>> markov_source().interactions[0].state
'symbol'

Source code in src/skcomplex/systems/mathematics/stochastic_processes/_sources.py

@register_system(
    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 system, with the symbols `source.symbol`.

    Examples:
        >>> from skcomplex.systems.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 system, with the values variables.x.

Examples:

>>> from skcomplex.systems.mathematics.stochastic_processes import on_off_intermittency
>>> on_off_intermittency(5).elements[0].size
5

Source code in src/skcomplex/systems/mathematics/stochastic_processes/_intermittency.py

@register_system(
    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 system, with the values `variables.x`.

    Examples:
        >>> from skcomplex.systems.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 system, with the directions walkers.d and the positions walkers.x.

Examples:

>>> from skcomplex.systems.mathematics.stochastic_processes import persistent_random_walk
>>> persistent_random_walk(10).elements[0].prepare().size_
10

Source code in src/skcomplex/systems/mathematics/stochastic_processes/_walks.py

@register_system(
    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 system, with the directions `walkers.d` and the positions `walkers.x`.

    Examples:
        >>> from skcomplex.systems.mathematics.stochastic_processes import persistent_random_walk
        >>> persistent_random_walk(10).elements[0].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 system, with the counts counters.count.

Examples:

>>> from skcomplex.systems.mathematics.stochastic_processes import poisson_process
>>> poisson_process(5).elements[0].prepare().size_
5

Source code in src/skcomplex/systems/mathematics/stochastic_processes/_counting.py

@register_system(
    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 system, with the counts `counters.count`.

    Examples:
        >>> from skcomplex.systems.mathematics.stochastic_processes import poisson_process
        >>> poisson_process(5).elements[0].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 system, with the states urns.red and urns.blue.

Examples:

>>> from skcomplex.systems.mathematics.stochastic_processes import polya_urn
>>> polya_urn(3).elements[0].prepare().size_
3

Source code in src/skcomplex/systems/mathematics/stochastic_processes/_urns.py

@register_system(
    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 system, with the states `urns.red` and `urns.blue`.

    Examples:
        >>> from skcomplex.systems.mathematics.stochastic_processes import polya_urn
        >>> polya_urn(3).elements[0].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_branch_map(n_orbits=200, *, p1=0.5, mu=1.0, random_state=0)

Return orbits of a random map of [0, 1] with the two branches (1 + x) / 2 and mu x².

At every step each orbit applies (1 + x) / 2 with probability p1 and mu x² otherwise (Dilão 2023, Exercise 5.10). The first branch pulls towards 1 and the second towards 0, so the orbits settle on a stationary distribution whose shape changes with p1 and mu. For mu = 0 the second branch resets the orbit to 0, and the stationary mean is p1 / (2 - p1). For mu = 1 both branches fix x = 1, which attracts the orbits once p1 > 1/2, since there the first branch halves the distance to 1 and the second doubles it.

Parameters:

Name	Type	Description	Default
n_orbits	int	The number of independent orbits.	200
p1	float	The probability of the branch (1 + x) / 2.	0.5
mu	float	The coefficient of the branch mu x², between 0 and 1.	1.0
random_state	int | Generator | None	The seed or generator of the initial values, uniform on [0, 1].	0

Returns:

Name	Type	Description
system	System	The system, with the group orbits of state x.

Examples:

>>> from skcomplex.systems.mathematics.stochastic_processes import random_branch_map
>>> random_branch_map(10, p1=0.3).elements[0].prepare().size_
10

Source code in src/skcomplex/systems/mathematics/stochastic_processes/_random_maps.py

@register_system(
    field='mathematics/stochastic_processes',
    tags=('random map', 'iterated function system', 'interval map', 'stationary distribution'),
    references=('Dilão 2023',),
    defaults={'until': 200, 'dt': 1},
)
def random_branch_map(
    n_orbits: int = 200,
    *,
    p1: float = 0.5,
    mu: float = 1.0,
    random_state: int | np.random.Generator | None = 0,
) -> System:
    """Return orbits of a random map of `[0, 1]` with the two branches `(1 + x) / 2` and `mu x²`.

    At every step each orbit applies `(1 + x) / 2` with probability `p1` and `mu x²` otherwise (Dilão 2023, Exercise
    5.10). The first branch pulls towards 1 and the second towards 0, so the orbits settle on a stationary
    distribution whose shape changes with `p1` and `mu`. For `mu = 0` the second branch resets the orbit to 0, and
    the stationary mean is `p1 / (2 - p1)`. For `mu = 1` both branches fix `x = 1`, which attracts the orbits once
    `p1 > 1/2`, since there the first branch halves the distance to 1 and the second doubles it.

    Args:
        n_orbits:
            The number of independent orbits.

        p1:
            The probability of the branch `(1 + x) / 2`.

        mu:
            The coefficient of the branch `mu x²`, between 0 and 1.

        random_state:
            The seed or generator of the initial values, uniform on `[0, 1]`.

    Returns:
        system:
            The system, with the group `orbits` of state `x`.

    Examples:
        >>> from skcomplex.systems.mathematics.stochastic_processes import random_branch_map
        >>> random_branch_map(10, p1=0.3).elements[0].prepare().size_
        10
    """
    start = np.random.default_rng(random_state).uniform(0.0, 1.0, n_orbits)
    return System(
        elements=[
            GroupElement(
                'orbits',
                element=RandomBranchMapElement,
                size=n_orbits,
                values={'x': start, 'p1': p1, 'mu': mu},
            ),
        ],
    )

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 system, with the positions walkers.x.

Examples:

>>> from skcomplex.systems.mathematics.stochastic_processes import random_walk
>>> random_walk(20).elements[0].prepare().size_
20

Source code in src/skcomplex/systems/mathematics/stochastic_processes/_walks.py

@register_system(
    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 system, with the positions `walkers.x`.

    Examples:
        >>> from skcomplex.systems.mathematics.stochastic_processes import random_walk
        >>> random_walk(20).elements[0].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 system, with the positions walkers.x and the velocities walkers.v.

Examples:

>>> from skcomplex.systems.mathematics.stochastic_processes import random_walk_with_memory
>>> random_walk_with_memory(10).elements[0].prepare().size_
10

Source code in src/skcomplex/systems/mathematics/stochastic_processes/_walks.py

@register_system(
    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 system, with the positions `walkers.x` and the velocities `walkers.v`.

    Examples:
        >>> from skcomplex.systems.mathematics.stochastic_processes import random_walk_with_memory
        >>> random_walk_with_memory(10).elements[0].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 system, with the group balls of state level.

Examples:

>>> from skcomplex.systems.mathematics.stochastic_processes import sample_space_reducing
>>> sample_space_reducing(4).elements[0].prepare().size_
4

Source code in src/skcomplex/systems/mathematics/stochastic_processes/_reducing.py

@register_system(
    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 system, with the group `balls` of state `level`.

    Examples:
        >>> from skcomplex.systems.mathematics.stochastic_processes import sample_space_reducing
        >>> sample_space_reducing(4).elements[0].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 system, with the positions particles.x.

Examples:

>>> from skcomplex.systems.mathematics.stochastic_processes import weakening_ornstein_uhlenbeck
>>> weakening_ornstein_uhlenbeck(10).elements[0].prepare().size_
10

Source code in src/skcomplex/systems/mathematics/stochastic_processes/_walks.py

@register_system(
    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 system, with the positions `particles.x`.

    Examples:
        >>> from skcomplex.systems.mathematics.stochastic_processes import weakening_ornstein_uhlenbeck
        >>> weakening_ornstein_uhlenbeck(10).elements[0].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])