Skip to content

Doubly Stochastic Poisson process

quantflow.sp.dsp.DSP pydantic-model

Bases: PoissonBase

Doubly Stochastic Poisson process.

It's a process where the inter-arrival time is exponentially distributed with rate \(\lambda_t\)

Fields:

intensity pydantic-field 

intensity

intensity process

poisson pydantic-field 

poisson

marginal 

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

frequency_range 

frequency_range(std, max_frequency=None)

Frequency range of the process

Source code in quantflow/sp/dsp.py 
29
30
31
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)
Source code in quantflow/sp/dsp.py 
33
34
def support(self, mean: float, std: float, points: int) -> FloatArray:
    return np.linspace(0, points, points + 1)

characteristic_exponent 

characteristic_exponent(t, u)
Source code in quantflow/sp/dsp.py 
36
37
38
39
40
def characteristic_exponent(self, t: FloatArrayLike, u: Vector) -> Vector:
    # phi_x is the per-unit-time Lévy exponent of the inner Poisson;
    # the integrated log-Laplace already absorbs the time horizon
    phi_x = self.poisson.characteristic_exponent(1, u)
    return -self.intensity.integrated_log_laplace(t, phi_x)

arrivals 

arrivals(t=1)
Source code in quantflow/sp/dsp.py 
42
43
44
45
def arrivals(self, t: float = 1) -> FloatArray:
    paths = self.intensity.sample(1, t, math.ceil(100 * t)).integrate()
    intensity = paths.data[-1, 0]
    return poisson_arrivals(intensity, t)

sample_jumps 

sample_jumps(n)
Source code in quantflow/sp/dsp.py 
47
48
def sample_jumps(self, n: int) -> FloatArray:
    return self.poisson.sample_jumps(n)

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))

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

domain_range 

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