Compound Poisson

quantflow.sp.poisson.CompoundPoissonProcess pydantic-model

Bases: PoissonBase, Generic[D]

A generic Compound Poisson process.

Fields:

• intensity (float)
• jumps (D)

intensity pydantic-field

intensity = 1.0

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

jumps pydantic-field

jumps

Jump size distribution

characteristic_exponent

characteristic_exponent(t, u)

The characteristic exponent of the Compound Poisson process, given by

\[\begin{equation} \phi_{x_t,u} = t\lambda \left(1 - \Phi_{j,u}\right) \end{equation}\]

where \(\Phi_{j,u}\) is the characteristic function of the jump distribution

Source code in quantflow/sp/poisson.py

def characteristic_exponent(self, t: FloatArrayLike, u: Vector) -> Vector:
    r"""The characteristic exponent of the Compound Poisson process,
    given by

    \begin{equation}
        \phi_{x_t,u} = t\lambda \left(1 - \Phi_{j,u}\right)
    \end{equation}

    where $\Phi_{j,u}$ is the characteristic function
    of the jump distribution
    """
    return t * self.intensity * (1 - self.jumps.characteristic(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

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)

Sample jump sizes from the jump distribution

Source code in quantflow/sp/poisson.py

def sample_jumps(self, n: int) -> FloatArray:
    """Sample jump sizes from the jump distribution"""
    return self.jumps.sample(n)

analytical_mean

analytical_mean(t)

Expected value at a time horizon

Source code in quantflow/sp/poisson.py

def analytical_mean(self, t: FloatArrayLike) -> FloatArrayLike:
    """Expected value at a time horizon"""
    return self.intensity * t * self.jumps.mean()

analytical_variance

analytical_variance(t)

Expected variance at a time horizon

Source code in quantflow/sp/poisson.py

def analytical_variance(self, t: FloatArrayLike) -> FloatArrayLike:
    """Expected variance at a time horizon"""
    return self.intensity * t * (self.jumps.variance() + self.jumps.mean() ** 2)

create classmethod

create(jump_distribution, *, vol=0.5, jump_intensity=100, jump_asymmetry=0.0)

Create a Compound Poisson process with a given jump distribution, volatility, jump intensity and jump asymmetry.

PARAMETER	DESCRIPTION
jump_distribution	The distribution of jump size (currently only Normal and DoubleExponential are supported) TYPE: type[D]
vol	Annualized standard deviation TYPE: float DEFAULT: 0.5
jump_intensity	The average number of jumps per year TYPE: float DEFAULT: 100
jump_asymmetry	The asymmetry of the jump distribution (0 for symmetric, only used by distributions with asymmetry) TYPE: float DEFAULT: 0.0

Source code in quantflow/sp/poisson.py

@classmethod
def create(
    cls,
    jump_distribution: Annotated[
        type[D],
        Doc(
            "The distribution of jump size (currently only"
            " [Normal][quantflow.utils.distributions.Normal] and"
            " [DoubleExponential][quantflow.utils.distributions.DoubleExponential]"
            " are supported)"
        ),
    ],
    *,
    vol: Annotated[float, Doc("Annualized standard deviation")] = 0.5,
    jump_intensity: Annotated[
        float, Doc("The average number of jumps per year")
    ] = 100,
    jump_asymmetry: Annotated[
        float,
        Doc(
            "The asymmetry of the jump distribution (0 for symmetric,"
            " only used by distributions with asymmetry)"
        ),
    ] = 0.0,
) -> CompoundPoissonProcess[D]:
    """Create a Compound Poisson process with a given jump distribution, volatility,
    jump intensity and jump asymmetry.
    """
    variance = vol * vol
    jump_distribution_variance = variance / jump_intensity
    jumps = jump_distribution.from_variance_and_asymmetry(
        jump_distribution_variance, jump_asymmetry
    )
    return cls(intensity=jump_intensity, jumps=jumps)

sample_from_draws

sample_from_draws(draws, *args)

Source code in quantflow/sp/poisson.py

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

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

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

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

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_pdf

analytical_pdf(t, x)

Analytical pdf of the process at time t

Implement if available

Source code in quantflow/sp/base.py

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

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

def marginal(self, t: FloatArrayLike) -> StochasticProcess1DMarginal:
    """Marginal distribution of the process at time `t`"""
    return StochasticProcess1DMarginal(process=self, t=t)

domain_range

domain_range()

Source code in quantflow/sp/poisson.py

def domain_range(self) -> Bounds:
    return Bounds(0, np.inf)

frequency_range

frequency_range(std, max_frequency=None)

Maximum frequency when calculating characteristic functions

Source code in quantflow/sp/base.py

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

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)