API Reference
Module containing components for stochastic processes, including base classes, transforms, and specific types of processes.
Bases: StochasticProcess
A class representing a Brownian motion.
The constructor is not intended for direct usage. Instead, user's should call the generate class method. See the Examples section below.
See the Notes section below for the mathematical details.
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.generate(n_trajectories=4, index=T, random_state=42)
>>> print(X)
Brownian motion 'X':
t 0.100000 0.433333 0.766667 1.100000
omega
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
generate(
mode="sim",
n_trajectories=None,
domain_name=None,
index=None,
name="X",
random_state=None,
)
Simulate trajectories of the Brownian motion by Monte Carlo simulation.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
mode
|
Literal['enum', 'sim']
|
This parameter is included only for consistency with the parent class |
"sim"
|
n_trajectories
|
int | None
|
The number of trajectories to simulate. |
None
|
domain_name
|
Hashable | None
|
|
None
|
index
|
IndexLike | None
|
The index of the stochastic process. |
None
|
random_state
|
int | Generator | None
|
An optional random state for reproducibility. |
None
|
name
|
Hashable
|
The name of the stochastic process. |
"X"
|
Returns:
| Name | Type | Description |
|---|---|---|
info |
dict[str, object]
|
A dictionary containing the names of the BrownianMotion-specific parameters and their values. |
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.generate(n_trajectories=4, index=T, random_state=42)
>>> print(X)
Brownian motion 'X':
t 0.100000 0.433333 0.766667 1.100000
omega
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
Bases: StochasticProcess
A class representing an independent and identically distributed (IID) stochastic process.
The constructor is not intended for direct usage. Instead, user's should call the generate method. See the Examples section below.
Examples:
Enumerate all length-2 trajectories of an IID process consisting of Bernoulli random variables.
>>> from scipy.stats import bernoulli, poisson
>>> from sigalg.core import Time
>>> from sigalg.processes import IIDProcess
>>> T = Time.discrete(length=2)
>>> X = IIDProcess.generate(
... mode="enum",
... distribution=bernoulli(p=0.25),
... support=[0, 1],
... index=T,
... )
>>> print(X)
IID process 'X':
t 0 1 2
omega
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
Simulate ten length-2 trajectories from an IID process consisting of Poisson random variables.
>>> Y = IIDProcess.generate(
... mode="sim",
... distribution=poisson(mu=1.0),
... n_trajectories=10,
... index=T,
... name="Y",
... random_state=42,
... )
>>> print(Y)
IID process 'Y':
t 0 1 2
omega
0 1 2 3
1 1 3 0
2 1 3 3
3 1 0 3
4 1 0 0
5 1 1 2
6 2 0 1
7 0 0 3
8 1 0 0
9 1 2 1
property
distribution
Get the distribution of the IID process.
Returns:
| Name | Type | Description |
|---|---|---|
distribution |
rv_frozen | multinomial_frozen
|
The distribution of the IID process. |
property
support
Get the support of the IID process.
Returns:
| Name | Type | Description |
|---|---|---|
support |
list | dict | None
|
The support of the IID process. If the support is infinite, this property will be |
generate(
distribution,
support=None,
mode="sim",
n_trajectories=None,
domain_name=None,
index=None,
length=None,
name="X",
random_state=None,
)
Generate trajectories of the IID process by either exhaustive enumeration or Monte Carlo simulation.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
distribution
|
rv_frozen | multinomial_frozen
|
A frozen random variable from scipy.stats representing the common distribution of the IID process. |
required |
support
|
list | dict | None
|
Either a list containing the support of |
None
|
mode
|
Literal['enum', 'sim']
|
Whether to generate trajectories by exhaustive enumeration or Monte Carlo simulation. |
"sim"
|
n_trajectories
|
int | None
|
The number of trajectories to simulate. This parameter is ignored if the generation mode is set to |
None
|
domain_name
|
Hashable | None
|
The name of the domain. If |
None
|
index
|
IndexLike | None
|
The index of the stochastic process. One of |
None
|
length
|
int | None
|
The length of the trajectories of the stochastic process. One of |
None
|
name
|
Hashable
|
The name of the stochastic process. |
"X"
|
random_state
|
int | Generator | None
|
An optional random state for reproducibility. |
None
|
Returns:
| Name | Type | Description |
|---|---|---|
info |
dict[str, object]
|
A dictionary containing the names of the IIDProcess-specific parameters and their values. |
Examples:
>>> from scipy.stats import bernoulli, poisson
>>> from sigalg.core import Time
>>> from sigalg.processes import IIDProcess
Enumerate all length-2 trajectories of an IID process consisting of Bernoulli random variables.
>>> T = Time.discrete(length=2)
>>> X = IIDProcess.generate(
... mode="enum",
... distribution=bernoulli(p=0.25),
... support=[0, 1],
... index=T,
... )
>>> print(X)
IID process 'X':
t 0 1 2
omega
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
Enumerate all length-2 trajectories of an IID process consisting of Bernoulli random variables supported on {a, b}.
>>> Y = IIDProcess.generate(
... mode="enum",
... distribution=bernoulli(p=0.25),
... support={0: "a", 1: "b"},
... index=T,
... name="Y",
... )
>>> print(Y)
IID process 'Y':
t 0 1 2
omega
0 a a a
1 a a b
2 a b a
3 a b b
4 b a a
5 b a b
6 b b a
7 b b b
Simulate 10 length-2 trajectories from an IID process consisting of Poisson random variables.
>>> Y = IIDProcess.generate(
... mode="sim",
... distribution=poisson(mu=1.0),
... n_trajectories=10,
... index=T,
... name="Y",
... random_state=42,
... )
>>> print(Y)
IID process 'Y':
t 0 1 2
omega
0 1 2 3
1 1 3 0
2 1 3 3
3 1 0 3
4 1 0 0
5 1 1 2
6 2 0 1
7 0 0 3
8 1 0 0
9 1 2 1
Bases: StochasticProcess
A class representing a Markov chain stochastic process.
The constructor is not intended for direct usage. Instead, user's should call the generate class method. See the Examples section below.
See the Notes section below for the mathematical details.
Examples:
>>> import numpy as np
>>> from sigalg.core import (
... Domain,
... ParametrizedProbabilityMeasure,
... ProbabilityMeasure,
... )
>>> from sigalg.processes import MarkovChain
Exhaustively enumerate all trajectories of an order-2 Markov chain of length 2 on a binary state space. The initial distribution and transition kernel are instances of ProbabilityMeasure and ParametrizedProbabilityMeasure, respectively.
>>> S = Domain.from_sequence(size=2, variable_name="s", name="S")
>>> S_0 = Domain.from_sequence(size=2, variable_name="s_0", name="S_0")
>>> pi = ProbabilityMeasure(domain=S, mapping={0: 0.2, 1: 0.8}, name="pi")
>>> K = ParametrizedProbabilityMeasure.from_domains(
... measure_domain=S,
... parameter_domain=S_0,
... mapping={
... (0, 0): 0.4,
... (0, 1): 0.6,
... (1, 0): 0.7,
... (1, 1): 0.3,
... },
... name="K",
... )
>>> X = MarkovChain.generate(
... kernel=K,
... initial_distribution=pi,
... mode="enum",
... length=2,
... )
>>> print(X)
Markov chain 'X':
t 0 1 2
omega
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
Check that the joint distribution factors according to the Markov property.
>>> P = X.measure
>>> np.allclose(
... P >> X, (P >> X[0]) * (P.conditional(X[0]) >> X[1]) * (P.conditional(X[1]) >> X[2])
... )
True
Markov chains may also be generated by passing array-like objects for the initial distribution and transition kernel.
>>> Y = MarkovChain.generate(
... kernel=[[0.4, 0.6], [0.7, 0.3]],
... initial_distribution=[0.2, 0.8],
... mode="enum",
... length=2,
... name="Y",
... )
>>> print(Y)
Markov chain 'Y':
t 0 1 2
omega
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
Exhaustively enumerate all trajectories of an order-3 Markov chain of length 3 on a binary state space. The initial distribution and transition kernel are randomly generated instances of ProbabilityMeasure and ParametrizedProbabilityMeasure, respectively. Note, however, that the initial distribution is defined on the second Cartesian power of the state space and the transition kernel is represented by a rank-3 tensor.
>>> rng = np.random.default_rng(42)
>>> pi = ProbabilityMeasure.from_rand(
... domain=S ^ 2,
... name="pi",
... random_state=rng,
... )
>>> K = ParametrizedProbabilityMeasure.from_rand(
... measure_domain=S,
... parameter_domain=S ^ 2,
... random_state=rng,
... )
>>> Z = MarkovChain.generate(
... kernel=K,
... initial_distribution=pi,
... mode="enum",
... length=3,
... name="Z",
... )
>>> print(Z)
Markov chain 'Z':
t 0 1 2 3
omega
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
8 1 0 0 0
9 1 0 0 1
10 1 0 1 0
11 1 0 1 1
12 1 1 0 0
13 1 1 0 1
14 1 1 1 0
15 1 1 1 1
Check that the joint distribution factors according to the order-3 Markov property.
>>> Q = Z.measure.with_name("Q")
>>> np.allclose(
... Q >> Z,
... (Q >> Z[0, 1])
... * (Q.conditional(Z[0, 1]) >> Z[2])
... * (Q.conditional(Z[1, 2]) >> Z[3]),
... )
True
property
initial_distribution
The initial distribution of the Markov chain as a NumPy array.
Returns:
| Name | Type | Description |
|---|---|---|
initial_distribution |
ndarray
|
The initial distribution of the Markov chain. |
Examples:
>>> import numpy as np
>>> from sigalg.processes import MarkovChain
>>> rng = np.random.default_rng(42)
Generate a random Markov chain of order 3 and length 5.
>>> X = MarkovChain.from_rand(
... order=3,
... n_states=3,
... mode="enum",
... length=5,
... random_state=rng,
... )
Print the shape of the initial distribution.
>>> X.initial_distribution.shape
(3, 3, 3)
property
kernel
The transition tensor (kernel) of the Markov chain.
Returns:
| Name | Type | Description |
|---|---|---|
transition_matrix |
ndarray
|
The transition tensor (kernel) of the Markov chain. |
Examples:
>>> import numpy as np
>>> from sigalg.processes import MarkovChain
>>> rng = np.random.default_rng(42)
Generate a random Markov chain of order 3 and length 5.
>>> X = MarkovChain.from_rand(
... order=3,
... n_states=3,
... mode="enum",
... length=5,
... random_state=rng,
... )
Print the shape of the transition tensor.
>>> X.kernel.shape
(3, 3, 3, 3)
property
n_states
Get the number of states in the Markov chain.
Returns:
| Name | Type | Description |
|---|---|---|
n_states |
int
|
The number of states in the Markov chain. |
Examples:
>>> import numpy as np
>>> from sigalg.processes import MarkovChain
>>> rng = np.random.default_rng(42)
Generate a random Markov chain of order 4 and length 5.
>>> X = MarkovChain.from_rand(
... order=4,
... n_states=3,
... mode="enum",
... length=5,
... random_state=rng,
... )
Print the number of states of the Markov chain.
>>> X.n_states
3
property
order
Get the order of the Markov chain.
Returns:
| Name | Type | Description |
|---|---|---|
order |
int
|
The order of the Markov chain. |
Examples:
>>> import numpy as np
>>> from sigalg.processes import MarkovChain
>>> rng = np.random.default_rng(42)
Generate a random Markov chain of order 4 and length 5.
>>> X = MarkovChain.from_rand(
... order=4,
... n_states=3,
... mode="enum",
... length=5,
... random_state=rng,
... )
Print the order of the Markov chain.
>>> X.order
4
property
state_space
Get the state space of the Markov chain.
Returns:
| Name | Type | Description |
|---|---|---|
state_space |
ndarray
|
An array representing the states of the Markov chain. |
classmethod
from_rand(
order=1,
n_states=2,
mode="sim",
n_trajectories=None,
domain_name=None,
index=None,
length=None,
name="X",
random_state=None,
)
Generate a random Markov chain.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
order
|
int
|
The order of the Markov chain. |
1
|
n_states
|
int
|
The number of states of the Markov chain. |
2
|
mode
|
Literal['enum', 'sim']
|
Whether to generate trajectories by exhaustive enumeration or Monte Carlo simulation. |
"sim"
|
n_trajectories
|
int | None
|
The number of trajectories to simulate. If the generation mode is set to |
None
|
domain_name
|
Hashable | None
|
The name of the domain. If |
None
|
index
|
IndexLike | None
|
The index of the stochastic process. One of |
None
|
length
|
int | None
|
The length of the trajectories of the stochastic process. One of |
None
|
name
|
Hashable
|
The name of the stochastic process. |
"X"
|
random_state
|
int | Generator | None
|
An optional random state for reproducibility. |
None
|
Returns:
| Name | Type | Description |
|---|---|---|
mc |
MarkovChain
|
A randomly generated Markov chain. |
Examples:
>>> import numpy as np
>>> from sigalg.processes import MarkovChain
>>> rng = np.random.default_rng(42)
Exhaustively generate all trajectories of an order-3 Markov chain of length 5.
>>> X = MarkovChain.from_rand(
... order=3,
... n_states=3,
... mode="enum",
... length=5,
... random_state=rng,
... )
>>> print(X)
Markov chain 'X':
t 0 1 2 3 4 5
omega
0 0 0 0 0 0 0
1 0 0 0 0 0 1
2 0 0 0 0 0 2
3 0 0 0 0 1 0
4 0 0 0 0 1 1
... .. .. .. .. .. ..
724 2 2 2 2 1 1
725 2 2 2 2 1 2
726 2 2 2 2 2 0
727 2 2 2 2 2 1
728 2 2 2 2 2 2
[729 rows x 6 columns]
Check that the joint probability measure factors according to the Markov property.
>>> P = X.measure
>>> np.allclose(
... P >> X,
... (P >> X[0, 1, 2])
... * (P.conditional(X[0, 1, 2]) >> X[3])
... * (P.conditional(X[1, 2, 3]) >> X[4])
... * (P.conditional(X[2, 3, 4]) >> X[5]),
... )
True
generate(
kernel,
initial_distribution,
mode="sim",
n_trajectories=None,
domain_name=None,
index=None,
length=None,
name="X",
random_state=None,
)
Generate trajectories of the Markov chain by either exhaustive enumeration or Monte Carlo simulation.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
kernel
|
ArrayLike | ParametrizedProbabilityMeasure
|
Either a parametrized probability measure the transition kernel or an object that can be coerced into a transition tensor (i.e., a multi-dimensional numpy array). |
required |
initial_distribution
|
ArrayLike | ProbabilityMeasure
|
Either a probability measure representing the initial distribution or an object that can be coerced into a probability tensor (i.e., a multi-dimensional numpy array). |
required |
mode
|
Literal['enum', 'sim']
|
Whether to generate trajectories by exhaustive enumeration or Monte Carlo simulation. |
"sim"
|
n_trajectories
|
int | None
|
The number of trajectories to simulate. If the generation mode is set to |
None
|
domain_name
|
Hashable | None
|
The name of the domain. If |
None
|
index
|
IndexLike | None
|
The index of the stochastic process. One of |
None
|
length
|
int | None
|
The length of the trajectories of the stochastic process. One of |
None
|
name
|
Hashable
|
The name of the stochastic process. |
"X"
|
random_state
|
int | Generator | None
|
An optional random state for reproducibility. |
None
|
Returns:
| Name | Type | Description |
|---|---|---|
info |
dict[str, object]
|
A dictionary containing the names of the MarkovChain-specific parameters and their values. |
Examples:
>>> import numpy as np
>>> from sigalg.core import (
... Domain,
... ParametrizedProbabilityMeasure,
... ProbabilityMeasure,
... )
>>> from sigalg.processes import MarkovChain
Exhaustively enumerate all trajectories of an order-2 Markov chain of length 2 on a binary state space. The initial distribution and transition kernel are instances of ProbabilityMeasure and ParametrizedProbabilityMeasure, respectively.
>>> S = Domain.from_sequence(size=2, variable_name="s", name="S")
>>> S_0 = Domain.from_sequence(size=2, variable_name="s_0", name="S_0")
>>> pi = ProbabilityMeasure(domain=S, mapping={0: 0.2, 1: 0.8}, name="pi")
>>> K = ParametrizedProbabilityMeasure.from_domains(
... measure_domain=S,
... parameter_domain=S_0,
... mapping={
... (0, 0): 0.4,
... (0, 1): 0.6,
... (1, 0): 0.7,
... (1, 1): 0.3,
... },
... name="K",
... )
>>> X = MarkovChain.generate(
... kernel=K,
... initial_distribution=pi,
... mode="enum",
... length=2,
... )
>>> print(X)
Markov chain 'X':
t 0 1 2
omega
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
Check that the joint distribution factors according to the Markov property.
>>> P = X.measure
>>> np.allclose(
... P >> X, (P >> X[0]) * (P.conditional(X[0]) >> X[1]) * (P.conditional(X[1]) >> X[2])
... )
True
Markov chains may also be generated by passing array-like objects for the initial distribution and transition kernel.
>>> Y = MarkovChain.generate(
... kernel=[[0.4, 0.6], [0.7, 0.3]],
... initial_distribution=[0.2, 0.8],
... mode="enum",
... length=2,
... name="Y",
... )
>>> print(Y)
Markov chain 'Y':
t 0 1 2
omega
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
Exhaustively enumerate all trajectories of an order-3 Markov chain of length 3 on a binary state space. The initial distribution and transition kernel are randomly generated instances of ProbabilityMeasure and ParametrizedProbabilityMeasure, respectively. Note, however, that the initial distribution is defined on the second Cartesian power of the state space and the transition kernel is represented by a rank-3 tensor.
>>> rng = np.random.default_rng(42)
>>> pi = ProbabilityMeasure.from_rand(
... domain=S ^ 2,
... name="pi",
... random_state=rng,
... )
>>> K = ParametrizedProbabilityMeasure.from_rand(
... measure_domain=S,
... parameter_domain=S ^ 2,
... random_state=rng,
... )
>>> Z = MarkovChain.generate(
... kernel=K,
... initial_distribution=pi,
... mode="enum",
... length=3,
... name="Z",
... )
>>> print(Z)
Markov chain 'Z':
t 0 1 2 3
omega
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
8 1 0 0 0
9 1 0 0 1
10 1 0 1 0
11 1 0 1 1
12 1 1 0 0
13 1 1 0 1
14 1 1 1 0
15 1 1 1 1
Check that the joint distribution factors according to the order-3 Markov property.
>>> Q = Z.measure.with_name("Q")
>>> np.allclose(
... Q >> Z,
... (Q >> Z[0, 1])
... * (Q.conditional(Z[0, 1]) >> Z[2])
... * (Q.conditional(Z[1, 2]) >> Z[3]),
... )
True
Bases: StochasticProcess
A class representing a Poisson stochastic process.
The constructor is not intended for direct usage. Instead, user's should call the class method generate. See the Examples section below.
See also the Notes section below for the mathematical details.
Examples:
>>> from math import ceil, sqrt
>>> import numpy as np
>>> 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))
Simulate 10 trajectories of the Poisson process with the specified parameters and print them.
>>> X = PoissonProcess.generate(
... rate=rate,
... max_count=max_count,
... index=time,
... n_trajectories=10,
... random_state=42,
... )
>>> print(X)
Poisson process 'X':
t 0.0000 1.5625 3.1250 4.6875 6.2500
omega
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.generate(
... rate=rate,
... max_count=max_count,
... index=time,
... n_trajectories=50_000,
... random_state=42,
... name="Y",
... )
Extract the simulated values of the final random variable and their empirical probabilities.
>>> final_time = Y.time[-1]
>>> empirical_probabilities = Y.prob_measure >> Y[final_time]
Compute the theoretical probabilities of the final random variable and compare to the empirical ones.
>>> theoretical_probabilities = poisson(mu=rate * final_time).pmf(
... empirical_probabilities.domain
... )
>>> np.allclose(
... empirical_probabilities.data,
... theoretical_probabilities,
... rtol=0,
... atol=1e-2,
... )
True
Notes
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.
property
max_count
The maximum count of events to simulate.
Returns:
| Name | Type | Description |
|---|---|---|
max_count |
int
|
The maximum count of events to simulate. |
property
rate
The rate (lambda) of the Poisson process.
Returns:
| Name | Type | Description |
|---|---|---|
rate |
Real
|
The rate of the Poisson process. |
generate(
rate,
max_count,
mode="sim",
n_trajectories=None,
domain_name=None,
index=None,
name="X",
random_state=None,
)
Generate trajectories of the Poisson process by Monte Carlo simulation.
In this implementation, trajectories are simulated until one trajectory reaches the specified max_count of events, and then the (required) user-provided 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.
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 |
mode
|
Literal['enum', 'sim']
|
This parameter is included only for consistency with the parent class |
"sim"
|
n_trajectories
|
int | None
|
The number of trajectories to simulate. |
None
|
domain_name
|
Hashable | None
|
The name of the domain. If |
None
|
index
|
Time | IndexLike | None
|
The index of the stochastic process. |
None
|
name
|
Hashable
|
The name of the stochastic process. |
"X"
|
random_state
|
int | Generator | None
|
An optional random state for reproducibility. |
None
|
Returns:
| Name | Type | Description |
|---|---|---|
info |
dict[str, object]
|
A dictionary containing the names of the PoissonProcess-specific parameters and their values. |
Examples:
>>> from math import ceil, sqrt
>>> 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))
Simulate 10 trajectories of the Poisson process with the specified parameters and print them.
>>> X = PoissonProcess.generate(
... rate=rate,
... max_count=max_count,
... index=time,
... n_trajectories=10,
... random_state=42,
... )
>>> print(X)
Poisson process 'X':
t 0.0000 1.5625 3.1250 4.6875 6.2500
omega
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
Class for operators and transforms on stochastic processes.
classmethod
at(process, stopping_time, name=None)
Pass.
classmethod
discount(process, rate=None, gross_return=None, name=None)
Return the discounted process of a given stochastic process.
See the Notes section below for the mathematical details.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
process
|
StochasticProcess
|
The original process to be discounted. |
required |
rate
|
Real | None
|
The per-period rate. If |
None
|
gross_return
|
Real | None
|
The per-period gross return. If |
None
|
name
|
Hashable | None
|
The name of the discounted process. If |
None
|
Returns:
| Name | Type | Description |
|---|---|---|
discounted_process |
StochasticProcess
|
The discounted process. |
Examples:
>>> from sigalg.finance import BinomialPricingModel
>>> from sigalg.processes import ProcessTransforms
>>> S = BinomialPricingModel.generate(
... initial_price=100,
... up_prob=0.7,
... up_factor=1.1,
... down_factor=0.9,
... risk_free_rate=0.05,
... mode="enum",
... length=3,
... )
>>> print(S)
Binomial price process 'S':
t 0 1 2 3
omega
0 100.0 110.0 121.0 133.1
1 100.0 110.0 121.0 108.9
2 100.0 110.0 99.0 108.9
3 100.0 110.0 99.0 89.1
4 100.0 90.0 99.0 108.9
5 100.0 90.0 99.0 89.1
6 100.0 90.0 81.0 89.1
7 100.0 90.0 81.0 72.9
>>> S_discount = ProcessTransforms.discount(process=S, rate=0.05)
>>> print(S_discount)
Stochastic process 'S_discount':
t 0 1 2 3
omega
0 100.0 104.761905 109.750567 114.976784
1 100.0 104.761905 109.750567 94.071914
2 100.0 104.761905 89.795918 94.071914
3 100.0 104.761905 89.795918 76.967930
4 100.0 85.714286 89.795918 94.071914
5 100.0 85.714286 89.795918 76.967930
6 100.0 85.714286 73.469388 76.967930
7 100.0 85.714286 73.469388 62.973761
>>> Q = S.emm()
>>> S_discount.is_martingale(prob_measure=Q)
True
Notes
The discounted process is given by
where \(S_t\) is the original process and \(R = 1 + r\) is the gross return and \(r\) is the rate.
classmethod
increments(process, forward=True, name=None)
Compute the increments of a stochastic process along its time index.
See the Notes section below for the mathematical details.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
process
|
StochasticProcess
|
The stochastic process for which to compute the increments. |
required |
forward
|
bool
|
If |
True
|
name
|
Hashable | None
|
The name of the transformed process. If |
None
|
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.generate(mode="enum", p=0.5, index=T, initial_state=3)
>>> print(X)
Random walk 'X':
t 0 1 2
omega
0 3 2 1
1 3 2 3
2 3 4 3
3 3 4 5
>>> X_increments = X.increments(forward=True)
>>> print(X_increments)
Stochastic process 'X_increments':
t 0 1
omega
0 -1 -1
1 -1 1
2 1 -1
3 1 1
>>> X_increments = X.increments(forward=False)
>>> print(X_increments)
Stochastic process 'X_increments':
t 1 2
omega
0 -1 -1
1 -1 1
2 1 -1
3 1 1
Notes
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
for each \(t=t_0,\ldots,T-1\). The second type are backward increments, which results in a stochastic process \(\Delta X_t\) where
for each \(t=t_0+1,\ldots,T\).
classmethod
insert_rv(process, time, rv=None, state=None, 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 |
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 |
None
|
state
|
Hashable | None
|
A constant state to assign to the inserted random variable for all trajectories. One or the other of |
None
|
name
|
Hashable | None
|
The name of the new stochastic process. If |
None
|
Returns:
| Name | Type | Description |
|---|---|---|
inserted_process |
StochasticProcess | None
|
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, ProcessTransforms
>>> T = Time.discrete(start=1, length=2)
>>> X = IIDProcess.generate(mode="enum", distribution=bernoulli(p=0.5), support=[0, 1], index=T)
>>> print(X)
IID process 'X':
t 1 2 3
omega
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.from_constant(*X.prob_space, constant=0)
>>> X_insert = ProcessTransforms.insert_rv(process=X, rv=X0, time=0)
>>> print(X_insert)
Stochastic process 'X_insert':
t 0 1 2 3
omega
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
classmethod
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
|
Returns:
| Name | Type | Description |
|---|---|---|
is_monotonic |
bool
|
|
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.generate(mode="enum", distribution=bernoulli(p=0.6), support=[0, 1], index=T)
>>> print(X)
IID process 'X':
t 1 2 3
omega
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':
t 1 2 3
omega
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
classmethod
ito_integral(integrand=None, integrator=None, 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. |
None
|
integrator
|
StochasticProcess
|
The stochastic process with respect to which the integral is computed. |
None
|
name
|
Hashable | None
|
The name of the transformed process. If |
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 ProcessTransforms, RandomWalk
>>> T = Time.discrete(length=2)
>>> X = RandomWalk.generate(mode="enum", p=0.6, initial_state=1, index=T)
>>> print(X)
Random walk 'X':
t 0 1 2
omega
0 1 0 -1
1 1 0 1
2 1 2 1
3 1 2 3
>>> integral = ProcessTransforms.ito_integral(integrator=X)
>>> print(1 + integral)
Stochastic process '(1 + int dX)':
t 0 1 2
omega
0 1 0.0 -1.0
1 1 0.0 1.0
2 1 2.0 1.0
3 1 2.0 3.0
classmethod
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
|
int | None
|
The position at which to remove the random variable. If |
None
|
name
|
Hashable | None
|
The name of the transformed process. If |
None
|
Returns:
| Name | Type | Description |
|---|---|---|
removed_process |
StochasticProcess | None
|
A new stochastic process with the random variable removed at the specified time. |
Examples:
>>> from sigalg.core import Time
>>> from sigalg.processes import ProcessTransforms, RandomWalk
>>> T = Time.discrete(start=1, length=2)
>>> X = RandomWalk.generate(mode="enum", p=0.6, initial_state=0, index=T)
>>> print(X)
Random walk 'X':
t 1 2 3
omega
0 0 -1 -2
1 0 -1 0
2 0 1 0
3 0 1 2
>>> X_remove = ProcessTransforms.remove_rv(process=X, time=2)
>>> print(X_remove)
Stochastic process 'X_remove':
t 1 3
omega
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.generate(mode="enum", p=0.6, initial_state=0, index=S, name="Y")
>>> print(Y)
Random walk 'Y':
t 0.0 0.1 0.2 0.3
omega
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
>>> Y_remove = ProcessTransforms.remove_rv(process=Y, pos=2)
>>> print(Y_remove)
Stochastic process 'Y_remove':
t 0.0 0.1 0.3
omega
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
classmethod
stopped(process, stopping_time, name=None)
Get the stopped process from a stopping time.
See the Notes section below for the mathematical details.
Examples:
>>> from math import inf
>>> from sigalg.core import Time
>>> from sigalg.processes import RandomWalk, StoppingTime
>>> T = Time.discrete(start=1, stop=10)
>>> S = RandomWalk.generate(
... mode="sim",
... p=0.4,
... initial_state=10,
... index=T,
... n_trajectories=8,
... random_state=42,
... name="S",
... )
>>> print(S)
Random walk 'S':
t 1 2 3 4 5 6 7 8 9 10
omega
0 10 11 10 11 12 11 12 13 14 13
1 10 9 8 9 10 11 10 9 8 7
2 10 11 12 13 12 13 14 15 14 13
3 10 9 8 9 10 11 10 9 8 7
4 10 9 8 7 8 7 8 9 8 9
5 10 11 10 9 10 9 8 7 8 9
6 10 11 12 11 10 9 8 9 8 7
7 10 11 12 11 10 9 8 7 6 5
>>> tau = StoppingTime.from_filtration(
... process=S,
... mapping={
... 0: inf,
... 1: 3,
... 2: inf,
... 3: 3,
... 4: 3,
... 5: 7,
... 6: 7,
... 7: 7,
... },
... )
>>> print(tau)
Stopping time 'tau':
tau
omega
0 inf
1 3.0
2 inf
3 3.0
4 3.0
5 7.0
6 7.0
7 7.0
>>> S_stopped = S.stopped(stopping_time=tau)
>>> print(S_stopped)
Stochastic process 'S^tau':
t 1 2 3 4 5 6 7 8 9 10
omega
0 10 11 10 11 12 11 12 13 14 13
1 10 9 8 8 8 8 8 8 8 8
2 10 11 12 13 12 13 14 15 14 13
3 10 9 8 8 8 8 8 8 8 8
4 10 9 8 8 8 8 8 8 8 8
5 10 11 10 9 10 9 8 8 8 8
6 10 11 12 11 10 9 8 8 8 8
7 10 11 12 11 10 9 8 8 8 8
Notes
Let \(X\) be a \(T\)-indexed stochastic process on a probability space \((\Omega, \mathcal{F},P)\), and let \(\tau: \Omega \to T\) be a stopping time. The stopped process, denoted \(X^\tau\), is defined for all \(t\in T\) by
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
|
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 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, variable_name="count")
Exponential interarrival times with given rate
>>> interarrival_times = IIDProcess.generate(
... mode="sim",
... distribution=expon(scale=1 / rate),
... name="interarrival_times",
... index=counts,
... n_trajectories=n_trajectories,
... random_state=random_state,
... )
>>> print(interarrival_times)
IID process 'interarrival_times':
count 1 2 3 4 5
omega
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")
>>> print(arrival_times)
Stochastic process 'arrival_times':
count 1 2 3 4 5
omega
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()
>>> T = 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=T,
... ).with_name("poisson")
>>> print(poisson)
Stochastic process 'poisson':
t 0.000000 0.769139 1.538278 2.307417 3.076556 3.845695
omega
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
Bases: StochasticProcess
A class representing a random walk stochastic process.
The constructor is not intended for direct usage. Instead, user's should call the generate method. See the Examples section below.
See the Notes section below for the mathematical details.
Examples:
>>> from math import comb
>>> from sigalg.core import Time
>>> from sigalg.processes import RandomWalk
Generate all trajectories of a random walk with probability p=0.75 of stepping right one unit, and 0.25 of stepping left one unit.
>>> T = Time.discrete(length=3)
>>> X = RandomWalk.generate(mode="enum", p=0.75, index=T)
>>> print(X)
Random walk 'X':
t 0 1 2 3
omega
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 the values of the X_3 component random variable and its corresponding law.
>>> print(X[3].pushforward())
Probability measure 'P_X_3':
P_X_3
X_3
-3 0.015625
-1 0.140625
1 0.421875
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
property
writable
initial_state
Get the initial state of the random walk.
The initial_state parameter is settable. See the Examples below.
Returns:
| Name | Type | Description |
|---|---|---|
initial_state |
Real
|
The initial state of the random walk. |
Examples:
>>> from sigalg.core import Time
>>> from sigalg.processes import RandomWalk
Generate a symmetric random walk (i.e., with p=0.5) with initial state 0 and print its trajectories.
>>> T = Time.discrete(start=1, length=2)
>>> X = RandomWalk.generate(mode="enum", p=0.5, initial_state=0, index=T)
>>> print(X)
Random walk 'X':
t 1 2 3
omega
0 0 -1 -2
1 0 -1 0
2 0 1 0
3 0 1 2
Set the initial_state parameter to 0.5, regenerate the trajectories, and print the new trajectories.
>>> X.initial_state = 0.5
>>> X.regenerate()
>>> print(X)
Random walk 'X':
t 1 2 3
omega
0 0.5 -0.5 -1.5
1 0.5 -0.5 0.5
2 0.5 1.5 0.5
3 0.5 1.5 2.5
property
writable
p
Get the p parameter of the random walk.
The p parameter is settable. See the Examples below.
Returns:
| Name | Type | Description |
|---|---|---|
p |
Real
|
The probability of stepping "to the right" in the random walk. |
Examples:
>>> from sigalg.core import Time
>>> from sigalg.processes import RandomWalk
Generate a symmetric random walk (i.e., with p=0.5) and print its trajectories and probability measure.
>>> T = Time.discrete(start=1, length=2)
>>> X = RandomWalk.generate(mode="enum", p=0.5, initial_state=0, index=T)
>>> print(X)
Random walk 'X':
t 1 2 3
omega
0 0 -1 -2
1 0 -1 0
2 0 1 0
3 0 1 2
>>> print(X.measure)
Probability measure 'P':
P
omega
0 0.25
1 0.25
2 0.25
3 0.25
Set the p parameter to 0.7, regenerate the trajectories, and print the new probability measure.
>>> X.p = 0.7
>>> X.regenerate()
>>> print(X.measure)
Probability measure 'P':
P
omega
0 0.09
1 0.21
2 0.21
3 0.49
generate(
p,
initial_state=0,
mode="sim",
n_trajectories=None,
domain_name=None,
index=None,
length=None,
name="X",
random_state=None,
)
Generate trajectories of the random walk by either exhaustive enumeration or Monte Carlo simulation.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
p
|
Real
|
The probability of stepping "to the right". |
required |
initial_state
|
Real
|
The initial state of the random walk. |
0
|
mode
|
Literal['enum', 'sim']
|
Whether to generate trajectories by exhaustive enumeration or Monte Carlo simulation. |
"sim"
|
n_trajectories
|
int | None
|
The number of trajectories to simulate. If the generation mode is set to |
None
|
domain_name
|
Hashable | None
|
The name of the domain. If |
None
|
index
|
IndexLike | None
|
The index of the stochastic process. One of |
None
|
length
|
int | None
|
The length of the trajectories of the stochastic process. One of |
None
|
random_state
|
int | Generator | None
|
An optional random state for reproducibility. |
None
|
name
|
Hashable
|
The name of the stochastic process. |
"X"
|
Returns:
| Name | Type | Description |
|---|---|---|
info |
dict[str, object]
|
A dictionary containing the parameters |
Examples:
>>> from sigalg.core import Time
>>> from sigalg.processes import RandomWalk
Generate all length-3 trajectories of a random walk with a probability of 0.75 of stepping right.
>>> T = Time.discrete(length=3)
>>> X = RandomWalk.generate(
... mode="enum",
... p=0.75,
... index=T,
... )
>>> print(X)
Random walk 'X':
t 0 1 2 3
omega
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
Simulate ten length-3 trajectories of a random walk that begins at 2 and has probability 0.4 of stepping right.
>>> Y = RandomWalk.generate(
... mode="sim",
... p=0.4,
... initial_state=2,
... n_trajectories=10,
... index=T,
... name="Y",
... random_state=42,
... )
>>> print(Y)
Random walk 'Y':
t 0 1 2 3
omega
0 2 3 2 3
1 2 3 2 3
2 2 3 4 3
3 2 1 0 1
4 2 3 4 3
5 2 1 0 -1
6 2 3 4 5
7 2 1 2 3
8 2 3 2 1
9 2 1 0 1
Bases: RandomVector, ProcessTransformMethods
Base class for stochastic processes.
See the Notes section below for the mathematical details.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
domain
|
IndexLike | None
|
The domain of the underlying probability space. |
None
|
sig_alg
|
SigmaAlgebra | None
|
The sigma-algebra of the underlying probability space. |
None
|
measure
|
ProbabilityMeasure | None
|
The probability measure of the underlying probability space. |
None
|
mapping
|
MappingLike | None
|
The mapping defining the stochastic process. |
None
|
domain_name
|
Hashable | None
|
The name of the domain. |
None
|
index
|
IndexLike | None
|
The index for the stochastic process. |
None
|
index_name
|
Hashable | None
|
The name of the index. |
None
|
name
|
Hashable | None
|
The name of the stochastic process. If |
None
|
Examples:
>>> from sigalg.core import ProbabilityMeasure, SampleSpace, SigmaAlgebra
>>> from sigalg.processes import StochasticProcess
Create a probability space and a stochastic process.
>>> Omega = SampleSpace.from_sequence(size=3, variable_name="omega")
>>> F = SigmaAlgebra(
... domain=Omega,
... mapping={
... 0: 0,
... 1: 0,
... 2: 1,
... },
... )
>>> P = ProbabilityMeasure.from_rand(domain=F, random_state=42)
>>> X = StochasticProcess(
... domain=Omega,
... sig_alg=F,
... measure=P,
... mapping={
... 0: (0, 1),
... 1: (0, 1),
... 2: (1, 0),
... },
... )
>>> print(X)
Stochastic process 'X':
t 0 1
omega
0 0 1
1 0 1
2 1 0
Notes
Let \((\Omega, \mathcal{F}, P)\) be a probability space and let \(T\) be a linearly-ordered set. A (\(T\)-indexed) stochastic process is a collection of random variables \(X_t\) defined on the probability space, one for each \(t\in T\).
property
is_discrete_time
Whether the stochastic process is a discrete-time process.
Returns:
| Name | Type | Description |
|---|---|---|
is_discrete_time |
bool | None
|
|
cached
property
last_rv
Get the random variable corresponding to the last time point.
Returns:
| Name | Type | Description |
|---|---|---|
last_rv |
RandomVariable
|
The random variable corresponding to the last time point. |
Examples:
>>> from scipy.stats import bernoulli
>>> from sigalg.processes import IIDProcess
Exhaustively enumerate all length-2 trajectories of an IID Bernoulli process.
>>> X = IIDProcess.generate(
... mode="enum",
... distribution=bernoulli(p=0.5),
... support=[0, 1],
... length=2,
... )
>>> print(X)
IID process 'X':
t 0 1 2
omega
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
Get the last random variable of the process.
>>> print(X.last_rv)
Random variable 'X_2':
X_2
omega
0 0
1 1
2 0
3 1
4 0
5 1
6 0
7 1
property
writable
length
Get the length of the trajectories of the process, defined as the length of the underlying time interval.
The length property is settable. See the Examples section below for usage.
Returns:
| Name | Type | Description |
|---|---|---|
length |
int | None
|
The length of the trajectories of the process, or |
Examples:
>>> from scipy.stats import bernoulli
>>> from sigalg.processes import IIDProcess
Generate trajectories of an IID process with a specified length.
>>> X = IIDProcess.generate(
... mode="sim",
... distribution=bernoulli(0.75),
... support=[0, 1],
... n_trajectories=10,
... length=2,
... random_state=42,
... )
>>> print(X)
IID process 'X':
t 0 1 2
omega
0 0 1 0
1 1 1 0
2 0 0 1
3 1 1 0
4 1 0 1
5 1 1 1
6 0 1 0
7 1 0 0
8 0 1 1
9 1 1 1
Set the length of the trajectories and regenerate the process.
>>> X.length = 3
>>> X.regenerate()
>>> print(X)
IID process 'X':
t 0 1 2 3
omega
0 1 0 1 1
1 1 1 1 1
2 1 1 1 0
3 1 1 0 0
4 1 1 1 1
5 1 1 0 1
6 1 0 1 1
7 1 1 1 1
8 1 0 1 1
9 1 1 1 1
property
writable
mode
Get the generation mode of the process.
The mode property is settable. If the trajectories of a process were first generated by calling from_enumeration, be sure to set the n_trajectories and random_state properties before setting the mode property to sim. See the Examples section below for usage.
Returns:
| Name | Type | Description |
|---|---|---|
mode |
Literal["enum", "sim"] | None:
|
The generation mode of the process, or |
Examples:
>>> from scipy.stats import bernoulli
>>> from sigalg.core import Time
>>> from sigalg.processes import IIDProcess, RandomWalk
Exhaustively enumerate all trajectories of an IID Bernoulli process.
>>> T = Time.discrete(length=2)
>>> X = IIDProcess.generate(
... mode="enum",
... distribution=bernoulli(0.75),
... support=[0, 1],
... index=T,
... )
>>> print(X)
IID process 'X':
t 0 1 2
omega
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
Set the n_trajectories and random_state properties before setting mode to sim. Regenerate the trajectories and print them.
>>> X.n_trajectories = 10
>>> X.random_state = 42
>>> X.mode = "sim"
>>> X.regenerate()
>>> print(X)
IID process 'X':
t 0 1 2
omega
0 0 1 0
1 1 1 0
2 0 0 1
3 1 1 0
4 1 0 1
5 1 1 1
6 0 1 0
7 1 0 0
8 0 1 1
9 1 1 1
Now, we go the other way. First simulate trajectories of a random walk, and then switch the generation mode to enumeration.
>>> Y = RandomWalk.generate(
... mode="sim",
... p=0.7,
... initial_state=0,
... n_trajectories=10,
... index=T,
... random_state=42,
... name="Y",
... )
>>> print(Y)
Random walk 'Y':
t 0 1 2
omega
0 0 -1 0
1 0 -1 0
2 0 1 0
3 0 -1 -2
4 0 1 2
5 0 1 0
6 0 1 0
7 0 1 2
8 0 1 2
9 0 -1 0
>>> Y.mode = "enum"
>>> Y.regenerate()
>>> print(Y)
Random walk 'Y':
t 0 1 2
omega
0 0 -1 -2
1 0 -1 0
2 0 1 0
3 0 1 2
property
writable
n_trajectories
Get the number of trajectories in the stochastic process.
The n_trajectories property is settable. See the Examples section below for usage.
Returns:
| Name | Type | Description |
|---|---|---|
n_trajectories |
int | None
|
The number of trajectories in the stochastic process, or |
Examples:
>>> from scipy.stats import bernoulli
>>> from sigalg.core import Time
>>> from sigalg.processes import IIDProcess
Simulate ten trajectories of an IID process.
>>> T = Time.discrete(length=2)
>>> X = IIDProcess.generate(
... mode="sim",
... distribution=bernoulli(0.75),
... support=[0, 1],
... n_trajectories=10,
... index=T,
... random_state=42,
... )
>>> print(X)
IID process 'X':
t 0 1 2
omega
0 0 1 0
1 1 1 0
2 0 0 1
3 1 1 0
4 1 0 1
5 1 1 1
6 0 1 0
7 1 0 0
8 0 1 1
9 1 1 1
Set the number of trajectories to 5 and regenerate the process.
>>> X.n_trajectories = 5
>>> X.regenerate()
>>> print(X)
IID process 'X':
t 0 1 2
omega
0 1 0 1
1 1 1 1
2 1 1 1
3 1 1 0
4 1 1 0
cached
property
natural_filtration
Get the natural filtration of the stochastic process.
See the Notes section below for the mathematical details.
Returns:
| Name | Type | Description |
|---|---|---|
natural_filtration |
Filtration | None
|
The natural filtration of the stochastic process. |
Examples:
>>> from scipy.stats import bernoulli
>>> from sigalg.processes import IIDProcess
Exhaustively enumerate all length-2 trajectories of an IID Bernoulli process.
>>> X = IIDProcess.generate(
... mode="enum",
... distribution=bernoulli(p=0.5),
... support=[0, 1],
... length=2,
... )
>>> print(X)
IID process 'X':
t 0 1 2
omega
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 its natural filtration. Notice that the atom identifiers of the sigma-algebras are partial trajectories of the process.
>>> print(X.natural_filtration)
Filtration 'F'
==============
* Time 'I':
t
0
1
2
* At index 0:
Sigma algebra 'F_0':
F_0
omega
0 0
1 0
2 0
3 0
4 1
5 1
6 1
7 1
* At index 1:
Sigma algebra 'F_1':
i 0 1
omega
0 0 0
1 0 0
2 0 1
3 0 1
4 1 0
5 1 0
6 1 1
7 1 1
* At index 2:
Sigma algebra 'F_2':
i 0 1 2
omega
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
Notes
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\).
property
writable
random_state
Get the random state of the stochastic process.
The random_state property is settable. See the Examples section below for usage.
Returns:
| Name | Type | Description |
|---|---|---|
random_state |
int | Generator | None
|
The random state of the stochstic process. |
Examples:
>>> from scipy.stats import bernoulli
>>> from sigalg.core import Time
>>> from sigalg.processes import IIDProcess
Simulate ten trajectories of an IID process with a specified random state.
>>> T = Time.discrete(length=2)
>>> X = IIDProcess.generate(
... mode="sim",
... distribution=bernoulli(0.75),
... support=[0, 1],
... n_trajectories=10,
... index=T,
... random_state=42,
... )
>>> print(X)
IID process 'X':
t 0 1 2
omega
0 0 1 0
1 1 1 0
2 0 0 1
3 1 1 0
4 1 0 1
5 1 1 1
6 0 1 0
7 1 0 0
8 0 1 1
9 1 1 1
Set the random state to a different value and regenerate the process.
>>> X.random_state = 101
>>> X.regenerate()
>>> print(X)
IID process 'X':
t 0 1 2
omega
0 0 1 0
1 1 1 0
2 0 1 0
3 1 0 1
4 1 1 0
5 1 1 1
6 1 1 1
7 1 1 1
8 0 0 1
9 1 0 0
property
writable
time
Get the time index of the process.
This property is an alias for the index property of the superclass RandomVector.
The time property is settable. See the Examples section below for usage.
Returns:
| Name | Type | Description |
|---|---|---|
time |
Time | None
|
The time index of the stochastic process. |
Examples:
>>> from scipy.stats import bernoulli
>>> from sigalg.core import Time
>>> from sigalg.processes import IIDProcess
Generate trajectories of an IID process with an explicit time index.
>>> T = Time.discrete(length=2)
>>> X = IIDProcess.generate(
... mode="sim",
... distribution=bernoulli(0.75),
... support=[0, 1],
... n_trajectories=10,
... index=T,
... random_state=42,
... )
>>> print(X)
IID process 'X':
t 0 1 2
omega
0 0 1 0
1 1 1 0
2 0 0 1
3 1 1 0
4 1 0 1
5 1 1 1
6 0 1 0
7 1 0 0
8 0 1 1
9 1 1 1
Set the time index and regenerate the trajectories.
>>> X.time = Time.discrete(length=3)
>>> X.regenerate()
>>> print(X)
IID process 'X':
t 0 1 2 3
omega
0 1 0 1 1
1 1 1 1 1
2 1 1 1 0
3 1 1 0 0
4 1 1 1 1
5 1 1 0 1
6 1 0 1 1
7 1 1 1 1
8 1 0 1 1
9 1 1 1 1
generate(
*args,
mode="sim",
n_trajectories=None,
domain_name=None,
index=None,
length=None,
name="X",
random_state=None,
)
Generate trajectories of the stochastic process by either exhaustive enumeration or Monte Carlo simulation.
This is an abstract method that must be implemented in subclasses. It should validate any subclass-specific parameters, pack them into a dictionary, and return the dictionary. Provided that the user decorates with the @generator decorator, the base class will handle the rest of the generation logic. The implementation of the generate method in the subclass IIDProcess is a good template to review to help understand the needed steps.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
mode
|
Literal['enum', 'sim']
|
The generation mode of the process, either |
"sim"
|
n_trajectories
|
int | None
|
The number of trajectories to simulate. If |
None
|
domain_name
|
Hashable | None
|
The name of the domain. If |
None
|
index
|
IndexLike | None
|
The index of the stochastic process. One of |
None
|
length
|
int | None
|
The length of the trajectories of the stochastic process. One of |
None
|
name
|
Hashable
|
The name of the stochastic process. |
"X"
|
random_state
|
int | Generator | None
|
An optional random state for reproducibility. |
None
|
Returns:
| Name | Type | Description |
|---|---|---|
result |
dict[str, object]
|
A dictionary containing the names of the subclass-specific parameters and their values. |
staticmethod
generator(func)
Decorator that handles common setup and execution for generate methods.
is_adapted(filtration)
Check if the stochastic process is adapted to a given filtration.
See the Notes section below for the mathematical details.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
filtration
|
Filtration
|
The filtration to check adaptation against. |
required |
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
Generate a random walk of length 2.
>>> T = Time.discrete(start=0, stop=2)
>>> X = RandomWalk.generate(mode="enum", p=0.7, initial_state=0, index=T)
>>> print(X)
Random walk 'X':
t 0 1 2
omega
0 0 -1 -2
1 0 -1 0
2 0 1 0
3 0 1 2
Define a stochastic process Y for which each Y[t] is a function of X[0], X[1], ... , X[t].
>>> def f0(X: StochasticProcess) -> RandomVariable:
... return X[0] + 1
>>> 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':
t 0 1 2
omega
0 1 -1 -1
1 1 -1 1
2 1 1 -1
3 1 1 1
Check that Y is adapted to the natural filtration of X.
>>> print(Y.is_adapted(filtration=X.natural_filtration))
True
Notes
Let \(X_t\) be a stochastic process with index set \(T\) on a probability space \((\Omega, \mathcal{F}, P)\) and let \(\mathcal{F}_t\) be a filtration of \(\mathcal{F}\). The process \(X_t\) is said to be adapted to the filtration \(\mathcal{F}_t\) if for every \(t \in T\), the random variable \(X_t\) is \(\mathcal{F}_t\)-measurable.
is_martingale(
filtration=None,
prob_measure=None,
kind="plain",
rtol=1e-05,
atol=1e-08,
)
Check if the stochastic process is a martingale with respect to an optional filtration.
Beware that the check is computationally intensive, as it requires calculating conditional expectations at each time step.
See the Notes section below for the mathematical details.
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
|
prob_measure
|
ProbabilityMeasure | None
|
The probability measure with respect to which the martingale property is checked. If |
None
|
rtol
|
float
|
The relative tolerance parameter for numerical comparison. Internally passed to |
1e-05
|
atol
|
float
|
The absolute tolerance parameter for numerical comparison. Internally passed to |
1e-08
|
Returns:
| Name | Type | Description |
|---|---|---|
is_martingale |
bool
|
|
Examples:
>>> from sigalg.core import Time
>>> from sigalg.processes import RandomWalk
A symmetric random walk (i.e., with p=0.5) is a martingale.
>>> T = Time.discrete(start=1, length=2)
>>> X = RandomWalk.generate(mode="enum", p=0.5, initial_state=0, index=T)
>>> print(X)
Random walk 'X':
t 1 2 3
omega
0 0 -1 -2
1 0 -1 0
2 0 1 0
3 0 1 2
>>> print(X.is_martingale())
True
A random walk with p > 0.5 is a submartingale
>>> X.p = 0.7
>>> X.regenerate()
>>> print(X.is_martingale(kind="sub"))
True
A random walk with p < 0.5 is a supermartingale.
>>> X.p = 0.2
>>> X.regenerate()
>>> print(X.is_martingale(kind="super"))
True
Notes
Let \(X_t\) be a stochastic process with index set \(T\) on a probability space \((\Omega, \mathcal{F}, P)\) and let \(\mathcal{F}_t\) be a filtration of \(\mathcal{F}\). The process \(X_t\) is said to be a martingale with respect to the filtration \(\mathcal{F}_t\) if it satisfies the following conditions: (1) It is adapted to the filtration, meaning that \(X_t\) is \(\mathcal{F}_t\)-measurable for all \(t \in T\), and (2) it satisfies the martingale property
for all \(t\in T\) for which \(t+1 \in T\).
plot_trajectories(
ax=None,
colors=None,
plot_kwargs=None,
x_label="time",
y_label=None,
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
|
colors
|
list
|
A list of colors to use for the trajectories. If |
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 | None
|
Label for the y-axis. If |
None
|
title
|
str
|
Title of the plot. If |
None
|
Returns:
| Name | Type | Description |
|---|---|---|
ax |
Axes
|
The matplotlib Axes object with the plot. |
regenerate()
Regenerate the trajectories of the stochastic process using the current parameters.
Examples:
>>> from sigalg.core import Time
>>> from sigalg.processes import RandomWalk
Generate a symmetric random walk (i.e., with p=0.5) and print its trajectories and probability measure.
>>> T = Time.discrete(start=1, length=2)
>>> X = RandomWalk.generate(mode="enum", p=0.5, initial_state=0, index=T)
>>> print(X)
Random walk 'X':
t 1 2 3
omega
0 0 -1 -2
1 0 -1 0
2 0 1 0
3 0 1 2
>>> print(X.measure)
Probability measure 'P':
P
omega
0 0.25
1 0.25
2 0.25
3 0.25
Set the p parameter to 0.7, regenerate the trajectories, and print the new probability measure.
>>> X.p = 0.7
>>> X.regenerate()
>>> print(X.measure)
Probability measure 'P':
P
omega
0 0.09
1 0.21
2 0.21
3 0.49
Bases: RandomVariable
A class representing a stopping time.
The constructor is not meant to be used directly. Instead, the user should call the from_filtration class method.
See the Notes section below for the mathematical details.
Examples:
>>> from math import inf
>>> from sigalg.core import Time
>>> from sigalg.processes import RandomWalk, StoppingTime
A gambler begins with 10 units of currency in the bank and plays a game with unit stakes. The gambler has probability 0.4 of winning one unit, and probability 0.6 of losing one unit. We suppose that the money in the gambler's bank is modeled by a random walk, and that the gambler plays ten games.
In SigAlg, we set up this random walk as follows, and generate eight random trajectories:
>>> T = Time.discrete(start=1, stop=10)
>>> S = RandomWalk.generate(
... mode="sim",
... p=0.4,
... initial_state=10,
... index=T,
... n_trajectories=8,
... random_state=42,
... name="S",
... )
>>> print(S)
Random walk 'S':
t 1 2 3 4 5 6 7 8 9 10
omega
0 10 11 10 11 12 11 12 13 14 13
1 10 9 8 9 10 11 10 9 8 7
2 10 11 12 13 12 13 14 15 14 13
3 10 9 8 9 10 11 10 9 8 7
4 10 9 8 7 8 7 8 9 8 9
5 10 11 10 9 10 9 8 7 8 9
6 10 11 12 11 10 9 8 9 8 7
7 10 11 12 11 10 9 8 7 6 5
The gambler decides that they will stop playing the game when their holdings equal 8 units, or they suffer a 20% loss compared to their initial holdings of 10 units. So, using the printout of S above, we see that price trajectory 0 will never stop, price trajectory 1 should stop at time 3, price trajectory 2 will never stop, price trajectory 3 should stop at time 3, and so on. These values define a stopping time, which we implement as follows:
>>> tau = StoppingTime.from_filtration(
... process=S,
... mapping={
... 0: inf, # play will never stop
... 1: 3,
... 2: inf, # play will never stop
... 3: 3,
... 4: 3,
... 5: 7,
... 6: 7,
... 7: 7,
... },
... )
>>> print(tau)
Stopping time 'tau':
tau
omega
0 inf
1 3.0
2 inf
3 3.0
4 3.0
5 7.0
6 7.0
7 7.0
Notes
Let \((\Omega, \mathcal{F}, P)\) be a probability space and \(\{\mathcal{F}_t\}_{t\in T}\) a filtration of \(\mathcal{F}\), index by a linearly ordered set \(T\). A random variable \(\tau: \Omega \to T\) is called a stopping time if, for each \(t\in T\), we have
classmethod
from_filtration(
process=None, filtration=None, mapping=None, name="tau"
)
Pass.