Skip to content

Poisson process

quantflow.sp.poisson.PoissonBase pydantic-model

Bases: StochasticProcess1D

sample_jumps abstractmethod 

sample_jumps(n)

Generate a list of jump sizes

PARAMETER DESCRIPTION
n

Number of jumps

TYPE: int

Source code in quantflow/sp/poisson.py 
25
26
27
@abstractmethod
def sample_jumps(self, n: Annotated[int, Doc("Number of jumps")]) -> np.ndarray:
    """Generate a list of jump sizes"""

arrivals abstractmethod 

arrivals(time_horizon=1)

Generate a list of jump arrivals times up to time time_horizon

PARAMETER DESCRIPTION
time_horizon

Time horizon

TYPE: float DEFAULT: 1

Source code in quantflow/sp/poisson.py 
29
30
31
32
33
@abstractmethod
def arrivals(
    self, time_horizon: Annotated[float, Doc("Time horizon")] = 1
) -> FloatArray:
    """Generate a list of jump arrivals times up to time `time_horizon`"""

sample 

sample(n, time_horizon=1, time_steps=100)

Sample a number of paths of the process up to a given time horizon and with a given number of time steps.

PARAMETER DESCRIPTION
n

Number of paths

TYPE: int

time_horizon

Time horizon

TYPE: float DEFAULT: 1

time_steps

Number of time steps

TYPE: int DEFAULT: 100

Source code in quantflow/sp/poisson.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def sample(
    self,
    n: Annotated[int, Doc("Number of paths")],
    time_horizon: Annotated[float, Doc("Time horizon")] = 1,
    time_steps: Annotated[int, Doc("Number of time steps")] = 100,
) -> Paths:
    """Sample a number of paths of the process up to a given time horizon and
    with a given number of time steps.
    """
    dt = time_horizon / time_steps
    paths = np.zeros((time_steps + 1, n))
    for p in range(n):
        arrivals = self.arrivals(time_horizon)
        if num_arrivals := len(arrivals):
            jumps = self.sample_jumps(num_arrivals)
            i = 1
            y = 0.0
            for j, arrival in enumerate(arrivals):
                while i <= time_steps and i * dt < arrival:
                    paths[i, p] = y
                    i += 1
                y += jumps[j]
            paths[i:, p] = y
    return Paths(t=time_horizon, data=paths)

sample_from_draws 

sample_from_draws(draws, *args)
Source code in quantflow/sp/poisson.py 
60
61
def sample_from_draws(self, draws: Paths, *args: Paths) -> Paths:
    raise NotImplementedError

domain_range 

domain_range()
Source code in quantflow/sp/poisson.py 
63
64
def domain_range(self) -> Bounds:
    return Bounds(0, np.inf)

characteristic_exponent abstractmethod 

characteristic_exponent(t, u)

Characteristic exponent at time t for a given input parameter

Source code in quantflow/sp/base.py 
41
42
43
@abstractmethod
def characteristic_exponent(self, t: FloatArrayLike, u: Vector) -> Vector:
    """Characteristic exponent at time `t` for a given input parameter"""

characteristic 

characteristic(t, u)

Characteristic function at time t for a given input parameter u

The characteristic function represents the Fourier transform of the probability density function

\[\begin{equation} \Phi = {\mathbb E} \left[e^{i u x_t}\right] = e^{-\phi(t, u)} \end{equation}\]

where \(\phi\) is the characteristic exponent, which can be more easily computed for many processes.

PARAMETER DESCRIPTION
t

Time horizon

TYPE: FloatArrayLike

u

Characteristic function input parameter

TYPE: Vector

Source code in quantflow/sp/base.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def characteristic(
    self,
    t: Annotated[FloatArrayLike, Doc("Time horizon")],
    u: Annotated[Vector, Doc("Characteristic function input parameter")],
) -> Vector:
    r"""Characteristic function at time `t` for a given input parameter `u`

    The characteristic function represents the Fourier transform of the
    probability density function

    \begin{equation}
        \Phi = {\mathbb E} \left[e^{i u x_t}\right] = e^{-\phi(t, u)}
    \end{equation}

    where $\phi$ is the characteristic exponent, which can be more easily
    computed for many processes.
    """
    return np.exp(-self.characteristic_exponent(t, u))

convexity_correction 

convexity_correction(t)

Convexity correction for the process

Source code in quantflow/sp/base.py 
64
65
66
def convexity_correction(self, t: FloatArrayLike) -> Vector:
    """Convexity correction for the process"""
    return -self.characteristic_exponent(t, complex(0, -1)).real

analytical_std 

analytical_std(t)

Analytical standard deviation of the process at time t

This has a closed form solution if the process has an analytical variance

Source code in quantflow/sp/base.py 
68
69
70
71
72
73
def analytical_std(self, t: FloatArrayLike) -> FloatArrayLike:
    """Analytical standard deviation of the process at time `t`

    This has a closed form solution if the process has an analytical variance
    """
    return np.sqrt(self.analytical_variance(t))

analytical_mean 

analytical_mean(t)

Analytical mean of the process at time t

Implement if available

Source code in quantflow/sp/base.py 
75
76
77
78
79
80
def analytical_mean(self, t: FloatArrayLike) -> FloatArrayLike:
    """Analytical mean of the process at time `t`

    Implement if available
    """
    raise NotImplementedError

analytical_variance 

analytical_variance(t)

Analytical variance of the process at time t

Implement if available

Source code in quantflow/sp/base.py 
82
83
84
85
86
87
def analytical_variance(self, t: FloatArrayLike) -> FloatArrayLike:
    """Analytical variance of the process at time `t`

    Implement if available
    """
    raise NotImplementedError

analytical_pdf 

analytical_pdf(t, x)

Analytical pdf of the process at time t

Implement if available

Source code in quantflow/sp/base.py 
89
90
91
92
93
94
def analytical_pdf(self, t: FloatArrayLike, x: FloatArrayLike) -> FloatArrayLike:
    """Analytical pdf of the process at time `t`

    Implement if available
    """
    raise NotImplementedError

analytical_cdf 

analytical_cdf(t, x)

Analytical cdf of the process at time t

Implement if available

Source code in quantflow/sp/base.py 
 96
 97
 98
 99
100
101
def analytical_cdf(self, t: FloatArrayLike, x: FloatArrayLike) -> FloatArrayLike:
    """Analytical cdf of the process at time `t`

    Implement if available
    """
    raise NotImplementedError

marginal 

marginal(t)

Marginal distribution of the process at time t

Source code in quantflow/sp/base.py 
109
110
111
def marginal(self, t: FloatArrayLike) -> StochasticProcess1DMarginal:
    """Marginal distribution of the process at time `t`"""
    return StochasticProcess1DMarginal(process=self, t=t)

frequency_range 

frequency_range(std, max_frequency=None)

Maximum frequency when calculating characteristic functions

Source code in quantflow/sp/base.py 
116
117
118
119
120
def frequency_range(self, std: float, max_frequency: float | None = None) -> Bounds:
    """Maximum frequency when calculating characteristic functions"""
    if max_frequency is None:
        max_frequency = np.sqrt(40 / std / std)
    return Bounds(0, max_frequency)

support 

support(mean, std, points)

Support of the process at time t

Source code in quantflow/sp/base.py 
122
123
124
125
126
127
def support(self, mean: float, std: float, points: int) -> FloatArray:
    """Support of the process at time `t`"""
    bounds = self.domain_range()
    start = float(sigfig(bound_from_any(bounds.lb, mean - std)))
    end = float(sigfig(bound_from_any(bounds.ub, mean + std)))
    return np.linspace(start, end, points + 1)

quantflow.sp.poisson.PoissonProcess pydantic-model

Bases: PoissonBase

A Poisson process is a pure jump process where the number of jumps in a time interval follows a Poisson distribution and the jump sizes are always 1.

The expected number of jumps and the variance in a unit of time is given by the non-negative intensity parameter \(\lambda\).

Fields:

intensity pydantic-field 

intensity = 1.0

Intensity rate \(\lambda\) of the Poisson process

marginal 

marginal(t)
Source code in quantflow/sp/poisson.py 
95
96
def marginal(self, t: FloatArrayLike) -> StochasticProcess1DMarginal:
    return MarginalDiscrete1D(process=self, t=t)

characteristic_exponent 

characteristic_exponent(t, u)
Source code in quantflow/sp/poisson.py 
98
99
def characteristic_exponent(self, t: Vector, u: Vector) -> Vector:
    return t * self.intensity * (1 - np.exp(Im * u))

arrivals 

arrivals(time_horizon=1)

Generate a list of jump arrivals times up to time time_horizon

PARAMETER DESCRIPTION
time_horizon

Time horizon

TYPE: float DEFAULT: 1

Source code in quantflow/sp/poisson.py 
101
102
103
104
105
def arrivals(
    self, time_horizon: Annotated[float, Doc("Time horizon")] = 1
) -> FloatArray:
    """Generate a list of jump arrivals times up to time `time_horizon`"""
    return poisson_arrivals(self.intensity, time_horizon)

sample_jumps 

sample_jumps(n)

For a poisson process this is just a list of 1s

Source code in quantflow/sp/poisson.py 
107
108
109
def sample_jumps(self, n: int) -> np.ndarray:
    """For a poisson process this is just a list of 1s"""
    return np.ones((n,))

frequency_range 

frequency_range(std, max_frequency=None)

Frequency range of the process

Source code in quantflow/sp/poisson.py 
111
112
113
def frequency_range(self, std: float, max_frequency: float | None = None) -> Bounds:
    """Frequency range of the process"""
    return Bounds(0, np.pi)

support 

support(mean, std, points)

Support of the process at time t

Source code in quantflow/sp/poisson.py 
115
116
117
def support(self, mean: float, std: float, points: int) -> FloatArray:
    """Support of the process at time `t`"""
    return np.linspace(0, points, points + 1)

analytical_mean 

analytical_mean(t)

Expected value at a time horizon

Source code in quantflow/sp/poisson.py 
119
120
121
def analytical_mean(self, t: FloatArrayLike) -> FloatArrayLike:
    """Expected value at a time horizon"""
    return self.intensity * t

analytical_variance 

analytical_variance(t)

Expected variance at a time horizon

Source code in quantflow/sp/poisson.py 
123
124
125
def analytical_variance(self, t: FloatArrayLike) -> FloatArrayLike:
    """Expected variance at a time horizon"""
    return self.intensity * t

analytical_cdf 

analytical_cdf(t, n)

CDF of the number of events at time t.

It's given by

\[\begin{equation} F\left(n\right)=\frac{\Gamma\left(\left\lfloor n+1\right\rfloor ,\lambda\right)}{\left\lfloor n\right\rfloor !} \tag{1} \end{equation}\]

where \(\Gamma\) is the upper incomplete gamma function.

Source code in quantflow/sp/poisson.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
def analytical_cdf(self, t: FloatArrayLike, n: FloatArrayLike) -> FloatArrayLike:
    r"""
    CDF of the number of events at time ``t``.

    It's given by

    \begin{equation}
        F\left(n\right)=\frac{\Gamma\left(\left\lfloor n+1\right\rfloor
        ,\lambda\right)}{\left\lfloor n\right\rfloor !}
        \tag{1}
    \end{equation}

    where $\Gamma$ is the upper incomplete gamma function.
    """
    return poisson.cdf(n, t * self.intensity)

analytical_pdf 

analytical_pdf(t, n)

Probability density function of the number of events at time t.

It's given by

\[\begin{equation} f\left(n\right)=\frac{(\lambda t)^{n}e^{-\lambda t}}{n!} \tag{2} \end{equation}\]
Source code in quantflow/sp/poisson.py 
143
144
145
146
147
148
149
150
151
152
153
154
def analytical_pdf(self, t: FloatArrayLike, n: FloatArrayLike) -> FloatArrayLike:
    r"""
    Probability density function of the number of events at time ``t``.

    It's given by

    \begin{equation}
        f\left(n\right)=\frac{(\lambda t)^{n}e^{-\lambda t}}{n!}
        \tag{2}
    \end{equation}
    """
    return poisson.pmf(n, t * self.intensity)

cdf_jacobian 

cdf_jacobian(t, n)

Jacobian of the CDF

It's given by

\[\begin{equation} \frac{\partial F}{\partial\lambda}=-\frac{(\lambda t)^{\left\lfloor n\right\rfloor }e^{-\lambda}}{\left\lfloor n\right\rfloor !} \tag{3} \end{equation}\]
Source code in quantflow/sp/poisson.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
def cdf_jacobian(self, t: FloatArrayLike, n: Vector) -> np.ndarray:
    r"""
    Jacobian of the CDF

    It's given by

    \begin{equation}
        \frac{\partial F}{\partial\lambda}=-\frac{(\lambda t)^{\left\lfloor
        n\right\rfloor }e^{-\lambda}}{\left\lfloor n\right\rfloor !}
        \tag{3}
    \end{equation}
    """
    k = np.floor(n).astype(int)
    rate = self.intensity
    return np.array([-(rate**k) * np.exp(-rate)]) / factorial(k)

sample_from_draws 

sample_from_draws(draws, *args)
Source code in quantflow/sp/poisson.py 
60
61
def sample_from_draws(self, draws: Paths, *args: Paths) -> Paths:
    raise NotImplementedError

sample 

sample(n, time_horizon=1, time_steps=100)

Sample a number of paths of the process up to a given time horizon and with a given number of time steps.

PARAMETER DESCRIPTION
n

Number of paths

TYPE: int

time_horizon

Time horizon

TYPE: float DEFAULT: 1

time_steps

Number of time steps

TYPE: int DEFAULT: 100

Source code in quantflow/sp/poisson.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def sample(
    self,
    n: Annotated[int, Doc("Number of paths")],
    time_horizon: Annotated[float, Doc("Time horizon")] = 1,
    time_steps: Annotated[int, Doc("Number of time steps")] = 100,
) -> Paths:
    """Sample a number of paths of the process up to a given time horizon and
    with a given number of time steps.
    """
    dt = time_horizon / time_steps
    paths = np.zeros((time_steps + 1, n))
    for p in range(n):
        arrivals = self.arrivals(time_horizon)
        if num_arrivals := len(arrivals):
            jumps = self.sample_jumps(num_arrivals)
            i = 1
            y = 0.0
            for j, arrival in enumerate(arrivals):
                while i <= time_steps and i * dt < arrival:
                    paths[i, p] = y
                    i += 1
                y += jumps[j]
            paths[i:, p] = y
    return Paths(t=time_horizon, data=paths)

characteristic 

characteristic(t, u)

Characteristic function at time t for a given input parameter u

The characteristic function represents the Fourier transform of the probability density function

\[\begin{equation} \Phi = {\mathbb E} \left[e^{i u x_t}\right] = e^{-\phi(t, u)} \end{equation}\]

where \(\phi\) is the characteristic exponent, which can be more easily computed for many processes.

PARAMETER DESCRIPTION
t

Time horizon

TYPE: FloatArrayLike

u

Characteristic function input parameter

TYPE: Vector

Source code in quantflow/sp/base.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def characteristic(
    self,
    t: Annotated[FloatArrayLike, Doc("Time horizon")],
    u: Annotated[Vector, Doc("Characteristic function input parameter")],
) -> Vector:
    r"""Characteristic function at time `t` for a given input parameter `u`

    The characteristic function represents the Fourier transform of the
    probability density function

    \begin{equation}
        \Phi = {\mathbb E} \left[e^{i u x_t}\right] = e^{-\phi(t, u)}
    \end{equation}

    where $\phi$ is the characteristic exponent, which can be more easily
    computed for many processes.
    """
    return np.exp(-self.characteristic_exponent(t, u))

convexity_correction 

convexity_correction(t)

Convexity correction for the process

Source code in quantflow/sp/base.py 
64
65
66
def convexity_correction(self, t: FloatArrayLike) -> Vector:
    """Convexity correction for the process"""
    return -self.characteristic_exponent(t, complex(0, -1)).real

analytical_std 

analytical_std(t)

Analytical standard deviation of the process at time t

This has a closed form solution if the process has an analytical variance

Source code in quantflow/sp/base.py 
68
69
70
71
72
73
def analytical_std(self, t: FloatArrayLike) -> FloatArrayLike:
    """Analytical standard deviation of the process at time `t`

    This has a closed form solution if the process has an analytical variance
    """
    return np.sqrt(self.analytical_variance(t))

domain_range 

domain_range()
Source code in quantflow/sp/poisson.py 
63
64
def domain_range(self) -> Bounds:
    return Bounds(0, np.inf)