Heston process

quantflow.sp.heston.Heston pydantic-model

Bases: StochasticProcess1D

The Heston stochastic volatility model

The classical square-root stochastic volatility model of Heston (1993) can be regarded as a standard Brownian motion \(x_t\) time changed by a CIR activity rate process.

\[\begin{align} d x_t &= d w^1_t \\ d v_t &= \kappa (\theta - v_t) dt + \nu \sqrt{v_t} dw^2_t \\ \rho dt &= {\tt E}[dw^1 dw^2] \end{align}\]

Fields:

• variance_process (CIR)
• rho (float)

variance_process pydantic-field

variance_process

The variance process is a Cox-Ingersoll-Ross process which is guaranteed to be positive if the Feller condition is satisfied

rho pydantic-field

rho = 0

Correlation between the Brownian motions, it provides the leverage effect and therefore the skewness of the distribution

create classmethod

create(*, rate=1.0, vol=0.5, kappa=1, sigma=0.8, rho=0, theta=None)

Create an Heston model.

To understand the parameters lets introduce the following notation:

\[\begin{align} {\tt var} &= {\tt vol}^2 \\ v_0 &= {\tt rate}\cdot{\tt var} \end{align}\]

PARAMETER	DESCRIPTION
rate	Initial rate of the variance process TYPE: float DEFAULT: 1.0
vol	Volatility of the price process, normalized by the square root of time, as time tends to infinity (the long term standard deviation) TYPE: float DEFAULT: 0.5
kappa	Mean reversion speed for the variance process, the lower the more pronounced the volatility clustering and therefore the fatter the tails of the distribution of the price process TYPE: float DEFAULT: 1
sigma	Volatility of the variance process (a.k.a. the vol of vol) TYPE: float DEFAULT: 0.8
rho	Correlation between the Brownian motions of the variance and price processes TYPE: float DEFAULT: 0
theta	Long-term mean of the variance process. If None, it defaults to the variance given by \({\tt var}\) the long term variance described above. TYPE: float | None DEFAULT: None

Source code in quantflow/sp/heston.py

@classmethod
def create(
    cls,
    *,
    rate: Annotated[float, Doc("Initial rate of the variance process")] = 1.0,
    vol: Annotated[
        float,
        Doc(
            "Volatility of the price process, normalized by the "
            "square root of time, as time tends to infinity "
            "(the long term standard deviation)"
        ),
    ] = 0.5,
    kappa: Annotated[
        float,
        Doc(
            "Mean reversion speed for the variance process, the lower the "
            "more pronounced the volatility clustering and therefore the fatter "
            "the tails of the distribution of the price process"
        ),
    ] = 1,
    sigma: Annotated[
        float, Doc("Volatility of the variance process (a.k.a. the vol of vol)")
    ] = 0.8,
    rho: Annotated[
        float,
        Doc(
            "Correlation between the Brownian motions of the "
            "variance and price processes"
        ),
    ] = 0,
    theta: Annotated[
        float | None,
        Doc(
            "Long-term mean of the variance process. "
            r"If `None`, it defaults to the variance given by ${\tt var}$"
            " the long term variance described above."
        ),
    ] = None,
) -> Self:
    r"""Create an Heston model.

    To understand the parameters lets introduce the following notation:

    \begin{align}
        {\tt var} &= {\tt vol}^2 \\
        v_0 &= {\tt rate}\cdot{\tt var}
    \end{align}
    """
    variance = vol * vol
    return cls(
        variance_process=CIR(
            rate=rate * variance,
            kappa=kappa,
            sigma=sigma,
            theta=theta if theta is not None else variance,
        ),
        rho=rho,
    )

characteristic_exponent

characteristic_exponent(t, u)

Characteristic exponent of the Heston model in closed form.

Define the correlation-adjusted drift and the characteristic frequency:

\[\begin{equation} \tilde{\kappa} = \kappa - i u \nu \rho, \qquad \gamma = \sqrt{\tilde{\kappa}^2 + u^2 \nu^2} \end{equation}\]

Then the exponent is

\[\begin{equation} \phi_{x_t,u} = \frac{\kappa\theta}{\nu^2} \left[2 \ln c_{t,u} + (\gamma - \tilde{\kappa})\,t\right] + v_0\, b_{t,u} \end{equation}\]

where

\[\begin{align*} c_{t,u} &= \frac{\gamma - \tfrac{1}{2}(\gamma - \tilde{\kappa}) (1 - e^{-\gamma t})}{\gamma} \\ b_{t,u} &= \frac{u^2 (1 - e^{-\gamma t})} {(\gamma + \tilde{\kappa}) + (\gamma - \tilde{\kappa})\,e^{-\gamma t}} \end{align*}\]

and \(\nu\) is the vol of vol, \(\kappa\) the mean-reversion speed, \(\theta\) the long-term variance, \(\rho\) the correlation, and \(v_0\) the initial variance.

PARAMETER	DESCRIPTION
t	Time horizon or array of evaluation times TYPE: FloatArrayLike
u	Characteristic exponent argument TYPE: Vector

Source code in quantflow/sp/heston.py

def characteristic_exponent(
    self,
    t: Annotated[FloatArrayLike, Doc("Time horizon or array of evaluation times")],
    u: Annotated[Vector, Doc("Characteristic exponent argument")],
) -> Vector:
    r"""Characteristic exponent of the Heston model in closed form.

    Define the correlation-adjusted drift and the characteristic frequency:

    \begin{equation}
        \tilde{\kappa} = \kappa - i u \nu \rho, \qquad
        \gamma = \sqrt{\tilde{\kappa}^2 + u^2 \nu^2}
    \end{equation}

    Then the exponent is

    \begin{equation}
        \phi_{x_t,u} = \frac{\kappa\theta}{\nu^2}
            \left[2 \ln c_{t,u} + (\gamma - \tilde{\kappa})\,t\right]
            + v_0\, b_{t,u}
    \end{equation}

    where

    \begin{align*}
        c_{t,u} &= \frac{\gamma - \tfrac{1}{2}(\gamma - \tilde{\kappa})
            (1 - e^{-\gamma t})}{\gamma} \\
        b_{t,u} &= \frac{u^2 (1 - e^{-\gamma t})}
            {(\gamma + \tilde{\kappa}) + (\gamma - \tilde{\kappa})\,e^{-\gamma t}}
    \end{align*}

    and $\nu$ is the vol of vol, $\kappa$ the mean-reversion speed,
    $\theta$ the long-term variance, $\rho$ the correlation, and $v_0$
    the initial variance.
    """
    eta = self.variance_process.sigma
    eta2 = eta * eta
    theta_kappa = self.variance_process.theta * self.variance_process.kappa
    # adjusted drift
    kappa = self.variance_process.kappa - 1j * u * eta * self.rho
    u2 = u * u
    gamma = np.sqrt(kappa * kappa + u2 * eta2)
    egt = np.exp(-gamma * t)
    c = (gamma - 0.5 * (gamma - kappa) * (1 - egt)) / gamma
    b = u2 * (1 - egt) / ((gamma + kappa) + (gamma - kappa) * egt)
    a = theta_kappa * (2 * np.log(c) + (gamma - kappa) * t) / eta2
    return a + b * self.variance_process.rate

sample

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

PARAMETER	DESCRIPTION
n	Number of sample paths TYPE: int
time_horizon	Time horizon TYPE: float DEFAULT: 1
time_steps	Number of discrete time steps TYPE: int DEFAULT: 100

Source code in quantflow/sp/heston.py

def sample(
    self,
    n: Annotated[int, Doc("Number of sample paths")],
    time_horizon: Annotated[float, Doc("Time horizon")] = 1,
    time_steps: Annotated[int, Doc("Number of discrete time steps")] = 100,
) -> Paths:
    dw1 = Paths.normal_draws(n, time_horizon, time_steps)
    dw2 = Paths.normal_draws(n, time_horizon, time_steps)
    return self.sample_from_draws(dw1, dw2)

sample_from_draws

sample_from_draws(path1, *args)

PARAMETER	DESCRIPTION
path1	Pre-drawn standard normal increments for the first Brownian motion TYPE: Paths
*args	Optional pre-drawn increments for the second Brownian motion; new draws are generated if omitted TYPE: Paths DEFAULT: ()

Source code in quantflow/sp/heston.py

def sample_from_draws(
    self,
    path1: Annotated[
        Paths,
        Doc("Pre-drawn standard normal increments for the first Brownian motion"),
    ],
    *args: Annotated[
        Paths,
        Doc(
            "Optional pre-drawn increments for the second Brownian motion; "
            "new draws are generated if omitted"
        ),
    ],
) -> Paths:
    if args:
        path2 = args[0]
    else:
        path2 = Paths.normal_draws(path1.samples, path1.t, path1.time_steps)
    dz = path1.data
    dw = self.rho * dz + np.sqrt(1 - self.rho * self.rho) * path2.data
    v = self.variance_process.sample_from_draws(path1)
    dx = dw * np.sqrt(v.data * path1.dt)
    paths = np.zeros(dx.shape)
    paths[1:] = np.cumsum(dx[:-1], axis=0)
    return Paths(t=path1.t, 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_mean

analytical_mean(t)

Analytical mean of the process at time t

Implement if available

Source code in quantflow/sp/base.py

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

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

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)

quantflow.sp.heston.HestonJ pydantic-model

Bases: Heston, Generic[D]

A generic Heston stochastic volatility model with jumps

The Heston model with jumps is an extension of the classical square-root stochastic volatility model of Heston (1993) with the addition of jump processes. The jumps are modeled via a compound Poisson process

\[\begin{align} d x_t &= d w^1_t + d N_t\\ d v_t &= \kappa (\theta - v_t) dt + \nu \sqrt{v_t} dw^2_t \\ \rho dt &= {\tt E}[dw^1 dw^2] \end{align}\]

This model is generic and therefore allows for different types of jumps distributions D.

The Bates model is obtained by using the Normal distribution for the jump sizes.

Fields:

• variance_process (CIR)
• rho (float)
• jumps (CompoundPoissonProcess[D])

jumps pydantic-field

jumps

Jump process

variance_process pydantic-field

variance_process

The variance process is a Cox-Ingersoll-Ross process which is guaranteed to be positive if the Feller condition is satisfied

rho pydantic-field

rho = 0

Correlation between the Brownian motions, it provides the leverage effect and therefore the skewness of the distribution

create classmethod

create(jump_distribution, *, rate=1.0, vol=0.5, kappa=1, sigma=0.8, rho=0, theta=None, jump_intensity=100, jump_fraction=0.1, jump_asymmetry=0.0)

Create an Heston model with DoubleExponential jumps.

To understand the parameters lets introduce the following notation:

\[\begin{align} {\tt var} &= {\tt vol}^2 \\ {\tt var}_j &= {\tt var} \cdot {\tt jump\_fraction} \\ {\tt var}_d &= {\tt var} - {\tt var}_j \\ v_0 &= {\tt rate}\cdot{\tt var}_d \end{align}\]

PARAMETER	DESCRIPTION
jump_distribution	The distribution of jump size (currently only Normal and DoubleExponential are supported) TYPE: type[D]
rate	Define the initial value of the variance process TYPE: float DEFAULT: 1.0
vol	The standard deviation of the price process, normalized by the square root of time, as time tends to infinity (the long term standard deviation) TYPE: float DEFAULT: 0.5
kappa	The mean reversion speed for the variance process TYPE: float DEFAULT: 1
sigma	The volatility of the variance process TYPE: float DEFAULT: 0.8
rho	The correlation between the Brownian motions of the variance and price processes TYPE: float DEFAULT: 0
theta	The long-term mean of the variance process, if None, it defaults to the diffusion variance given by \({\tt var}_d\) TYPE: float | None DEFAULT: None
jump_intensity	The average number of jumps per year TYPE: float DEFAULT: 100
jump_fraction	The fraction of variance due to jumps (between 0 and 1) TYPE: float DEFAULT: 0.1
jump_asymmetry	The asymmetry of the jump distribution (0 for symmetric jumps) TYPE: float DEFAULT: 0.0

Source code in quantflow/sp/heston.py

@classmethod
def create(  # type: ignore [override]
    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)"
        ),
    ],
    *,
    rate: Annotated[
        float, Doc("Define the initial value of the variance process")
    ] = 1.0,
    vol: Annotated[
        float,
        Doc(
            "The standard deviation of the price process, normalized by the"
            " square root of time, as time tends to infinity"
            " (the long term standard deviation)"
        ),
    ] = 0.5,
    kappa: Annotated[
        float, Doc("The mean reversion speed for the variance process")
    ] = 1,
    sigma: Annotated[float, Doc("The volatility of the variance process")] = 0.8,
    rho: Annotated[
        float,
        Doc(
            "The correlation between the Brownian motions of the"
            " variance and price processes"
        ),
    ] = 0,
    theta: Annotated[
        float | None,
        Doc(
            r"The long-term mean of the variance process, if `None`, it"
            r" defaults to the diffusion variance given by ${\tt var}_d$"
        ),
    ] = None,
    jump_intensity: Annotated[
        float, Doc("The average number of jumps per year")
    ] = 100,
    jump_fraction: Annotated[
        float, Doc("The fraction of variance due to jumps (between 0 and 1)")
    ] = 0.1,
    jump_asymmetry: Annotated[
        float, Doc("The asymmetry of the jump distribution (0 for symmetric jumps)")
    ] = 0.0,
) -> HestonJ[D]:
    r"""Create an Heston model with
    [DoubleExponential][quantflow.utils.distributions.DoubleExponential] jumps.

    To understand the parameters lets introduce the following notation:

    \begin{align}
        {\tt var} &= {\tt vol}^2 \\
        {\tt var}_j &= {\tt var} \cdot {\tt jump\_fraction} \\
        {\tt var}_d &= {\tt var} - {\tt var}_j \\
        v_0 &= {\tt rate}\cdot{\tt var}_d
    \end{align}
    """
    jd = JumpDiffusion.create(
        jump_distribution,
        vol=vol,
        jump_intensity=jump_intensity,
        jump_fraction=jump_fraction,
        jump_asymmetry=jump_asymmetry,
    )
    total_variance = vol * vol
    diffusion_variance = total_variance * (1 - jump_fraction)
    return cls(
        variance_process=CIR(
            rate=rate * diffusion_variance,
            kappa=kappa,
            sigma=sigma,
            theta=theta if theta is not None else diffusion_variance,
        ),
        rho=rho,
        jumps=jd.jumps,
    )

characteristic_exponent

characteristic_exponent(t, u)

Characteristic exponent as the sum of the Heston and jump exponents.

\[\begin{equation} \phi_{x_t,u} = \phi^{\text{Heston}}_{x_t,u} + \phi^{\text{jumps}}_{x_t,u} \end{equation}\]

PARAMETER	DESCRIPTION
t	Time horizon or array of evaluation times TYPE: FloatArrayLike
u	Characteristic exponent argument (imaginary frequency) TYPE: Vector

Source code in quantflow/sp/heston.py

def characteristic_exponent(
    self,
    t: Annotated[FloatArrayLike, Doc("Time horizon or array of evaluation times")],
    u: Annotated[
        Vector, Doc("Characteristic exponent argument (imaginary frequency)")
    ],
) -> Vector:
    r"""Characteristic exponent as the sum of the Heston and jump exponents.

    \begin{equation}
        \phi_{x_t,u} = \phi^{\text{Heston}}_{x_t,u} + \phi^{\text{jumps}}_{x_t,u}
    \end{equation}
    """
    return super().characteristic_exponent(
        t, u
    ) + self.jumps.characteristic_exponent(t, u)

sample_from_draws

sample_from_draws(path1, *args)

Source code in quantflow/sp/heston.py

def sample_from_draws(self, path1: Paths, *args: Paths) -> Paths:
    diffusion = super().sample_from_draws(path1, *args)
    jump_path = self.jumps.sample(
        diffusion.samples, diffusion.t, diffusion.time_steps
    )
    return Paths(t=diffusion.t, data=diffusion.data + jump_path.data)

sample

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

PARAMETER	DESCRIPTION
n	Number of sample paths TYPE: int
time_horizon	Time horizon TYPE: float DEFAULT: 1
time_steps	Number of discrete time steps TYPE: int DEFAULT: 100

Source code in quantflow/sp/heston.py

def sample(
    self,
    n: Annotated[int, Doc("Number of sample paths")],
    time_horizon: Annotated[float, Doc("Time horizon")] = 1,
    time_steps: Annotated[int, Doc("Number of discrete time steps")] = 100,
) -> Paths:
    dw1 = Paths.normal_draws(n, time_horizon, time_steps)
    dw2 = Paths.normal_draws(n, time_horizon, time_steps)
    return self.sample_from_draws(dw1, dw2)

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_mean

analytical_mean(t)

Analytical mean of the process at time t

Implement if available

Source code in quantflow/sp/base.py

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

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

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)

quantflow.sp.heston.DoubleHeston pydantic-model

Bases: StochasticProcess1D

Double Heston stochastic volatility model.

Two independent Heston processes drive a single log-price:

\[\begin{align} d x_t &= \sqrt{v^1_t}\,d w^1_t + \sqrt{v^2_t}\,d w^3_t \\ d v^i_t &= \kappa_i (\theta_i - v^i_t) dt + \nu_i \sqrt{v^i_t}\,d w^{2i}_t \\ \rho_i\,dt &= {\tt E}[d w^{2i-1} d w^{2i}] \quad i = 1, 2 \end{align}\]

Because the two components are independent, the characteristic exponent is the sum of the two individual Heston exponents.

Fields:

• heston1 (Heston)
• heston2 (Heston)

heston1 pydantic-field

heston1

First Heston variance process

heston2 pydantic-field

heston2

Second Heston variance process

characteristic_exponent

characteristic_exponent(t, u)

Characteristic exponent as the sum of two independent Heston exponents.

\[\begin{equation} \phi_{x_t,u} = \phi^{(1)}_{x_t,u} + \phi^{(2)}_{x_t,u} \end{equation}\]

PARAMETER	DESCRIPTION
t	Time horizon or array of evaluation times TYPE: FloatArrayLike
u	Characteristic exponent argument (imaginary frequency) TYPE: Vector

Source code in quantflow/sp/heston.py

def characteristic_exponent(
    self,
    t: Annotated[FloatArrayLike, Doc("Time horizon or array of evaluation times")],
    u: Annotated[
        Vector, Doc("Characteristic exponent argument (imaginary frequency)")
    ],
) -> Vector:
    r"""Characteristic exponent as the sum of two independent Heston exponents.

    \begin{equation}
        \phi_{x_t,u} = \phi^{(1)}_{x_t,u} + \phi^{(2)}_{x_t,u}
    \end{equation}
    """
    return self.heston1.characteristic_exponent(
        t, u
    ) + self.heston2.characteristic_exponent(t, u)

sample

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

PARAMETER	DESCRIPTION
n	Number of sample paths TYPE: int
time_horizon	Time horizon TYPE: float DEFAULT: 1
time_steps	Number of discrete time steps TYPE: int DEFAULT: 100

Source code in quantflow/sp/heston.py

def sample(
    self,
    n: Annotated[int, Doc("Number of sample paths")],
    time_horizon: Annotated[float, Doc("Time horizon")] = 1,
    time_steps: Annotated[int, Doc("Number of discrete time steps")] = 100,
) -> Paths:
    dw1 = Paths.normal_draws(n, time_horizon, time_steps)
    dw2 = Paths.normal_draws(n, time_horizon, time_steps)
    dw3 = Paths.normal_draws(n, time_horizon, time_steps)
    dw4 = Paths.normal_draws(n, time_horizon, time_steps)
    return self.sample_from_draws(dw1, dw2, dw3, dw4)

sample_from_draws

sample_from_draws(path1, *args)

PARAMETER	DESCRIPTION
path1	First Brownian motion draws for heston1 TYPE: Paths
*args	args[0]: second BM draws for heston1; args[1], args[2]: first and second BM draws for heston2 TYPE: Paths DEFAULT: ()

Source code in quantflow/sp/heston.py

def sample_from_draws(
    self,
    path1: Annotated[Paths, Doc("First Brownian motion draws for heston1")],
    *args: Annotated[
        Paths,
        Doc(
            "args[0]: second BM draws for heston1; "
            "args[1], args[2]: first and second BM draws for heston2"
        ),
    ],
) -> Paths:
    paths1 = self.heston1.sample_from_draws(path1, args[0])
    paths2 = self.heston2.sample_from_draws(args[1], args[2])
    return Paths(t=path1.t, data=paths1.data + paths2.data)

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_mean

analytical_mean(t)

Analytical mean of the process at time t

Implement if available

Source code in quantflow/sp/base.py

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

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

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)

quantflow.sp.heston.DoubleHestonJ pydantic-model

Bases: DoubleHeston, Generic[D]

Double Heston stochastic volatility model with jumps.

Extends DoubleHeston by replacing the first (short-maturity) Heston process with a HestonJ that carries a jump component. Jumps are assigned to the short end because they fade away at longer maturities:

\[\begin{equation} \phi_{x_t,u} = \phi^{(1)}_{x_t,u} + \phi^{\text{jumps}}_{x_t,u} + \phi^{(2)}_{x_t,u} \end{equation}\]

where \(\phi^{(1)}\) and \(\phi^{\text{jumps}}\) are both provided by the HestonJ first process.

Fields:

• heston2 (Heston)
• heston1 (HestonJ[D])

heston1 pydantic-field

heston1

First (short-maturity) Heston process with jumps

heston2 pydantic-field

heston2

Second Heston variance process

sample_from_draws

sample_from_draws(path1, *args)

PARAMETER	DESCRIPTION
path1	First Brownian motion draws for heston1 TYPE: Paths
*args	args[0]: second BM draws for heston1; args[1], args[2]: first and second BM draws for heston2 TYPE: Paths DEFAULT: ()

Source code in quantflow/sp/heston.py

def sample_from_draws(
    self,
    path1: Annotated[Paths, Doc("First Brownian motion draws for heston1")],
    *args: Annotated[
        Paths,
        Doc(
            "args[0]: second BM draws for heston1; "
            "args[1], args[2]: first and second BM draws for heston2"
        ),
    ],
) -> Paths:
    paths1 = self.heston1.sample_from_draws(path1, args[0])
    paths2 = self.heston2.sample_from_draws(args[1], args[2])
    return Paths(t=path1.t, data=paths1.data + paths2.data)

sample

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

PARAMETER	DESCRIPTION
n	Number of sample paths TYPE: int
time_horizon	Time horizon TYPE: float DEFAULT: 1
time_steps	Number of discrete time steps TYPE: int DEFAULT: 100

Source code in quantflow/sp/heston.py

def sample(
    self,
    n: Annotated[int, Doc("Number of sample paths")],
    time_horizon: Annotated[float, Doc("Time horizon")] = 1,
    time_steps: Annotated[int, Doc("Number of discrete time steps")] = 100,
) -> Paths:
    dw1 = Paths.normal_draws(n, time_horizon, time_steps)
    dw2 = Paths.normal_draws(n, time_horizon, time_steps)
    dw3 = Paths.normal_draws(n, time_horizon, time_steps)
    dw4 = Paths.normal_draws(n, time_horizon, time_steps)
    return self.sample_from_draws(dw1, dw2, dw3, dw4)

characteristic_exponent

characteristic_exponent(t, u)

Characteristic exponent as the sum of two independent Heston exponents.

\[\begin{equation} \phi_{x_t,u} = \phi^{(1)}_{x_t,u} + \phi^{(2)}_{x_t,u} \end{equation}\]

PARAMETER	DESCRIPTION
t	Time horizon or array of evaluation times TYPE: FloatArrayLike
u	Characteristic exponent argument (imaginary frequency) TYPE: Vector

Source code in quantflow/sp/heston.py

def characteristic_exponent(
    self,
    t: Annotated[FloatArrayLike, Doc("Time horizon or array of evaluation times")],
    u: Annotated[
        Vector, Doc("Characteristic exponent argument (imaginary frequency)")
    ],
) -> Vector:
    r"""Characteristic exponent as the sum of two independent Heston exponents.

    \begin{equation}
        \phi_{x_t,u} = \phi^{(1)}_{x_t,u} + \phi^{(2)}_{x_t,u}
    \end{equation}
    """
    return self.heston1.characteristic_exponent(
        t, u
    ) + self.heston2.characteristic_exponent(t, u)

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_mean

analytical_mean(t)

Analytical mean of the process at time t

Implement if available

Source code in quantflow/sp/base.py

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

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

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)