API Reference

processes

Module containing components for stochastic processes, including base classes, transforms, and specific types of processes.

BranchingProcess

Bases: StochasticProcess

A class representing a branching process.

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
`support`	`list \| None`	Add description later.	`None`
`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

discount `staticmethod`

discount(process, rate, name=None)

Return the discounted process of a given stochastic process.

The discounted process is given by

\[ \tilde{S}_t = \frac{S_t}{(1+r)^t}, \]

where \(S_t\) is the original process and \(r\) is the discount rate.

Parameters:

Name	Type	Description	Default
`process`	`StochasticProcess`	The original process to be discounted.	required
`rate`	`Real`	The discount rate, which must be a positive real number.	required
`name`	`Hashable \| None`	The name of the discounted process. If `None`, the new name will be the name of the input process subscripted with `discount`, 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 `rate` is not a positive real number, or if `name` is not a hashable object or `None`.

Returns:

Name	Type	Description
`discounted_process`	`StochasticProcess`	The discounted process.

increments `staticmethod`

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

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

Given a stochastic process \(X_t\) with index set \(\{t_0,t_0+1,\ldots,T\}\) there are two types of increments that can be computed: The first are forward increments, which results in a stochastic process \(\Delta X_t\) defined as

\[ \Delta X_t = X_{t+1} - X_t, \]

for each \(t=t_0,\ldots,T-1\). The second type are backward increments, which results in a stochastic process \(\Delta X_t\) where

\[ \Delta X_t = X_t - X_{t-1}, \]

for each \(t=t_0+1,\ldots,T\).

Parameters:

Name	Type	Description	Default
`process`	`StochasticProcess`	The stochastic process for which to compute the increments.	required
`forward`	`bool`	If `True`, compute forward increments; otherwise, compute backward increments.	`True`
`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(forward=True))
Stochastic process 'X_increments':
time        0  1
trajectory
0          -1 -1
1          -1  1
2           1 -1
3           1  1
>>> print(X.increments(forward=False))
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,
    time,
    rv=None,
    state=None,
    name=None,
    in_place=False,
)

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
`time`	`Real`	The time at which to insert the random variable.	required
`rv`	`RandomVariable \| None`	The random variable to insert. One or the other of `rv` or `state` must be provided, but not both.	`None`
`state`	`Hashable \| None`	A constant state to assign to the inserted random variable for all trajectories. One or the other of `rv` or `state` must be provided, but not both.	`None`
`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`
`in_place`	`bool`	If `True`, modify the input process in place and return it. If `False`, return a new stochastic process with the random variable inserted, leaving the input process unchanged.	`False`

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 `staticmethod`

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
>>> time = Time().discrete(length=2)
>>> X = RandomWalk(p=0.6, time=time).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
>>> T = StochasticProcess(domain=X.domain, time=time, name="T").from_time()
>>> print(T)
Stochastic process 'T':
time        0  1  2
trajectory
0           0  1  2
1           0  1  2
2           0  1  2
3           0  1  2
>>> integral = X.increments().ito_integral(T)
>>> print(integral)
Stochastic process 'int X_increments dT':
time        0  1  2
trajectory
0           0 -1 -2
1           0 -1  0
2           0  1  0
3           0  1  2

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

mean `staticmethod`

mean(process, name=None)

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

Parameters:

Name	Type	Description	Default
`process`	`StochasticProcess`	The stochastic process for which to compute the mean.	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 `mean`, 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
`mean_variable`	`RandomVariable`	A new random variable representing the mean of the input process across its time index.

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

stopped `staticmethod`

stopped(process, stopping_time, name=None)

Pass.

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.

The types of stochastic processes modeled by SigAlg are the following: A collection of random variables \(X_t\), indexed by a finite set \(T\), usually refered to as time. The time index may be discrete or continuous, and the random variables may be discrete or continuous (i.e., discrete or continuous state).

The trajectories of an instance of StochasticProcess may be loaded directly from either a pd.DataFrame or np.ndarray object via the from_pandas and from_numpy methods, respectively. Other methods for construction include from_constant and from_time, while random trajectories may be generated through the from_randint and from_randnorm methods.

Alternatively, subclasses may implement from_enumeration and from_simulation methods to generate trajectories through enumeration or simulation, respectively.

Subclassing instructions: All subclasses must implement the private _simulation_logic method, which defines how to simulate trajectories for the specific type of stochastic process and returns a pd.DataFrame object containing the simulated trajectories as rows and time points as columns. Subclasses that support exhaustive enumeration of trajectories should also implement the private _enumeration_logic method, which defines how to enumerate trajectories for the specific type of stochastic process. This method must also return a pd.DataFrame object containing the enumerated trajectories as rows and time points as columns. For these latter subclasses, the private _generate_exact_prob_measure method must also be implemented to generate the exact probability measure for the enumerated stochastic process. The developer should consult the source code for the subclass IIDProcess for a simple example of how to implement a subclass of StochasticProcess that supports both enumeration and simulation.

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 and the time index will be generated later through data generation methods.	`None`
`is_discrete_time`	`bool \| None`	Whether the stochastic process is a discrete-time process. If `None`, then `time` parameter must be provided and the `is_discrete_time` attribute will be determined based on the discreteness of the provided time index.	`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 the `is_discrete_state` attribute is meant to be set later through data generation methods.	`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)
>>> time = Time.discrete(length=2)
>>> X = StochasticProcess(domain=domain, time=time).from_dict(
...     {
...         0: (1, 2, 3),
...         1: (4, 5, 6),
...         2: (7, 8, 9),
...     }
... )
>>> X
Stochastic process 'X':
time      0  1  2
sample
0         1  2  3
1         4  5  6
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.

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.

Examples:

>>> from sigalg.core import Time
>>> from sigalg.processes import StochasticProcess
>>> T = Time.discrete(length=3)
>>> X = StochasticProcess(time=T).from_randint(low=0, high=6, n_trajectories=4, random_state=42)
>>> X
Stochastic process 'X':
time        0  1  2  3
trajectory
0           0  4  3  2
1           2  5  0  4
2           1  0  3  5
3           4  4  4  4
>>> X.last_rv
Random variable 'X_3':
            X_3
trajectory
0             2
1             4
2             5
3             4

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.

Given a stochastic process \(X_t\) indexed by \(T\), the natural filtration is defined as the collection of \(\sigma\)-algebras \(\mathcal{F}_t\) where \(\mathcal{F}_t = \sigma(X_s : s \leq t)\) for each \(t \in T\).

Raises:

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

Returns:

Name	Type	Description
`natural_filtration`	`Filtration \| None`	The natural filtration of the stochastic process.

Examples:

>>> from sigalg.core import Time
>>> from sigalg.processes import StochasticProcess
>>> T = Time.discrete(length=3)
>>> X = StochasticProcess(time=T).from_randint(low=0, high=2, n_trajectories=3, random_state=42)
>>> X
Stochastic process 'X':
time        0  1  2  3
trajectory
0           0  1  1  0
1           0  1  0  1
2           0  0  1  1
>>> X.natural_filtration.data
time        0       1          2             3
trajectory
0           0  (0, 1)  (0, 1, 1)  (0, 1, 1, 0)
1           0  (0, 1)  (0, 1, 0)  (0, 1, 0, 1)
2           0  (0, 0)  (0, 0, 1)  (0, 0, 1, 1)

probability_measure `property` `writable`

probability_measure

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

The method will first try to obtain the probability measure through the _generate_exact_prob_measure method if the process is enumerated, or the _generate_empirical_prob_measure method if the process is simulated. If these methods fail, it will check if data has been generated for the stochastic process. If data is available, it will generate a uniform probability measure on the domain of the stochastic process. If not data is available, it will raise a ValueError.

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.

Examples:

>>> from sigalg.core import Time
>>> from sigalg.processes import StochasticProcess
>>> T = Time.discrete(length=1)
>>> X = StochasticProcess(time=T).from_randint(low=0, high=6, n_trajectories=2, random_state=42)
>>> X
Stochastic process 'X':
time        0  1
trajectory
0           0  4
1           3  2
>>> for rv in X.random_variables.values():
...     print(rv)
Random variable '0':
            0
trajectory
0           0
1           3
Random variable '1':
            1
trajectory
0           4
1           2

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.

Raises:

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

Returns:

Name	Type	Description
`range`	`StochasticProcess`	A `StochasticProcess` instance representing the range of the original process, with its own domain and probability measure derived from the trajectories of the original process.

Examples:

>>> from sigalg.core import Time
>>> from sigalg.processes import StochasticProcess
>>> T = Time.discrete(length=1)
>>> X = StochasticProcess(time=T).from_randint(low=0, high=2, n_trajectories=16, random_state=42)
>>> X
Stochastic process 'X':
time        0  1
trajectory
0           0  1
1           1  0
2           0  1
3           0  1
4           0  0
5           1  1
6           1  1
7           1  1
8           1  0
9           1  0
10          1  0
11          0  1
12          1  1
13          0  1
14          1  0
15          0  0
>>> X.range
Stochastic process 'X_range':
time        0  1
trajectory
0           0  0
1           0  1
2           1  0
3           1  1
>>> X.range.probability_measure
Probability measure 'P_X':
            probability
trajectory
0                0.1250
1                0.3125
2                0.3125
3                0.2500

time `property` `writable`

time

Get the time index.

This attribute is an alias for the 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.	`{}`

Raises:

Type	Description
`ValueError`	If `length` is given and is not a positive integer.

Returns:

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

Examples:

>>> from sigalg.core import Time
>>> from sigalg.processes import RandomWalk
>>> T = Time.discrete(length=3)
>>> X = RandomWalk(p=0.7, time=T).from_enumeration()
>>> 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

from_numpy

from_numpy(array, is_enumerated=None)

Generate a stochastic process from a np.ndarray object.

If a Time index is provided at construction, its length must match the number of columns in the array. If no Time index is provided, a new Time index will be generated based on the number of columns in the array, but the is_discrete_time property must be set at construction.

Parameters:

Name	Type	Description	Default
`array`	`ndarray`	A 2D array containing the data for the stochastic process, where each row corresponds to a trajectory and each column corresponds to a time point.	required
`is_enumerated`	`bool \| None`	Whether the stochastic process is enumerated. If `None`, the `_is_enumerated` attribute will not be set. This parameter is primarily intended for internal use during data generation methods.	`None`

Raises:

Type	Description
`ValueError`	If the length of the provided time index does not match the number of columns in the array.
`TypeError`	If `is_enumerated` is not a boolean or `None`.

Returns:

Name	Type	Description
`self`	`StochasticProcess`	A stochastic process with data from the provided array.

from_pandas

from_pandas(data, is_enumerated=None)

Generate a stochastic process from a pd.DataFrame object.

If a Time index is provided at construction, it will overwrite the column names of the data frame. If no Time index is provided, a new Time index will be generated based on the column names of the data frame, but the is_discrete_time property must be set at construction.

Parameters:

Name	Type	Description	Default
`data`	`DataFrame`	A data frame containing the data for the stochastic process, where each row corresponds to a trajectory and each column corresponds to a time point.	required
`is_enumerated`	`bool \| None`	Whether the stochastic process is enumerated. If `None`, the `_is_enumerated` attribute will not be set. This parameter is primarily intended for internal use during data generation methods.	`None`

Raises:

Type	Description
`ValueError`	If the length of the provided time index does not match the number of columns in the data frame.
`TypeError`	If `is_enumerated` is not a boolean or `None`.

Returns:

Name	Type	Description
`self`	`StochasticProcess`	A stochastic process with data from the provided data frame.

from_randint

from_randint(low, high, n_trajectories, random_state=None)

Generate a stochastic process with integer outputs uniformly sampled from the range [low, high).

A time index must be provided at construction, and the domain will be generated based on the number of trajectories.

Parameters:

Name	Type	Description	Default
`low`	`int`	The lower bound (inclusive) of the random integers.	required
`high`	`int`	The upper bound (exclusive) of the random integers.	required
`n_trajectories`	`int`	The number of trajectories to generate.	required
`random_state`	`int \| None`	An optional seed for the random number generator to ensure reproducibility. If `None`, the random number generator is not seeded.	`None`

Raises:

Type	Description
`ValueError`	If the time index is not provided at construction.

Returns:

Name	Type	Description
`self`	`StochasticProcess`	A stochastic process with integer outputs uniformly sampled from the range [low, high).

Examples:

>>> from sigalg.core import Time
>>> from sigalg.processes import StochasticProcess
>>> T = Time.discrete(length=3)
>>> X = StochasticProcess(time=T).from_randint(low=0, high=6, n_trajectories=4, random_state=42)
>>> X
Stochastic process 'X':
time        0  1  2  3
trajectory
0           0  4  3  2
1           2  5  0  4
2           1  0  3  5
3           4  4  4  4

from_randnorm

from_randnorm(
    loc=0.0, scale=1.0, n_trajectories=1, random_state=None
)

Generate a stochastic process with outputs sampled from a normal distribution.

Parameters:

Name	Type	Description	Default
`loc`	`float`	The mean (center) of the normal distribution.	`0.0`
`scale`	`float`	The standard deviation (spread or width) of the normal distribution.	`1.0`
`n_trajectories`	`int`	The number of trajectories to generate.	`1`
`random_state`	`int \| None`		`None`

Raises:

Type	Description
`ValueError`	If the time index is not provided at construction.

Returns:

Name	Type	Description
`self`	`StochasticProcess`	A stochastic process with outputs sampled from a normal distribution.

Examples:

>>> from sigalg.core import Time
>>> from sigalg.processes import StochasticProcess
>>> T = Time.discrete(length=2)
>>> X = StochasticProcess(time=T).from_randnorm(n_trajectories=3, random_state=42)
>>> X
Stochastic process 'X':
time               0         1         2
trajectory
0           0.304717 -1.039984  0.750451
1           0.940565 -1.951035 -1.302180
2           0.127840 -0.316243 -0.016801

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, or if `length` is provided and is not a positive integer.

Returns:

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

Examples:

>>> from sigalg.processes import RandomWalk
>>> X = RandomWalk(p=0.7, is_discrete_time=True).from_simulation(n_trajectories=5, length=3, random_state=42)
>>> X
Stochastic process 'X':
time        0  1  2  3
trajectory
0           0 -1  0 -1
1           0  1  2  1
2           0 -1 -2 -1
3           0  1  2  1
4           0  1  0  1

from_time

from_time()

Define a stochastic process whose trajectories are the time index itself.

Raises:

Type	Description
`ValueError`	If the time index or domain is not provided at construction.

Returns:

Name	Type	Description
`self`	`StochasticProcess`	A stochastic process whose trajectories are the time index itself.

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_time()
>>> print(X)
Stochastic process 'X':
time    0  1  2  3
sample
0       0  1  2  3
1       0  1  2  3

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 if the time indices of the process and the filtration do not have a non-empty intersection.

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,
    probability_measure=None,
    rtol=1e-05,
    atol=1e-08,
)

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

A stochastic process \(X_t\) with index set \(T\) is a martingale relative to a filtration \(\mathcal{F}_t\) if

\[ E(X_{t+1} | \mathcal{F}_t) = X_t \]

for all \(t\in T\) for which \(t+1 \in T\).

As of this writing, this method is only implemented for discrete-state processes. Even so, beware that the check is 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`
`probability_measure`	`ProbabilityMeasure \| None`	The probability measure with respect to which the martingale property is checked. If `None`, the probability measure 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, or if the provided probability measure is not an instance of ProbabilityMeasure, or its sample space does not match the domain 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_submartingale

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

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

A stochastic process \(X_t\) with index set \(T\) is a submartingale relative to a filtration \(\mathcal{F}_t\) if

\[ E(X_{t+1} | \mathcal{F}_t) \geq X_t \]

for all \(t\in T\) for which \(t+1 \in T\).

As of this writing, this method is only implemented for discrete-state processes. Even so, beware that the check is 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 submartingale property is checked. If None, the natural filtration of the process is used.	`None`
`probability_measure`	`ProbabilityMeasure \| None`	The probability measure with respect to which the submartingale property is checked. If `None`, the probability measure 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, or if the provided probability measure is not an instance of ProbabilityMeasure, or its sample space does not match the domain 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,
    probability_measure=None,
    rtol=1e-05,
    atol=1e-08,
)

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

A stochastic process \(X_t\) with index set \(T\) is a supermartingale relative to a filtration \(\mathcal{F}_t\) if

\[ E(X_{t+1} | \mathcal{F}_t) \leq X_t \]

for all \(t\in T\) for which \(t+1 \in T\).

As of this writing, this method is only implemented for discrete-state processes. Even so, beware that the check is 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 supermartingale property is checked. If None, the natural filtration of the process is used.	`None`
`probability_measure`	`ProbabilityMeasure \| None`	The probability measure with respect to which the supermartingale property is checked. If `None`, the probability measure 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, or if the provided probability measure is not an instance of ProbabilityMeasure, or its sample space does not match the domain 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.

StoppingTime

Bases: RandomVariable

Pass.

from_dict

from_dict(outputs)

Pass.

API Reference

processes

BranchingProcess

BrownianMotion

IIDProcess

MarkovChain

PoissonProcess

ProcessTransforms

cumprod staticmethod

cumsum staticmethod

discount staticmethod

increments staticmethod

insert_rv staticmethod

is_monotonic staticmethod

ito_integral staticmethod

max_value staticmethod

mean staticmethod

min_value staticmethod

pointwise_map staticmethod

remove_rv staticmethod

stopped staticmethod

sum staticmethod

to_counting_process classmethod

transform staticmethod

RandomWalk

StochasticProcess

at 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_numpy

from_pandas

from_randint

from_randnorm

from_simulation

from_time

is_adapted

is_martingale

is_submartingale

is_supermartingale

plot_trajectories

print_trajectories_and_probabilities

StoppingTime

from_dict

cumprod `staticmethod`

cumsum `staticmethod`

discount `staticmethod`

increments `staticmethod`

insert_rv `staticmethod`

is_monotonic `staticmethod`

ito_integral `staticmethod`

max_value `staticmethod`

mean `staticmethod`

min_value `staticmethod`

pointwise_map `staticmethod`

remove_rv `staticmethod`

stopped `staticmethod`

sum `staticmethod`

to_counting_process `classmethod`

transform `staticmethod`

at `property`

last_rv `property`

n_trajectories `property`

natural_filtration `property`

probability_measure `property` `writable`

random_variables `property`

range `property`

time `property` `writable`