Skip to content

Jump diffusions

Jump-diffusions models are a class of stochastic processes that combine a diffusion process with a jump process. The jump process is a Poisson process that generates jumps in the value of the underlying asset. The jump-diffusion model is a generalization of the Black-Scholes model that allows for the possibility of large, discontinuous jumps in the value of the underlying asset.

The most famous jump-diffusion model is the Merton model, which was introduced by Robert Merton in 1976. The Merton model assumes that the underlying asset follows a geometric Brownian motion with jumps that are normally distributed.

quantflow.sp.jump_diffusion.JumpDiffusion pydantic-model

Bases: StochasticProcess1D, Generic[D]

A generic jump-diffusion model

\[\begin{equation} dx_t = \sigma d w_t + d N_t \end{equation}\]

where \(w_t\) is a WienerProcess process with standard deviation \(\sigma\) and \(N_t\) is a CompoundPoissonProcess with intensity \(\lambda\) and generic jump distribution \(D\)

Fields:

diffusion pydantic-field

diffusion

diffusion process

jumps pydantic-field

jumps

The jump process

characteristic_exponent

characteristic_exponent(t, u)
Source code in quantflow/sp/jump_diffusion.py
def characteristic_exponent(self, t: FloatArrayLike, u: Vector) -> Vector:
    return self.diffusion.characteristic_exponent(
        t, u
    ) + self.jumps.characteristic_exponent(t, u)

sample

sample(n, time_horizon=1, time_steps=100)
Source code in quantflow/sp/jump_diffusion.py
def sample(self, n: int, time_horizon: float = 1, time_steps: int = 100) -> Paths:
    dw1 = Paths.normal_draws(n, time_horizon, time_steps)
    return self.sample_from_draws(dw1)

sample_from_draws

sample_from_draws(path_w, *args)
Source code in quantflow/sp/jump_diffusion.py
def sample_from_draws(self, path_w: Paths, *args: Paths) -> Paths:
    if args:
        path_j = args[0]
    else:
        path_j = self.jumps.sample(path_w.samples, path_w.t, path_w.time_steps)
    path_w = self.diffusion.sample_from_draws(path_w)
    return Paths(t=path_w.t, data=path_w.data + path_j.data)

analytical_mean

analytical_mean(t)
Source code in quantflow/sp/jump_diffusion.py
def analytical_mean(self, t: FloatArrayLike) -> FloatArrayLike:
    return self.diffusion.analytical_mean(t) + self.jumps.analytical_mean(t)

analytical_variance

analytical_variance(t)
Source code in quantflow/sp/jump_diffusion.py
def analytical_variance(self, t: FloatArrayLike) -> FloatArrayLike:
    return self.diffusion.analytical_variance(t) + self.jumps.analytical_variance(t)

create classmethod

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

Create a jump-diffusion model with a given jump distribution, volatility and jump fraction.

PARAMETER DESCRIPTION
jump_distribution

The distribution of jump sizes. Currently Normal and DoubleExponential are supported. If the jump distribution is set to the Normal distribution, the model reduces to a Merton jump-diffusion.

TYPE: type[D]

vol

total standard deviation per unit time

TYPE: float DEFAULT: 0.5

jump_intensity

The expected number of jumps per unit time

TYPE: float DEFAULT: 100

jump_fraction

The fraction of variance due to jumps (between 0 and 1)

TYPE: float DEFAULT: 0.5

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/jump_diffusion.py
@classmethod
def create(
    cls,
    jump_distribution: Annotated[
        type[D],
        Doc(
            "The distribution of jump sizes. Currently "
            "[Normal][quantflow.utils.distributions.Normal] and "
            "[DoubleExponential][quantflow.utils.distributions.DoubleExponential] "
            "are supported. If the jump distribution is set to the Normal "
            "distribution, the model reduces to a Merton jump-diffusion."
        ),
    ],
    vol: Annotated[float, Doc("total standard deviation per unit time")] = 0.5,
    jump_intensity: Annotated[
        float, Doc("The expected number of jumps per unit time")
    ] = 100,
    jump_fraction: Annotated[
        float, Doc("The fraction of variance due to jumps (between 0 and 1)")
    ] = 0.5,
    jump_asymmetry: Annotated[
        float,
        Doc(
            "The asymmetry of the jump distribution "
            "(0 for symmetric, only used by distributions with asymmetry)"
        ),
    ] = 0.0,
) -> JumpDiffusion[D]:
    """Create a jump-diffusion model with a given jump distribution, volatility
    and jump fraction.
    """
    variance = vol * vol
    if jump_fraction >= 1:
        raise ValueError("jump_fraction must be less than 1")
    elif jump_fraction <= 0:
        raise ValueError("jump_fraction must be greater than 0")
    else:
        jump_variance = variance * jump_fraction
        jump_distribution_variance = jump_variance / jump_intensity
        jumps = jump_distribution.from_variance_and_asymmetry(
            jump_distribution_variance, jump_asymmetry
        )
        return cls(
            diffusion=WienerProcess(sigma=np.sqrt(variance * (1 - jump_fraction))),
            jumps=CompoundPoissonProcess(intensity=jump_intensity, jumps=jumps),
        )

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/base.py
def domain_range(self) -> Bounds:
    return default_bounds()

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)