API Reference

processes

BrownianMotion

Bases: StochasticProcess

A class representing a Brownian motion.

Parameters:

Name	Type	Description	Default
`time`	`Time \| None`	The time index of the stochastic process. If `None`, then it will be generated in the `from_simulation` method.	`None`
`domain`	`SampleSpace \| None`	The sample space representing the domain of the stochastic process. If `None`, then it will be generated in the `from_simulation` method.	`None`
`name`	`Hashable \| None`	The name of the stochastic process.	`"X"`

Examples:

>>> from sigalg.core import Time
>>> from sigalg.processes import BrownianMotion
>>> T = Time.continuous(start=0.1, stop=1.1, dt=0.35)
>>> X = BrownianMotion(time=T).from_simulation(n_trajectories=4, random_state=42)
>>> print(X)
Stochastic process 'X':
time        0.100000  0.433333  0.766667  1.100000
trajectory
0                0.0  0.175928 -0.424507  0.008767
1                0.0  0.543035 -0.583395 -1.335209
2                0.0  0.073809 -0.108774 -0.118474
3                0.0 -0.492505  0.015216  0.464274

IIDProcess

Bases: StochasticProcess

A class representing an Independent and Identically Distributed (IID) stochastic process.

The is_discrete_state attribute from the parent class StochasticProcess is automatically determined based on whether the provided distribution is discrete or continuous.

Parameters:

Name	Type	Description	Default
`distribution`	`rv_frozen`	A frozen random variable from scipy.stats representing the common distribution of the IID process.	required
`time`	`Time \| None`	The time index of the stochastic process. If `None`, then the `is_discrete_time` property must be provided.	`None`
`is_discrete_time`	`bool \| None`	Whether the stochastic process is a discrete-time process. If `None`, then `time` parameter must be provided.	`None`
`domain`	`SampleSpace \| None`	The sample space representing the domain of the stochastic process. If `None`, it will be generated later through data generation methods.	`None`
`name`	`Hashable \| None`	The name of the stochastic process.	`"X"`

Raises:

Type	Description
`TypeError`	If `rv` is not an instance of `rv_frozen`.

Examples:

>>> from scipy.stats import bernoulli
>>> from sigalg.core import SampleSpace, Time
>>> from sigalg.processes import IIDProcess
>>> domain = SampleSpace().from_sequence(size=3, prefix="omega")
>>> time = Time.discrete(length=2)
>>> # Construct Bernoulli IID process via exhaustive enumeration
>>> X = IIDProcess(distribution=bernoulli(p=0.25), support=[0, 1], time=time).from_enumeration()
>>> X
Stochastic process 'X':
time  0  1  2
trajectory
0     0  0  0
1     0  0  1
2     0  1  0
3     0  1  1
4     1  0  0
5     1  0  1
6     1  1  0
7     1  1  1
>>> # Generate the exact probability measure associated with the enumerated process
>>> P = X.probability_measure
>>> P
Probability measure 'P':
        probability
trajectory
0        0.421875
1        0.140625
2        0.140625
3        0.046875
4        0.140625
5        0.046875
6        0.046875
7        0.015625
>>> # Construct Poisson IID process via simulation, with non-specified domain and time index
>>> from scipy.stats import poisson
>>> Y = IIDProcess(distribution=poisson(mu=1.0), is_discrete_time=True, name="Y").from_simulation(
...     n_trajectories=10_000, random_state=42, length=2
... )
>>> Y
Stochastic process 'Y':
time  0  1  2
trajectory
0     1  2  3
1     1  3  0
2     1  3  3
3     1  0  3
4     1  0  0
...  .. .. ..
9995  1  2  2
9996  0  3  0
9997  0  2  1
9998  1  3  2
9999  1  2  2

[10000 rows x 3 columns]

MarkovChain

Bases: StochasticProcess

A class representing a Markov chain stochastic process.

Parameters:

Name	Type	Description	Default
`transition_matrix`	`DataFrame`	A DataFrame representing the transition probabilities between states. The index and columns should correspond to the states of the Markov chain, and each row should sum to `1`.	required
`initial_distribution`	`ProbabilityMeasure`	A ProbabilityMeasure representing the initial distribution over the states of the Markov chain. Its sample space should match the states defined in the transition matrix.	required
`time`	`Time \| None`	The time index of the stochastic process. If `None`, then the `is_discrete_time` property must be provided.	`None`
`is_discrete_time`	`bool \| None`	Whether the stochastic process is a discrete-time process. If `None`, then `time` parameter must be provided.	`None`
`domain`	`SampleSpace \| None`	The sample space representing the domain of the stochastic process. If `None`, it will be generated later through data generation methods.	`None`
`name`	`Hashable \| None`	The name of the stochastic process.	`"X"`

Raises:

Type	Description
`TypeError`	If `transition_matrix` is not a pandas DataFrame or if `initial_distribution` is not a ProbabilityMeasure.
`ValueError`	If the index and columns of `transition_matrix` do not match the sample space of `initial_distribution`, if any row of `transition_matrix` does not sum to `1`, or if any entry in `transition_matrix` is negative.

Examples:

>>> import pandas as pd
>>> from sigalg.core import ProbabilityMeasure, SampleSpace
>>> from sigalg.processes import MarkovChain
>>> state_space = SampleSpace().from_list(["rain", "sun"])
>>> P = pd.DataFrame(
...     data=[
...         [0.9, 0.1],  # P(rain | rain) = 0.9, P(sun | rain) = 0.1
...         [0.4, 0.6],  # P(rain | sun) = 0.4, P(sun | sun) = 0.6
...     ],
...     index=state_space,
...     columns=state_space,
... )
>>> pi = ProbabilityMeasure(name="pi").from_dict({"rain": 0.25, "sun": 0.75})
>>> X = MarkovChain(
...     transition_matrix=P,
...     initial_distribution=pi,
...     is_discrete_time=True,
...     name="X",
... ).from_simulation(
...     n_trajectories=100_000,
...     length=2,
...     random_state=42,
... )
>>> X
Stochastic process 'X':
time      0     1     2
trajectory
0       sun   sun   sun
1       sun   sun  rain
2       sun   sun   sun
3       sun  rain  rain
4      rain  rain  rain
...     ...   ...   ...
99995   sun  rain  rain
99996   sun   sun  rain
99997   sun  rain  rain
99998  rain  rain  rain
99999   sun  rain  rain

[100000 rows x 3 columns]

PoissonProcess

Bases: StochasticProcess

A class representing a Poisson process.

The Poisson process is a process {X_t} where X_t counts the number of events that have occurred by time t. The rate parameter represents the average number of events per unit time.

In this implementation, trajectories are simulated until one trajectory reaches the specified max_count of events, and then the (required) user-provided time index is truncated to the length of this shortest complete trajectory.

If t_stop is the last time value in the time index, then a good choice for max_count is approximately rate * t_stop + 3 * sqrt(rate * t_stop), which is the mean of X_{rate * t_stop} (a Poisson random variable) plus 3 times its standard deviation.

The trajectories of Poisson processes are right-continuous step functions that jump by 1 at each event time. In order to plot these trajectories accurately, the user should select a continuous time index with a sufficiently large number of points.

The from_enumeration method is not implemented for PoissonProcess since it is a continuous-time process.

Parameters:

Name	Type	Description	Default
`rate`	`Real`	The rate (lambda) of the Poisson process, which must be a positive real number.	required
`max_count`	`int`	The maximum count of events to simulate, which must be a positive integer.	required
`time`	`Time`	The time index of the stochastic process.	required
`domain`	`SampleSpace \| None`	The sample space representing the domain of the stochastic process. If `None`, it will be generated later through data generation methods.	`None`
`name`	`Hashable \| None`	The name of the stochastic process.	`"X"`

Raises:

Type	Description
`TypeError`	If `rate` is not a positive real number or if `max_count` is not a positive integer.

Examples:

>>> from math import ceil, sqrt
>>> from scipy.stats import poisson
>>> from sigalg.core import Time
>>> from sigalg.processes import PoissonProcess
>>> # Parameters for the continuous time index. We select a very coarse time grid for printing purposes in the docstrings.
>>> start = 0.0
>>> stop = 6.25
>>> num_points = 5
>>> time = Time.continuous(
...     start=start,
...     stop=stop,
...     num_points=num_points,
... )
>>> # Parameters for the Poisson process. The max_count parameter follows the suggested rule of thumb described above.
>>> rate = 9.5
>>> max_count = ceil(rate * stop + 3 * sqrt(rate * stop))
>>> max_count
83
>>> # Simulate 10 trajectories of the Poisson process with the specified parameters and print them.
>>> X = PoissonProcess(rate=rate, max_count=max_count, time=time).from_simulation(
...     n_trajectories=10, random_state=42
... )
>>> X
Stochastic process 'X':
time        0.0000  1.5625  3.1250  4.6875  6.2500
trajectory
0              0.0    11.0    32.0    54.0    64.0
1              0.0    17.0    32.0    50.0    63.0
2              0.0    14.0    27.0    44.0    62.0
3              0.0    23.0    42.0    60.0    75.0
4              0.0    11.0    20.0    37.0    45.0
5              0.0    11.0    25.0    37.0    54.0
6              0.0    14.0    33.0    48.0    60.0
7              0.0     9.0    19.0    28.0    42.0
8              0.0    19.0    26.0    41.0    62.0
9              0.0     7.0    21.0    37.0    55.0
>>> # Simulate a Poisson process using 50,000 trajectories
>>> Y = PoissonProcess(
...     rate=rate, max_count=max_count, time=time, name="Y"
... ).from_simulation(n_trajectories=50_000, random_state=42)
>>> # Extract the simulated values of the final random variable Y_last
>>> final_counts = Y.last_rv.range
>>> simulated_outputs = final_counts.data
>>> # Extract the empirical probabilities of the final random variable Y_last
>>> simulated_probabilities = final_counts.probability_measure.data
>>> # Get the final time point, compute the theoretical probabilities of the final random variable Y_last, a Poisson random variable
>>> final_time = Y.time[-1]
>>> theoretical_probabilities = poisson(mu=rate * final_time).pmf(simulated_outputs)
>>> # Compare the simulated probabilities with the theoretical probabilities
>>> round(float(abs(simulated_probabilities - theoretical_probabilities).sum()), 4)
0.02

ProcessTransforms

A collection of methods for transforming stochastic processes.

cumprod `staticmethod`

cumprod(process, name=None)

Compute the cumulative product of a stochastic process along its time index.

Parameters:

Name	Type	Description	Default
`process`	`StochasticProcess`	The stochastic process for which to compute the cumulative product.	required
`name`	`Hashable \| None`	The name of the transformed process. If `None`, the new name will be the name of the input process subscripted with `cumprod`, provided that the name of the input process is a string.	`None`

Raises:

Type	Description
`TypeError`	If `process` is not an instance of `StochasticProcess`.

Returns:

Name	Type	Description
`cumprod_process`	`StochasticProcess`	A new stochastic process representing the cumulative product of the input process.

Examples:

>>> from sigalg.core import Time
>>> from sigalg.processes import RandomWalk
>>> T = Time.discrete(length=3)
>>> X = RandomWalk(p=0.5, time=T, initial_state=3).from_enumeration()
>>> print(X)
Stochastic process 'X':
time        0  1  2  3
trajectory
0           3  2  1  0
1           3  2  1  2
2           3  2  3  2
3           3  2  3  4
4           3  4  3  2
5           3  4  3  4
6           3  4  5  4
7           3  4  5  6
>>> print(X.cumprod())
Stochastic process 'X_cumprod':
time        0   1   2    3
trajectory
0           3   6   6    0
1           3   6   6   12
2           3   6  18   36
3           3   6  18   72
4           3  12  36   72
5           3  12  36  144
6           3  12  60  240
7           3  12  60  360

cumsum `staticmethod`

cumsum(process, name=None)

Compute the cumulative sum of a stochastic process along its time index.

Parameters:

Name	Type	Description	Default
`process`	`StochasticProcess`	The stochastic process for which to compute the cumulative sum.	required
`name`	`Hashable \| None`	The name of the transformed process. If `None`, the new name will be the name of the input process subscripted with `cumsum`, provided that the name of the input process is a string.	`None`

Raises:

Type	Description
`TypeError`	If `process` is not an instance of `StochasticProcess`.

Returns:

Name	Type	Description
`cumsum_process`	`StochasticProcess`	A new stochastic process representing the cumulative sum of the input process.

Examples:

>>> from scipy.stats import bernoulli
>>> from sigalg.core import Time
>>> from sigalg.processes import IIDProcess
>>> T = Time.discrete(start=1, length=2)
>>> X = IIDProcess(distribution=bernoulli(p=0.6), support=[0, 1], time=T).from_enumeration()
>>> print(X)
Stochastic process 'X':
time        1  2  3
trajectory
0           0  0  0
1           0  0  1
2           0  1  0
3           0  1  1
4           1  0  0
5           1  0  1
6           1  1  0
7           1  1  1
>>> print(X.cumsum())
Stochastic process 'X_cumsum':
time        1  2  3
trajectory
0           0  0  0
1           0  0  1
2           0  1  1
3           0  1  2
4           1  1  1
5           1  1  2
6           1  2  2
7           1  2  3

increments `staticmethod`

increments(process, forward=False, name=None)

Compute the increments of a stochastic process along its time index.

Parameters:

Name	Type	Description	Default
`process`	`StochasticProcess`	The stochastic process for which to compute the increments.	required
`forward`	`bool`	If `True`, compute forward increments, i.e., X(t) is replaced with X(t+1) - X(t). If `False`, compute backward increments, i.e., X(t) is replaced with X(t) - X(t-1).	`False`
`name`	`Hashable \| None`	The name of the transformed process. If `None`, the new name will be the name of the input process subscripted with `increments`, provided that the name of the input process is a string.	`None`

Raises:

Type	Description
`TypeError`	If `process` is not an instance of `StochasticProcess`.
`ValueError`	If `process` is one-dimensional.

Returns:

Name	Type	Description
`increments_process`	`StochasticProcess`	A new stochastic process representing the increments of the input process.

Examples:

>>> from sigalg.core import Time
>>> from sigalg.processes import RandomWalk
>>> T = Time.discrete(length=2)
>>> X = RandomWalk(p=0.5, time=T, initial_state=3).from_enumeration()
>>> print(X)
Stochastic process 'X':
time        0  1  2
trajectory
0           3  2  1
1           3  2  3
2           3  4  3
3           3  4  5
>>> print(X.increments())
Stochastic process 'X_increments':
time        1  2
trajectory
0          -1 -1
1          -1  1
2           1 -1
3           1  1

insert_rv `staticmethod`

insert_rv(process, rv, time, name=None)

Insert a random variable to a stochastic process at a specific time.

Parameters:

Name	Type	Description	Default
`process`	`StochasticProcess`	The stochastic process to which the random variable will be inserted.	required
`rv`	`RandomVariable`	The random variable to insert.	required
`time`	`Real`	The time at which to insert the random variable.	required
`name`	`Hashable \| None`	The name of the new stochastic process. If `None`, the new name will be `insert(process.name)` if `process.name` is not `None`.	`None`

Raises:

Type	Description
`TypeError`	If `process` is not an instance of `StochasticProcess`, or `rv` is not an instance of `RandomVariable`, or `time` is not a real number.
`ValueError`	If `process` has no data, or if `process` and `rv` do not have the same domain.

Returns:

Name	Type	Description
`inserted_process`	`StochasticProcess`	A new stochastic process with the random variable inserted at the specified time.

Examples:

>>> from scipy.stats import bernoulli
>>> from sigalg.core import RandomVariable, Time
>>> from sigalg.processes import IIDProcess
>>> T = Time().discrete(start=1, length=2)
>>> X = IIDProcess(distribution=bernoulli(p=0.5), support=[0, 1], time=T).from_enumeration()
>>> print(X)
Stochastic process 'X':
time        1  2  3
trajectory
0           0  0  0
1           0  0  1
2           0  1  0
3           0  1  1
4           1  0  0
5           1  0  1
6           1  1  0
7           1  1  1
>>> X0 = RandomVariable(domain=X.domain).from_constant(0)
>>> print(X.insert_rv(rv=X0, time=0))
Stochastic process 'insert(X)':
time        0  1  2  3
trajectory
0           0  0  0  0
1           0  0  0  1
2           0  0  1  0
3           0  0  1  1
4           0  1  0  0
5           0  1  0  1
6           0  1  1  0
7           0  1  1  1

is_monotonic `staticmethod`

is_monotonic(process, increasing=True)

Check if the trajectories of a stochastic process are monotonic.

Parameters:

Name	Type	Description	Default
`process`	`StochasticProcess`	The stochastic process to check for monotonicity.	required
`increasing`	`bool`	If `True`, check for monotonically increasing trajectories; if `False`, check for monotonically decreasing trajectories.	`True`

Raises:

Type	Description
`TypeError`	If `process` is not an instance of `StochasticProcess`, or if `increasing` is not a boolean value.

Returns:

Name	Type	Description
`is_monotonic`	`bool`	`True` if all trajectories are monotonic in the specified direction, `False` otherwise.

Examples:

>>> from scipy.stats import bernoulli
>>> from sigalg.core import Time
>>> from sigalg.processes import IIDProcess
>>> T = Time.discrete(start=1, length=2)
>>> X = IIDProcess(distribution=bernoulli(p=0.6), support=[0, 1], time=T).from_enumeration()
>>> print(X)
Stochastic process 'X':
time        1  2  3
trajectory
0           0  0  0
1           0  0  1
2           0  1  0
3           0  1  1
4           1  0  0
5           1  0  1
6           1  1  0
7           1  1  1
>>> print(X.cumsum())
Stochastic process 'X_cumsum':
time        1  2  3
trajectory
0           0  0  0
1           0  0  1
2           0  1  1
3           0  1  2
4           1  1  1
5           1  1  2
6           1  2  2
7           1  2  3
>>> print(X.cumsum().is_monotonic())
True

ito_integral `classmethod`

ito_integral(integrand, integrator, name=None)

Compute the Itô integral of a stochastic process with respect to another stochastic process.

Parameters:

Name	Type	Description	Default
`integrand`	`StochasticProcess`	The stochastic process to be integrated.	required
`integrator`	`StochasticProcess`	The stochastic process with respect to which the integral is computed.	required
`name`	`Hashable \| None`	The name of the transformed process. If `None`, the new name will be `int X dW`, where `X` is the name of the integrand and `W` is the name of the integrator.	`None`

Returns:

Name	Type	Description
`ito_integral_process`	`StochasticProcess`	A new stochastic process representing the Itô integral of the input process with respect to the integrator.

Examples:

>>> from sigalg.core import Time
>>> from sigalg.processes import RandomWalk, StochasticProcess
>>> T = Time().discrete(length=2)
>>> X = RandomWalk(p=0.6, time=T, initial_state=2).from_enumeration()
>>> print(X)
Stochastic process 'X':
time        0  1  2
trajectory
0           2  1  0
1           2  1  2
2           2  3  2
3           2  3  4
>>> one = StochasticProcess(domain=X.domain, time=T, name=1).from_constant(1)
>>> print(one)
Stochastic process '1':
time        0  1  2
trajectory
0           1  1  1
1           1  1  1
2           1  1  1
3           1  1  1
>>> # The Itô integral of a process, plus its initial state, is equal to the final random variable in the process
>>> print(one.ito_integral(integrator=X) + 2)
Random variable '(int 1 dX+2)':
            (int 1 dX+2)
trajectory
0                      0
1                      2
2                      2
3                      4

max_value `staticmethod`

max_value(process)

Get the maximum value across all trajectories and time points of a stochastic process.

Parameters:

Name	Type	Description	Default
`process`	`StochasticProcess`	The stochastic process for which to find the maximum value.	required

Raises:

Type	Description
`TypeError`	If `process` is not an instance of `StochasticProcess`.

Returns:

Name	Type	Description
`max_value`	`Real`	The maximum value found in the stochastic process.

Examples:

>>> from sigalg.core import Time
>>> from sigalg.processes import RandomWalk
>>> T = Time.discrete(length=2)
>>> X = RandomWalk(p=0.5, time=T, initial_state=3).from_enumeration()
>>> print(X)
Stochastic process 'X':
time        0  1  2
trajectory
0           3  2  1
1           3  2  3
2           3  4  3
3           3  4  5
>>> print(X.max_value())
5

min_value `staticmethod`

min_value(process)

Get the minimum value across all trajectories and time points of a stochastic process.

Parameters:

Name	Type	Description	Default
`process`	`StochasticProcess`	The stochastic process for which to find the minimum value.	required

Raises:

Type	Description
`TypeError`	If `process` is not an instance of `StochasticProcess`.

Returns:

Name	Type	Description
`min_value`	`Real`	The minimum value found in the stochastic process.

Examples:

>>> from sigalg.core import Time
>>> from sigalg.processes import RandomWalk
>>> T = Time.discrete(length=2)
>>> X = RandomWalk(p=0.5, time=T, initial_state=3).from_enumeration()
>>> print(X)
Stochastic process 'X':
time        0  1  2
trajectory
0           3  2  1
1           3  2  3
2           3  4  3
3           3  4  5
>>> print(X.min_value())
1

pointwise_map `staticmethod`

pointwise_map(process, function, name=None)

Apply a function pointwise to the values of a stochastic process.

Parameters:

Name	Type	Description	Default
`process`	`StochasticProcess`	The stochastic process to which the function will be applied.	required
`function`	`Callable[[Hashable], Hashable]`	A function that takes a single value and returns a transformed value. This function will be applied to each value in the stochastic process.	required
`name`	`Hashable \| None`	The name of the transformed process. If `None`, the new name will be the name of the input process subscripted with `mapped`, provided that the name of the input process is a string.	`None`

Raises:

Type	Description
`TypeError`	If `process` is not an instance of `StochasticProcess`, or if `function` is not callable.
`ValueError`	If `process` does not have data to apply the function to.

Returns:

Name	Type	Description
`mapped_process`	`StochasticProcess`	A new stochastic process with the function applied pointwise to its values.

Examples:

>>> from sigalg.core import Time
>>> from sigalg.processes import RandomWalk
>>> T = Time.discrete(length=2)
>>> X = RandomWalk(p=0.5, time=T, initial_state=3).from_enumeration()
>>> print(X)
Stochastic process 'X':
time        0  1  2
trajectory
0           3  2  1
1           3  2  3
2           3  4  3
3           3  4  5
>>> def f(x):
...     return x + 1
>>> print(X.pointwise_map(function=f))
Stochastic process 'X_mapped':
time        0  1  2
trajectory
0           4  3  2
1           4  3  4
2           4  5  4
3           4  5  6

remove_rv `staticmethod`

remove_rv(process, time=None, pos=None, name=None)

Remove a random variable from a stochastic process at a specified time.

Parameters:

Name	Type	Description	Default
`process`	`StochasticProcess`	The stochastic process from which to remove the random variable.	required
`time`	`Real \| None`	The time point at which to remove the random variable. If `None`, `pos` must be specified.	`None`
`pos`	`int \| None`	The position at which to remove the random variable. If `None`, `time` must be specified.	`None`
`name`	`Hashable \| None`	The name of the transformed process. If `None`, the new name will be the name of the input process subscripted with `remove`, provided that the name of the input process is a string.	`None`

Raises:

Type	Description
`TypeError`	If `process` is not an instance of `StochasticProcess`, if `time` is not a real number, or if `pos` is not an integer.
`ValueError`	If `process` has no data, if `time` is not in the process time index, if both `time` and `pos` are specified, or if neither is specified.

Returns:

Name	Type	Description
`removed_process`	`StochasticProcess`	A new stochastic process with the random variable removed at the specified time.

Examples:

>>> from sigalg.core import Time
>>> from sigalg.processes import RandomWalk
>>> T = Time.discrete(start=1, length=2)
>>> X = RandomWalk(p=0.6, time=T).from_enumeration()
>>> print(X)
Stochastic process 'X':
time        1  2  3
trajectory
0           0 -1 -2
1           0 -1  0
2           0  1  0
3           0  1  2
>>> print(X.remove_rv(time=2))
Stochastic process 'remove(X)':
time        1  3
trajectory
0           0 -2
1           0  0
2           0  0
3           0  2
>>> S = Time.continuous(start=0, stop=0.3, dt=0.101)
>>> Y = RandomWalk(p=0.6, time=S, name="Y").from_enumeration()
>>> print(Y)
Stochastic process 'Y':
time        0.0  0.1  0.2  0.3
trajectory
0             0   -1   -2   -3
1             0   -1   -2   -1
2             0   -1    0   -1
3             0   -1    0    1
4             0    1    0   -1
5             0    1    0    1
6             0    1    2    1
7             0    1    2    3
>>> print(Y.remove_rv(pos=2))
Stochastic process 'remove(Y)':
time        0.0  0.1  0.3
trajectory
0             0   -1   -3
1             0   -1   -1
2             0   -1   -1
3             0   -1    1
4             0    1   -1
5             0    1    1
6             0    1    1
7             0    1    3

sum `staticmethod`

sum(process, name=None)

Compute the sum of a stochastic process across its time index.

Parameters:

Name	Type	Description	Default
`process`	`StochasticProcess`	The stochastic process for which to compute the sum.	required
`name`	`Hashable \| None`	The name of the transformed random variable. If `None`, the new name will be the name of the input process subscripted with `sum`, provided that the name of the input process is a string.	`None`

Raises:

Type	Description
`TypeError`	If `process` is not an instance of `StochasticProcess`.

Returns:

Name	Type	Description
`sum_variable`	`RandomVariable`	A new random variable representing the sum of the input process across its time index.

Examples:

>>> from sigalg.core import Time
>>> from sigalg.processes import RandomWalk
>>> T = Time.discrete(length=2)
>>> X = RandomWalk(p=0.5, time=T, initial_state=3).from_enumeration()
>>> print(X)
Stochastic process 'X':
time        0  1  2
trajectory
0           3  2  1
1           3  2  3
2           3  4  3
3           3  4  5
>>> print(X.sum())
Random variable 'X_sum':
            X_sum
trajectory
0               6
1               8
2              10
3              12

to_counting_process `classmethod`

to_counting_process(process, time, name=None)

Convert a stochastic process of "arrival times" to a counting process.

The trajectories in the given process are assumed to be the occurrence times of some event, while its time index represents the cumulative counts of those events. This method creates a new stochastic process where, at each time point in the provided time index, the value represents the total count of events that have occurred up to that time.

Parameters:

Name	Type	Description	Default
`process`	`StochasticProcess`	The original stochastic process to be converted. The process trajectories must be monotonically increasing.	required
`time`	`Time`	The time index for the counting process.	required
`name`	`Hashable \| None`	The name of the transformed process. If `None`, the new name will be the name of the input process subscripted with `counting`, provided that the name of the input process is a string.	`None`

Raises:

Type	Description
`TypeError`	If `process` is not an instance of `StochasticProcess`.
`ValueError`	If the trajectories in `process` are not monotonically increasing.

Returns:

Name	Type	Description
`counting_process`	`StochasticProcess`	A new stochastic process representing the counting process.

Examples:

>>> from scipy.stats import expon
>>> from sigalg.core import Index, Time
>>> from sigalg.processes import IIDProcess
>>> # Parameters for a Poisson process
>>> rate = 2.0
>>> n_trajectories = 5
>>> random_state = 42
>>> max_count = 5
>>> # Create an index for the counts
>>> counts = Time.discrete(start=1, stop=max_count, data_name="count", name=None)
>>> # Exponential interarrival times with given rate
>>> interarrival_times = IIDProcess(
...     distribution=expon(scale=1 / rate),
...     name="interarrival_times",
...     time=counts,
... ).from_simulation(n_trajectories=n_trajectories, random_state=random_state)
>>> interarrival_times
Stochastic process 'interarrival_times':
count         1         2         3         4         5
trajectory
0      1.202104  1.168095  1.192380  0.139897  0.043219
1      0.726330  0.704980  1.562148  0.039647  0.523280
2      0.035218  0.544512  0.865664  0.193447  0.615793
3      0.076887  0.045789  0.157590  0.450600  0.206493
4      0.623693  0.111788  0.918985  0.613543  0.327898
>>> # Compute arrival times by cumulative sum of interarrival times
>>> arrival_times = interarrival_times.cumsum().with_name("arrival_times")
>>> arrival_times
Stochastic process 'arrival_times':
count         1         2         3         4         5
trajectory
0      1.202104  2.370199  3.562580  3.702477  3.745695
1      0.726330  1.431311  2.993459  3.033106  3.556386
2      0.035218  0.579730  1.445394  1.638841  2.254634
3      0.076887  0.122675  0.280265  0.730864  0.937357
4      0.623693  0.735481  1.654466  2.268009  2.595907
>>> # Determine time grid for Poisson process
>>> longest_trajectory = arrival_times.max_value()
>>> time = Time.continuous(
...     start=0.0,
...     stop=longest_trajectory + 0.1,
...     num_points=6,
... )
>>> # Convert to Poisson counting process
>>> poisson = arrival_times.to_counting_process(
...     time=time,
... ).with_name("poisson")
>>> poisson
Stochastic process 'poisson':
time        0.000000  0.769139  1.538278  2.307417  3.076556  3.845695
trajectory
0                0.0       0.0       1.0       1.0       2.0       5.0
1                0.0       1.0       2.0       2.0       4.0       5.0
2                0.0       2.0       3.0       5.0       5.0       5.0
3                0.0       4.0       5.0       5.0       5.0       5.0
4                0.0       2.0       2.0       4.0       5.0       5.0

transform `staticmethod`

transform(process, functions, time=None, name=None)

Apply a transformation to a stochastic process.

Parameters:

Name	Type	Description	Default
`process`	`StochasticProcess`	The stochastic process to transform.	required
`functions`	`list[Callable[[StochasticProcess], RandomVariable]]`	A list of functions to apply to the stochastic process.	required
`time`	`Time \| None`	The new time index for the transformed process. If `None`, the original time index of `process` will be used.	`None`
`name`	`Hashable \| None`	The name of the transformed process. If `None`, the new name will be `function(process.name)` if `process.name` is not `None`.	`None`

Raises:

Type	Description
`TypeError`	If `process` is not an instance of `StochasticProcess`, or `functions` is not a list of callables, or `time` is not an instance of `Time`.
`ValueError`	If the length of `functions` does not match the length of `time`.

Returns:

Name	Type	Description
`transformed_process`	`StochasticProcess`	The transformed stochastic process.

Examples:

>>> from scipy.stats import bernoulli
>>> from sigalg.core import RandomVariable, Time
>>> from sigalg.processes import IIDProcess, StochasticProcess
>>> T = Time().discrete(start=0, length=2)
>>> X = IIDProcess(distribution=bernoulli(p=0.5), support=[0, 1], time=T).from_enumeration()
>>> print(X)
Stochastic process 'X':
time        0  1  2
trajectory
0           0  0  0
1           0  0  1
2           0  1  0
3           0  1  1
4           1  0  0
5           1  0  1
6           1  1  0
7           1  1  1
>>> S = Time().discrete(start=4, stop=5)
>>> def f4(process: StochasticProcess) -> RandomVariable:
...     X0, X1, _ = X
...     return X0 + X1
>>> def f5(process: StochasticProcess) -> RandomVariable:
...     _, X1, X2 = X
...     return X1 + X2
>>> print(X.transform(functions=[f4, f5], time=S))
Stochastic process 'function(X)':
time        4  5
trajectory
0           0  0
1           0  1
2           1  1
3           1  2
4           1  0
5           1  1
6           2  1
7           2  2

RandomWalk

Bases: StochasticProcess

A class representing a random walk stochastic process.

Parameters:

Name	Type	Description	Default
`p`	`Real`	The probability that the particle takes a step to the right, so `1-p` is the probability that it steps left. Must be between `0` and `1`.	required
`initial_state`	`int`	The initial state of the random walk at the first time point. Must be an integer.	`0`
`time`	`Time \| None`	The time index of the stochastic process. If `None`, then the `is_discrete_time` property must be provided.	`None`
`is_discrete_time`	`bool \| None`	Whether the stochastic process is a discrete-time process. If `None`, then `time` parameter must be provided.	`None`
`domain`	`SampleSpace \| None`	The sample space representing the domain of the stochastic process. If `None`, it will be generated later through data generation methods.	`None`
`name`	`Hashable \| None`	The name of the stochastic process.	`"X"`

Raises:

Type	Description
`TypeError`	If `p` is not a real number between `0` and `1`.

Examples:

>>> from math import comb
>>> from sigalg.processes import RandomWalk
>>> # Define a random walk with probability p=0.75 of stepping right one unit, and 0.25 of stepping left one unit
>>> X = RandomWalk(p=0.75, name="X", is_discrete_time=True).from_enumeration(length=3)
>>> # Print the trajectories and their probabilities
>>> X.range.print_trajectories_and_probabilities()
            0  1  2  3  probability
trajectory
0           0 -1 -2 -3     0.015625
1           0 -1 -2 -1     0.046875
2           0 -1  0 -1     0.046875
3           0 -1  0  1     0.140625
4           0  1  0 -1     0.046875
5           0  1  0  1     0.140625
6           0  1  2  1     0.140625
7           0  1  2  3     0.421875
>>> # Print the values of the X_3 random variable and their corresponding probabilities
>>> X.at[3].range.print_values_and_probabilities()
        X_3  probability
output
x_3_0    -3     0.015625
x_3_1    -1     0.140625
x_3_2     1     0.421875
x_3_3     3     0.421875
>>> # Print binomial probabilities and note they match the law of X_3
>>> for k in range(4):
...     print(comb(3, k) * (0.75**k) * (0.25**(3-k)))
0.015625
0.140625
0.421875
0.421875

StochasticProcess

Bases: RandomVector, ProcessTransformMethods

A class representing a stochastic process.

Parameters:

Name	Type	Description	Default
`time`	`Time \| None`	The time index of the stochastic process. If `None`, then the `is_discrete_time` property must be provided.	`None`
`is_discrete_time`	`bool \| None`	Whether the stochastic process is a discrete-time process. If `None`, then `time` parameter must be provided.	`None`
`domain`	`SampleSpace \| None`	The sample space representing the domain of the stochastic process. If `None`, it will be generated later through data generation methods.	`None`
`is_discrete_state`	`bool \| None`	Whether the stochastic process is a discrete-state process. If `None`, then subclasses should set this property based on the specific type of stochastic process.	`None`
`name`	`Hashable \| None`	The name of the stochastic process.	`"X"`
`**kwargs`		Additional keyword arguments for subclasses.	`{}`

Examples:

>>> from sigalg.core import SampleSpace, Time
>>> from sigalg.processes import StochasticProcess
>>> domain = SampleSpace().from_sequence(size=3, prefix="omega")
>>> time = Time.discrete(length=2)
>>> X = StochasticProcess(domain=domain, time=time).from_dict(
...     {
...         "omega_0": (1, 2, 3),
...         "omega_1": (4, 5, 6),
...         "omega_2": (7, 8, 9),
...     }
... )
>>> X
Stochastic process 'X':
time      0  1  2
sample
omega_0   1  2  3
omega_1   4  5  6
omega_2   7  8  9

at `property`

at

Get an indexer for accessing component random variables at specific times.

Returns:

Name	Type	Description
`at`	`_RVAtIndexer`	An indexer for accessing component random variables at specific times.

is_enumerated `property`

is_enumerated

Check if the stochastic process is enumerated.

Raises:

Type	Description
`ValueError`	If the `is_enumerated` property is not set.

Returns:

Name	Type	Description
`is_enumerated`	`bool`	`True` if the stochastic process is enumerated, `False` otherwise.

last_rv `property`

last_rv

Get the random variable corresponding to the last time point.

Raises:

Type	Description
`ValueError`	If data has not been generated for the stochastic process.

Returns:

Name	Type	Description
`last_rv`	`RandomVariable`	The random variable corresponding to the last time point.

n_trajectories `property`

n_trajectories

Get the number of trajectories in the stochastic process.

Returns:

Name	Type	Description
`n_trajectories`	`int \| None`	The number of trajectories in the stochastic process. `None` if data has not been generated.

natural_filtration `property`

natural_filtration

Get the natural filtration of the stochastic process.

Raises:

Type	Description
`ValueError`	If `name_prefix` is not a string.

Returns:

Name	Type	Description
`natural_filtration`	`Filtration \| None`	The natural filtration of the stochastic process, or `None` if data has not been generated for the stochastic process.

probability_measure `property` `writable`

probability_measure

Generate a probability measure on the domain of the stochastic process.

Raises a ValueError if data has not been generated for the stochastic process. Data generation must be implemented in subclasses.

Raises:

Type	Description
`ValueError`	If data has not been generated for the stochastic process.

Returns:

Name	Type	Description
`prob_measure`	`ProbabilityMeasure`	The generated probability measure.

random_variables `property`

random_variables

Get the dictionary of random variables corresponding to each time point.

Raises:

Type	Description
`ValueError`	If data has not been generated for the stochastic process.

Returns:

Name	Type	Description
`random_variables`	`dict[RandomVariable]`	The dictionary of random variables corresponding to each time point.

range `property`

range

Get the range of the stochastic process.

Overrides the range property of the superclass RandomVector to return a StochasticProcess instance representing the range of the process, with its own domain and probability measure derived from the trajectories of the original process.

time `property` `writable`

time

Get the time index.

This attribute is an alias for public attribute index of the superclass RandomVector.

Returns:

Name	Type	Description
`time`	`Time \| None`	The time index of the stochastic process.

from_constant

from_constant(value, length=None)

Create a stochastic process with all trajectories equal to a constant value.

Parameters:

Name	Type	Description	Default
`value`	`Real`	The constant value for all trajectories.	required
`length`	`int \| None`	The length of each trajectory. If `None`, the length of the existing time index is used.	`None`

Raises:

Type	Description
`ValueError`	If `length` is not a positive integer or if the domain is not initialized.
`TypeError`	If `value` is not a real number.

Returns:

Name	Type	Description
`self`	`StochasticProcess`	The stochastic process with constant trajectories.

Examples:

>>> from sigalg.core import SampleSpace, Time
>>> from sigalg.processes import StochasticProcess
>>> Omega = SampleSpace().from_sequence(size=2)
>>> T = Time().discrete(length=3)
>>> X = StochasticProcess(domain=Omega, time=T).from_constant(2)
>>> print(X)
Stochastic process 'X':
time    0  1  2  3
sample
0       2  2  2  2
1       2  2  2  2

from_enumeration

from_enumeration(length=None, **kwargs)

Generate data by exhaustively enumerating all possible trajectories.

For this method to be used, a subclass must implement the _enumeration_logic method, which defines how to enumerate trajectories for the specific type of stochastic process.

Parameters:

Name	Type	Description	Default
`length`	`int \| None`	The length of each trajectory. If `None`, the length of the existing index is used.	`None`
`**kwargs`		Additional keyword arguments for subclasses, which may include parameters needed for the enumeration logic.	`{}`

Returns:

Name	Type	Description
`self`	`StochasticProcess`	The stochastic process with enumerated trajectories.

from_simulation

from_simulation(
    n_trajectories, length=None, random_state=None
)

Generate data by simulating trajectories.

For this method to be used, a subclass must implement the _simulation_logic method, which defines how to simulate trajectories for the specific type of stochastic process.

Parameters:

Name	Type	Description	Default
`n_trajectories`	`int`	The number of trajectories to simulate.	required
`length`	`int \| None`	The length of each trajectory. If `None`, the length of the existing time index is used.	`None`
`random_state`	`int \| None`	An optional random seed for reproducibility.	`None`

Raises:

Type	Description
`ValueError`	If `n_trajectories` is not a positive integer.

Returns:

Name	Type	Description
`self`	`StochasticProcess`	The stochastic process with simulated trajectories.

is_adapted

is_adapted(filtration)

Check if the stochastic process is adapted to a given filtration.

Parameters:

Name	Type	Description	Default
`filtration`	`Filtration`	The filtration to check adaptation against.	required

Raises:

Type	Description
`ValueError`	If data has not been generated for the stochastic process.
`TypeError`	If the provided filtration is not an instance of Filtration, or its sample space does not match the domain of the process, or its time index does not match the time index of the process.

Returns:

Name	Type	Description
`is_adapted`	`bool`	True if the stochastic process is adapted to the given filtration, False otherwise.

Examples:

>>> from sigalg.core import RandomVariable, Time
>>> from sigalg.processes import RandomWalk, StochasticProcess
>>> T = Time.discrete(start=0, stop=2)
>>> X = RandomWalk(p=0.7, time=T).from_enumeration()
>>> print(X)
Stochastic process 'X':
time        0  1  2
trajectory
0           0 -1 -2
1           0 -1  0
2           0  1  0
3           0  1  2
>>> def f0(X: StochasticProcess) -> RandomVariable:
...     return X[0]
>>> def f1(X: StochasticProcess) -> RandomVariable:
...     return 2 * X[0] + X[1]
>>> def f2(X: StochasticProcess) -> RandomVariable:
...     return X[2] - X[1] + X[0]
>>> Y = X.transform(functions=[f0, f1, f2], name="Y")
>>> print(Y)
Stochastic process 'Y':
time        0  1  2
trajectory
0           0 -1 -1
1           0 -1  1
2           0  1 -1
3           0  1  1
>>> print(Y.is_adapted(filtration=X.natural_filtration))
True

is_martingale

is_martingale(filtration=None, rtol=1e-05, atol=1e-08)

Check if the stochastic process is a martingale (with respect to an optional filtration).

Take care when using this method for non-enumerated processes, as the check may be inaccurate due to probabilities being approximated. Adjusting the rtol and atol parameters may help in such cases. Also, even for enumerated processes, this method is very computationally intensive, as it requires calculating conditional expectations at each time step.

Parameters:

Name	Type	Description	Default
`filtration`	`Filtration \| None`	The filtration with respect to which the martingale property is checked. If None, the natural filtration of the process is used.	`None`
`rtol`	`float`	The relative tolerance parameter for numerical comparison. Internally passed to `numpy.allclose`.	`1e-05`
`atol`	`float`	The absolute tolerance parameter for numerical comparison. Internally passed to `numpy.allclose`.	`1e-08`

Raises:

Type	Description
`ValueError`	If data has not been generated for the stochastic process, or if the process is not discrete-state.
`TypeError`	If the provided filtration is not an instance of Filtration, or its sample space does not match the domain of the process, or its time index does not match the time index of the process.

Returns:

Name	Type	Description
`is_martingale`	`bool`	`True` if the stochastic process is a martingale, `False` otherwise.

Examples:

>>> from sigalg.core import Time
>>> from sigalg.processes import RandomWalk
>>> T = Time.discrete(start=1, length=2)
>>> # Symmetric random walks are martingales
>>> X = RandomWalk(p=0.5, time=T).from_enumeration()
>>> print(X.is_martingale())
True
>>> # Non-symmetric random walks are not martingales
>>> Y = RandomWalk(p=0.7, time=T).from_enumeration()
>>> print(Y.is_martingale())
False

is_predictable

is_predictable(filtration)

Check if the stochastic process is predictable with respect to a given filtration.

The time index of self must match all but the first time indices of the filtration.

Parameters:

Name	Type	Description	Default
`filtration`	`Filtration`	The filtration to check predictability against.	required

Raises:

Type	Description
`ValueError`	If data has not been generated for the stochastic process.
`TypeError`	If the provided filtration is not an instance of `Filtration`, or its sample space does not match the domain of the process, or if the time indices of the process do not match all but the first time indices of the filtration.

Returns:

Name	Type	Description
`is_predictable`	`bool`	`True` if the stochastic process is predictable with respect to the given filtration, `False` otherwise.

Examples:

>>> from sigalg.core import RandomVariable, Time
>>> from sigalg.processes import RandomWalk, StochasticProcess
>>> T = Time.discrete(start=0, stop=3)
>>> X = RandomWalk(p=0.7, time=T).from_enumeration()
>>> print(X)
Stochastic process 'X':
time        0  1  2  3
trajectory
0           0 -1 -2 -3
1           0 -1 -2 -1
2           0 -1  0 -1
3           0 -1  0  1
4           0  1  0 -1
5           0  1  0  1
6           0  1  2  1
7           0  1  2  3
>>> def f1(X: StochasticProcess) -> RandomVariable:
...     return 2 * X[0]
>>> def f2(X: StochasticProcess) -> RandomVariable:
...     return X[1] + X[0]
>>> def f3(X: StochasticProcess) -> RandomVariable:
...     return X[2] - 5 * X[1]
>>> S = Time.discrete(start=1, stop=3)
>>> Y = X.transform(functions=[f1, f2, f3], time=S, name="Y")
>>> print(Y)
Stochastic process 'Y':
time        1  2  3
trajectory
0           0 -1  3
1           0 -1  3
2           0 -1  5
3           0 -1  5
4           0  1 -5
5           0  1 -5
6           0  1 -3
7           0  1 -3
>>> print(Y.is_predictable(filtration=X.natural_filtration))
True

is_submartingale

is_submartingale(filtration=None, rtol=1e-05, atol=1e-08)

Check if the stochastic process is a submartingale (with respect to an optional filtration).

Parameters:

Name	Type	Description	Default
`filtration`	`Filtration \| None`	The filtration with respect to which the submartingale property is checked. If None, the natural filtration of the process is used.	`None`
`rtol`	`float`	The relative tolerance parameter for numerical comparison. Internally passed to `numpy.allclose`.	`1e-05`
`atol`	`float`	The absolute tolerance parameter for numerical comparison. Internally passed to `numpy.allclose`.	`1e-08`

Raises:

Type	Description
`ValueError`	If data has not been generated for the stochastic process, or if the process is not discrete-state.
`TypeError`	If the provided filtration is not an instance of Filtration, or its sample space does not match the domain of the process, or its time index does not match the time index of the process.

Returns:

Name	Type	Description
`is_submartingale`	`bool`	`True` if the stochastic process is a submartingale, `False` otherwise.

Examples:

>>> from sigalg.core import Time
>>> from sigalg.processes import RandomWalk
>>> T = Time.discrete(start=1, length=2)
>>> # A random walk with upward drift is a submartingale
>>> X = RandomWalk(p=0.6, time=T).from_enumeration()
>>> print(X.is_submartingale())
True

is_supermartingale

is_supermartingale(filtration=None, rtol=1e-05, atol=1e-08)

Check if the stochastic process is a supermartingale (with respect to an optional filtration).

Parameters:

Name	Type	Description	Default
`filtration`	`Filtration \| None`	The filtration with respect to which the supermartingale property is checked. If None, the natural filtration of the process is used.	`None`
`rtol`	`float`	The relative tolerance parameter for numerical comparison. Internally passed to `numpy.allclose`.	`1e-05`
`atol`	`float`	The absolute tolerance parameter for numerical comparison. Internally passed to `numpy.allclose`.	`1e-08`

Raises:

Type	Description
`ValueError`	If data has not been generated for the stochastic process, or if the process is not discrete-state.
`TypeError`	If the provided filtration is not an instance of Filtration, or its sample space does not match the domain of the process, or its time index does not match the time index of the process.

Returns:

Name	Type	Description
`is_supermartingale`	`bool`	True if the stochastic process is a supermartingale, False otherwise.

Examples:

>>> from sigalg.core import Time
>>> from sigalg.processes import RandomWalk
>>> T = Time.discrete(start=1, length=2)
>>> # A random walk with downward drift is a supermartingale
>>> X = RandomWalk(p=0.4, time=T).from_enumeration()
>>> print(X.is_supermartingale())
True

plot_trajectories

plot_trajectories(
    ax=None,
    colors=None,
    plot_kwargs=None,
    x_label="time",
    y_label="state",
    title=None,
)

Plot the trajectories of the stochastic process.

Requires the data to be generated for the stochastic process. Only subclasses that implement data generation methods can use this method.

Parameters:

Name	Type	Description	Default
`ax`	`Axes`	A matplotlib Axes object to plot on. If `None`, a new figure and axes will be created.	`None`
`colors`	`list`	A list of colors to use for the trajectories. If `None`, default matplotlib colors will be used.	`None`
`plot_kwargs`	`dict`	Additional keyword arguments to pass to the plotting function.	`None`
`x_label`	`str`	Label for the x-axis.	`"time"`
`y_label`	`str`	Label for the y-axis.	`"state"`
`title`	`str`	Title of the plot. If `None`, a default title will be generated.	`None`

Raises:

Type	Description
`ValueError`	If data has not been generated for the stochastic process.
`TypeError`	If ax is not a matplotlib Axes object.

Returns:

Name	Type	Description
`ax`	`Axes`	The matplotlib Axes object with the plot.

print_trajectories_and_probabilities

print_trajectories_and_probabilities()

Print the trajectories and their corresponding probabilities.

API Reference

processes

BrownianMotion

IIDProcess

MarkovChain

PoissonProcess

ProcessTransforms

cumprod staticmethod

cumsum staticmethod

increments staticmethod

insert_rv staticmethod

is_monotonic staticmethod

ito_integral classmethod

max_value staticmethod

min_value staticmethod

pointwise_map staticmethod

remove_rv staticmethod

sum staticmethod

to_counting_process classmethod

transform staticmethod

RandomWalk

StochasticProcess

at property

is_enumerated property

last_rv property

n_trajectories property

natural_filtration property

probability_measure property writable

random_variables property

range property

time property writable

from_constant

from_enumeration

from_simulation

is_adapted

is_martingale

is_predictable

is_submartingale

is_supermartingale

plot_trajectories

print_trajectories_and_probabilities

cumprod `staticmethod`

cumsum `staticmethod`

increments `staticmethod`

insert_rv `staticmethod`

is_monotonic `staticmethod`

ito_integral `classmethod`

max_value `staticmethod`

min_value `staticmethod`

pointwise_map `staticmethod`

remove_rv `staticmethod`

sum `staticmethod`

to_counting_process `classmethod`

transform `staticmethod`

at `property`

is_enumerated `property`

last_rv `property`

n_trajectories `property`

natural_filtration `property`

probability_measure `property` `writable`

random_variables `property`

range `property`

time `property` `writable`