Distributions

quantflow.utils.distributions.Distribution1D pydantic-model

Bases: Marginal1D

Base class for 1D distributions to be used as jump distributions in CompoundPoisson

sample abstractmethod

sample(n)

Sample from the distribution

PARAMETER	DESCRIPTION
n	Number of samples to draw TYPE: int

Source code in quantflow/utils/distributions.py

@abstractmethod
def sample(self, n: Annotated[int, Doc("Number of samples to draw")]) -> np.ndarray:
    """Sample from the distribution"""

from_variance_and_asymmetry classmethod

from_variance_and_asymmetry(variance, asymmetry)

Create a distribution from variance and asymmetry

Source code in quantflow/utils/distributions.py

@classmethod
def from_variance_and_asymmetry(cls, variance: float, asymmetry: float) -> Self:
    """Create a distribution from variance and asymmetry"""
    raise NotImplementedError

asymmetry

asymmetry()

Asymmetry of the distribution, 0 for symmetric distributions

Source code in quantflow/utils/distributions.py

def asymmetry(self) -> float:
    """Asymmetry of the distribution, 0 for symmetric distributions"""
    raise NotImplementedError

set_variance

set_variance(variance)

Set the variance of the distribution

Source code in quantflow/utils/distributions.py

def set_variance(self, variance: float) -> None:
    """Set the variance of the distribution"""
    raise NotImplementedError

set_asymmetry

set_asymmetry(asymmetry)

Set the asymmetry of the distribution

Source code in quantflow/utils/distributions.py

def set_asymmetry(self, asymmetry: float) -> None:
    """Set the asymmetry of the distribution"""
    raise NotImplementedError

call_option

call_option(n=None, *, pricing_method=CARR_MADAN, cos_moneyness_std_precision=12, max_moneyness=1.5, max_frequency=None, alpha=None, simpson_rule=False, use_fft=False)

Price call options via one of the available Fourier-based methods, as a function of log-strike

PARAMETER	DESCRIPTION
n	Number of discretization points for the transform. Defaults to 128. TYPE: int | None DEFAULT: None
pricing_method	Method to use for option pricing via Fourier transform of the call option price. Defaults to Lewis. TYPE: OptionPricingMethod DEFAULT: CARR_MADAN
cos_moneyness_std_precision	Truncation parameter for COS: the integration interval is set to [-cos_moneyness_std_precisionstd, cos_moneyness_std_precisionstd]. TYPE: float DEFAULT: 12
max_moneyness	Maximum moneyness to calculate prices. The log-strike grid is set to [-max_moneynessstd, max_moneynessstd]. Used by Lewis and Carr & Madan methods only. TYPE: float DEFAULT: 1.5
max_frequency	Maximum frequency for the transform grid. Defaults to frequency_range(). TYPE: float | None DEFAULT: None
alpha	Damping parameter for integrability of the Carr-Madan integrand, it is ignored if not applicable. TYPE: float | None DEFAULT: None
simpson_rule	Use Simpson's rule for integration. Default is False. TYPE: bool DEFAULT: False
use_fft	Use FFT for the transform rather than FRFT. Default is False. TYPE: bool DEFAULT: False

Source code in quantflow/utils/marginal.py

def call_option(
    self,
    n: Annotated[
        int | None,
        Doc("Number of discretization points for the transform. Defaults to 128."),
    ] = None,
    *,
    pricing_method: Annotated[
        OptionPricingMethod,
        Doc(
            "Method to use for option pricing via Fourier transform of the "
            "call option price. Defaults to Lewis."
        ),
    ] = OptionPricingMethod.CARR_MADAN,
    cos_moneyness_std_precision: Annotated[
        float,
        Doc(
            "Truncation parameter for COS: the integration interval is set to "
            "[-cos_moneyness_std_precision*std, cos_moneyness_std_precision*std]."
        ),
    ] = 12,
    max_moneyness: Annotated[
        float,
        Doc(
            "Maximum moneyness to calculate prices. The log-strike grid is set to "
            "[-max_moneyness*std, max_moneyness*std]. "
            "Used by Lewis and Carr & Madan methods only."
        ),
    ] = 1.5,
    max_frequency: Annotated[
        float | None,
        Doc(
            "Maximum frequency for the transform grid. "
            "Defaults to frequency_range()."
        ),
    ] = None,
    alpha: Annotated[
        float | None,
        Doc(
            "Damping parameter for integrability of the Carr-Madan integrand, "
            "it is ignored if not applicable."
        ),
    ] = None,
    simpson_rule: Annotated[
        bool, Doc("Use Simpson's rule for integration. Default is False.")
    ] = False,
    use_fft: Annotated[
        bool, Doc("Use FFT for the transform rather than FRFT. Default is False.")
    ] = False,
) -> OptionPricingResult:
    """Price call options via one of the available Fourier-based methods,
    as a function of log-strike
    """
    match pricing_method:
        case OptionPricingMethod.COS:
            return self.call_option_cos(
                n,
                moneyness_std_precision=cos_moneyness_std_precision,
            )
        case OptionPricingMethod.CARR_MADAN:
            return self.call_option_carr_madan(
                n,
                max_frequency=max_frequency,
                max_moneyness=max_moneyness,
                alpha=alpha,
                simpson_rule=simpson_rule,
                use_fft=use_fft,
            )
        case _:
            return self.call_option_lewis(
                n,
                max_frequency=max_frequency,
                max_moneyness=max_moneyness,
                simpson_rule=simpson_rule,
                use_fft=use_fft,
            )

call_option_carr_madan

call_option_carr_madan(n=None, *, max_moneyness=1.5, max_frequency=None, alpha=None, simpson_rule=False, use_fft=False)

Call option price via Carr & Madan method

PARAMETER	DESCRIPTION
n	Number of discretization points for the transform. Defaults to 128. TYPE: int | None DEFAULT: None
max_moneyness	Maximum moneyness to calculate prices. The log-strike grid is set to [-max_moneynessstd, max_moneynessstd]. TYPE: float DEFAULT: 1.5
max_frequency	Maximum frequency for the transform grid. Defaults to frequency_range(). TYPE: float | None DEFAULT: None
alpha	Damping parameter for integrability of the Carr-Madan integrand. Defaults to call_option_carr_madan_alpha. TYPE: float | None DEFAULT: None
simpson_rule	Use Simpson's rule for integration. Default is False. TYPE: bool DEFAULT: False
use_fft	Use FFT for the transform. Default is False. TYPE: bool DEFAULT: False

Source code in quantflow/utils/marginal.py

def call_option_carr_madan(
    self,
    n: Annotated[
        int | None,
        Doc("Number of discretization points for the transform. Defaults to 128."),
    ] = None,
    *,
    max_moneyness: Annotated[
        float,
        Doc(
            "Maximum moneyness to calculate prices. The log-strike grid is set to "
            "[-max_moneyness*std, max_moneyness*std]. "
        ),
    ] = 1.5,
    max_frequency: Annotated[
        float | None,
        Doc(
            "Maximum frequency for the transform grid. "
            "Defaults to frequency_range()."
        ),
    ] = None,
    alpha: Annotated[
        float | None,
        Doc(
            "Damping parameter for integrability of the Carr-Madan integrand. "
            "Defaults to "
            "[call_option_carr_madan_alpha][..call_option_carr_madan_alpha]."
        ),
    ] = None,
    simpson_rule: Annotated[
        bool, Doc("Use Simpson's rule for integration. Default is False.")
    ] = False,
    use_fft: Annotated[
        bool, Doc("Use FFT for the transform. Default is False.")
    ] = False,
) -> OptionPricingTransformResult:
    """Call option price via Carr & Madan method"""
    max_log_strike = max_moneyness * self.std_validated()
    transform = self.get_transform(
        n,
        lambda m: self.option_support(m + 1, max_log_strike=max_log_strike),
        max_frequency=max_frequency,
        simpson_rule=simpson_rule,
        use_fft=use_fft,
    )
    alpha = alpha or self.call_option_carr_madan_alpha()
    phi = cast(
        np.ndarray,
        self.call_option_transform(transform.frequency_domain - 1j * alpha),
    )
    result = transform(phi, use_fft=use_fft)
    return OptionPricingTransformResult(
        log_strikes=result.x,
        call_prices=result.y * np.exp(-alpha * result.x),
    )

call_option_carr_madan_alpha

call_option_carr_madan_alpha()

Option alpha to use for Carr & Madan transform to ensure integrability of the call option transform.

Defaults to 1.5, the value suggested in the original Carr & Madan paper.

Source code in quantflow/utils/marginal.py

def call_option_carr_madan_alpha(self) -> float:
    """Option alpha to use for Carr & Madan transform to ensure integrability
    of the call option transform.

    Defaults to 1.5, the value suggested in the original Carr & Madan paper.
    """
    return 1.5

call_option_cos

call_option_cos(n=None, *, moneyness_std_precision=12)

Call option price via the COS method (Fang & Oosterlee 2008).

The call price at log-strike \(k\) is approximated by the cosine series

\[\begin{equation} C(k) \approx e^k \sum_{j=0}^{N-1}{}^{\prime} \mathrm{Re}\!\left[ \phi\!\left(\frac{j\pi}{b-a}\right) e^{-i j\pi (k+a)/(b-a)} \right] V_j \end{equation}\]

where the prime denotes a half weight on the \(j=0\) term, \([a,b]\) is the truncation interval, and \(V_j\) are the cosine payoff coefficients for the normalised call payoff \((e^y - 1)^+\) integrated over \([0, b]\). The \(e^k\) factor converts from strike-normalised to forward-space pricing.

Returns an OptionPricingCosResult with the precomputed coefficient vector. Use call_price to evaluate at arbitrary log-strikes in \(O(N)\) per strike.

PARAMETER	DESCRIPTION
n	Number of cosine series terms. Defaults to 128. TYPE: int | None DEFAULT: None
moneyness_std_precision	Truncation parameter: the integration interval is set to [-moneyness_std_precisionstd, moneyness_std_precisionstd]. TYPE: float DEFAULT: 12

Source code in quantflow/utils/marginal.py

def call_option_cos(
    self,
    n: Annotated[
        int | None,
        Doc("Number of cosine series terms. Defaults to 128."),
    ] = None,
    *,
    moneyness_std_precision: Annotated[
        float,
        Doc(
            "Truncation parameter: the integration interval is set to "
            "[-moneyness_std_precision*std, moneyness_std_precision*std]."
        ),
    ] = 12,
) -> OptionPricingCosResult:
    r"""Call option price via the COS method (Fang & Oosterlee 2008).

    The call price at log-strike $k$ is approximated by the cosine series

    \begin{equation}
        C(k) \approx e^k \sum_{j=0}^{N-1}{}^{\prime}
            \mathrm{Re}\!\left[
                \phi\!\left(\frac{j\pi}{b-a}\right)
                e^{-i j\pi (k+a)/(b-a)}
            \right] V_j
    \end{equation}

    where the prime denotes a half weight on the $j=0$ term, $[a,b]$ is the
    truncation interval, and $V_j$ are the cosine payoff coefficients for the
    normalised call payoff $(e^y - 1)^+$ integrated over $[0, b]$.
    The $e^k$ factor converts from strike-normalised to forward-space pricing.

    Returns an
    [OptionPricingCosResult][quantflow.utils.marginal.OptionPricingCosResult]
    with the precomputed coefficient vector. Use
    [call_price][quantflow.utils.marginal.OptionPricingCosResult.call_price]
    to evaluate at arbitrary log-strikes in $O(N)$ per strike.
    """
    n = n or 128
    std = self.std_validated()
    a = -moneyness_std_precision * std
    b = moneyness_std_precision * std
    bma = b - a

    j = np.arange(n)
    nu = j * np.pi / bma

    phi = np.asarray(self.characteristic_corrected(nu))

    sin_nu_a = np.sin(nu * a)
    cos_nu_a = np.cos(nu * a)
    sign_j = (-1.0) ** j
    safe_nu = np.where(j == 0, 1.0, nu)
    chi = (np.exp(b) * sign_j - cos_nu_a + nu * sin_nu_a) / np.where(
        j == 0, 1.0, 1.0 + nu**2
    )
    psi = np.where(j == 0, b, sin_nu_a / safe_nu)
    V = 2.0 / bma * (chi - psi)
    weights = np.ones(n)
    weights[0] = 0.5
    return OptionPricingCosResult(
        a=a,
        nu=nu,
        coeff=weights * phi * V,
    )

call_option_lewis

call_option_lewis(n=None, *, max_moneyness=1.5, max_frequency=None, simpson_rule=False, use_fft=False)

Call option price via the Lewis (2001) formula

PARAMETER	DESCRIPTION
n	Number of discretization points for the transform. Defaults to 128. TYPE: int | None DEFAULT: None
max_moneyness	Maximum moneyness to calculate prices. The log-strike grid is set to [-max_moneynessstd, max_moneynessstd]. TYPE: float DEFAULT: 1.5
max_frequency	Maximum frequency for the transform grid. Defaults to frequency_range(). TYPE: float | None DEFAULT: None
simpson_rule	Use Simpson's rule for integration. Default is False. TYPE: bool DEFAULT: False
use_fft	Use FFT for the transform. Default is False. TYPE: bool DEFAULT: False

Source code in quantflow/utils/marginal.py

def call_option_lewis(
    self,
    n: Annotated[
        int | None,
        Doc("Number of discretization points for the transform. Defaults to 128."),
    ] = None,
    *,
    max_moneyness: Annotated[
        float,
        Doc(
            "Maximum moneyness to calculate prices. The log-strike grid is set to "
            "[-max_moneyness*std, max_moneyness*std]. "
        ),
    ] = 1.5,
    max_frequency: Annotated[
        float | None,
        Doc(
            "Maximum frequency for the transform grid. "
            "Defaults to frequency_range()."
        ),
    ] = None,
    simpson_rule: Annotated[
        bool, Doc("Use Simpson's rule for integration. Default is False.")
    ] = False,
    use_fft: Annotated[
        bool, Doc("Use FFT for the transform. Default is False.")
    ] = False,
) -> OptionPricingTransformResult:
    """Call option price via the Lewis (2001) formula"""
    max_log_strike = max_moneyness * self.std_validated()
    transform = self.get_transform(
        n,
        lambda m: self.option_support(m + 1, max_log_strike=max_log_strike),
        max_frequency=max_frequency,
        simpson_rule=simpson_rule,
        use_fft=use_fft,
    )
    phi = cast(np.ndarray, self.lewis_transform(transform.frequency_domain))
    result = transform(phi, use_fft=use_fft)
    k = result.x
    return OptionPricingTransformResult(
        log_strikes=k,
        call_prices=1.0 - np.exp(0.5 * k) * result.y,
    )

call_option_transform

call_option_transform(u)

Call option transform

PARAMETER	DESCRIPTION
u	Frequency domain points (possibly complex-shifted). TYPE: Vector

Source code in quantflow/utils/marginal.py

def call_option_transform(
    self,
    u: Annotated[
        Vector, Doc("Frequency domain points (possibly complex-shifted).")
    ],
) -> Vector:
    """Call option transform"""
    uj = 1j * u
    return self.characteristic_corrected(u - 1j) / (uj * uj + uj)

cdf

cdf(x)

Compute the cumulative distribution function

PARAMETER	DESCRIPTION
x	Location in the stochastic process domain space. If a numpy array, the output should have the same shape as the input. TYPE: FloatArrayLike

Source code in quantflow/utils/marginal.py

def cdf(
    self,
    x: Annotated[
        FloatArrayLike,
        Doc(
            "Location in the stochastic process domain space. If a numpy array,"
            " the output should have the same shape as the input."
        ),
    ],
) -> FloatArrayLike:
    """
    Compute the cumulative distribution function
    """
    raise NotImplementedError("Analytical CDF not available")

cdf_from_characteristic

cdf_from_characteristic(n=None, *, max_frequency=None, simpson_rule=False, use_fft=False, frequency_n=None)

PARAMETER	DESCRIPTION
n	Number of discretization points for the transform. Defaults to 128. TYPE: int | None DEFAULT: None
max_frequency	Maximum frequency for the transform grid. Defaults to frequency_range(). TYPE: float | None DEFAULT: None
simpson_rule	Use Simpson's rule for integration. Default is False. TYPE: bool DEFAULT: False
use_fft	Use FFT for the transform. Default is False. TYPE: bool DEFAULT: False
frequency_n	Number of points for the frequency grid. Overrides n if provided. TYPE: int | None DEFAULT: None

Source code in quantflow/utils/marginal.py

def cdf_from_characteristic(
    self,
    n: Annotated[
        int | None,
        Doc("Number of discretization points for the transform. Defaults to 128."),
    ] = None,
    *,
    max_frequency: Annotated[
        float | None,
        Doc(
            "Maximum frequency for the transform grid. "
            "Defaults to frequency_range()."
        ),
    ] = None,
    simpson_rule: Annotated[
        bool, Doc("Use Simpson's rule for integration. Default is False.")
    ] = False,
    use_fft: Annotated[
        bool, Doc("Use FFT for the transform. Default is False.")
    ] = False,
    frequency_n: Annotated[
        int | None,
        Doc("Number of points for the frequency grid. Overrides n if provided."),
    ] = None,
) -> TransformResult:
    raise NotImplementedError("CDF not available")

cdf_jacobian

cdf_jacobian(x)

Jacobian of the cdf with respect to the parameters of the process. It is useful for optimization purposes if necessary.

Optional to implement, otherwise raises NotImplementedError if called.

PARAMETER	DESCRIPTION
x	Location in the state space of the process. TYPE: FloatArrayLike

Source code in quantflow/utils/marginal.py

def cdf_jacobian(
    self,
    x: Annotated[
        FloatArrayLike, Doc("Location in the state space of the process.")
    ],
) -> np.ndarray:
    """
    Jacobian of the cdf with respect to the parameters of the process.
    It is useful for optimization purposes if necessary.

    Optional to implement, otherwise raises ``NotImplementedError`` if called.
    """
    raise NotImplementedError("Analytical CDF Jacobian not available")

characteristic abstractmethod

characteristic(u)

Compute the characteristic function on frequency domain points \(u\)

PARAMETER	DESCRIPTION
u	Frequency domain points. TYPE: Vector

Source code in quantflow/utils/marginal.py

@abstractmethod
def characteristic(
    self,
    u: Annotated[Vector, Doc("Frequency domain points.")],
) -> Vector:
    """
    Compute the characteristic function on frequency domain points $u$
    """

characteristic_corrected

characteristic_corrected(u)

Characteristic function corrected for the convexity of the log-price distribution

PARAMETER	DESCRIPTION
u	Frequency domain points. TYPE: Vector

Source code in quantflow/utils/marginal.py

def characteristic_corrected(
    self,
    u: Annotated[Vector, Doc("Frequency domain points.")],
) -> Vector:
    """Characteristic function corrected for the convexity of the log-price
    distribution"""
    convexity = np.log(self.characteristic(-1j))
    return self.characteristic(u) * np.exp(-1j * u * convexity)

characteristic_df

characteristic_df(n=None, *, max_frequency=None, simpson_rule=False)

Compute the characteristic function with n discretization points and a max frequency

PARAMETER	DESCRIPTION
n	Number of discretization points for the transform. Defaults to 128. TYPE: int | None DEFAULT: None
max_frequency	Maximum frequency for the transform grid. Defaults to frequency_range(). TYPE: float | None DEFAULT: None
simpson_rule	Use Simpson's rule for integration. Default is False. TYPE: bool DEFAULT: False

Source code in quantflow/utils/marginal.py

def characteristic_df(
    self,
    n: Annotated[
        int | None,
        Doc("Number of discretization points for the transform. Defaults to 128."),
    ] = None,
    *,
    max_frequency: Annotated[
        float | None,
        Doc(
            "Maximum frequency for the transform grid. "
            "Defaults to frequency_range()."
        ),
    ] = None,
    simpson_rule: Annotated[
        bool, Doc("Use Simpson's rule for integration. Default is False.")
    ] = False,
) -> pd.DataFrame:
    """
    Compute the characteristic function with n discretization points
    and a max frequency
    """
    transform = self.get_transform(
        n,
        self.support,
        max_frequency=max_frequency,
        simpson_rule=simpson_rule,
    )
    psi = self.characteristic(transform.frequency_domain)
    return transform.characteristic_df(cast(np.ndarray, psi))

domain_range

domain_range()

The space domain range for the random variable

This should be overloaded if required

Source code in quantflow/utils/marginal.py

def domain_range(self) -> Bounds:
    """The space domain range for the random variable

    This should be overloaded if required
    """
    return default_bounds()

frequency_range

frequency_range(max_frequency=None)

The frequency domain range for the characteristic function

This should be overloaded if required

PARAMETER	DESCRIPTION
max_frequency	Upper bound of the frequency grid. Defaults to 20 if None. TYPE: float | None DEFAULT: None

Source code in quantflow/utils/marginal.py

def frequency_range(
    self,
    max_frequency: Annotated[
        float | None,
        Doc("Upper bound of the frequency grid. Defaults to 20 if None."),
    ] = None,
) -> float:
    """The frequency domain range for the characteristic function

    This should be overloaded if required
    """
    return Bounds(0, max_frequency or 20)

get_transform

get_transform(n, support, *, max_frequency=None, simpson_rule=False, use_fft=False)

PARAMETER	DESCRIPTION
n	Number of discretization points. Defaults to 128. TYPE: int | None
support	Function returning the space domain grid given the number of points. TYPE: Callable[[int], FloatArray]
max_frequency	Maximum frequency for the transform grid. Defaults to frequency_range(). TYPE: float | None DEFAULT: None
simpson_rule	Use Simpson's rule for integration. Default is False. TYPE: bool DEFAULT: False
use_fft	Use FFT for the transform. Default is False. TYPE: bool DEFAULT: False

Source code in quantflow/utils/marginal.py

def get_transform(
    self,
    n: Annotated[
        int | None,
        Doc("Number of discretization points. Defaults to 128."),
    ],
    support: Annotated[
        Callable[[int], FloatArray],
        Doc("Function returning the space domain grid given the number of points."),
    ],
    *,
    max_frequency: Annotated[
        float | None,
        Doc(
            "Maximum frequency for the transform grid. "
            "Defaults to frequency_range()."
        ),
    ] = None,
    simpson_rule: Annotated[
        bool, Doc("Use Simpson's rule for integration. Default is False.")
    ] = False,
    use_fft: Annotated[
        bool, Doc("Use FFT for the transform. Default is False.")
    ] = False,
) -> Transform:
    n = n or 128
    if use_fft:
        bounds = self.domain_range()
    else:
        x = support(n)
        bounds = Bounds(float(np.min(x)), float(np.max(x)))
    return Transform.create(
        n,
        frequency_range=self.frequency_range(max_frequency),
        domain_range=bounds,
        simpson_rule=simpson_rule,
    )

lewis_transform

lewis_transform(u)

Lewis (2001) call option transform - no damping parameter required

PARAMETER	DESCRIPTION
u	Frequency domain points. TYPE: Vector

Source code in quantflow/utils/marginal.py

def lewis_transform(
    self,
    u: Annotated[Vector, Doc("Frequency domain points.")],
) -> Vector:
    """Lewis (2001) call option transform - no damping parameter required"""
    return self.characteristic_corrected(u - 0.5j) / (u * u + 0.25)

mean

mean()

Expected value

By default it uses the mean_from_characteristic method. This should be overloaded if a more efficient/analytical way of computing the mean is available.

Source code in quantflow/utils/marginal.py

def mean(self) -> FloatArrayLike:
    """Expected value

    By default it uses the
    [mean_from_characteristic][..mean_from_characteristic] method.
    This should be overloaded if a more efficient/analytical way of computing
    the mean is available.
    """
    return self.mean_from_characteristic()

mean_from_characteristic

mean_from_characteristic(*, d=0.001)

Calculate mean as first derivative of characteristic function at 0

PARAMETER	DESCRIPTION
d	Step size for finite-difference approximation of the derivative. TYPE: float DEFAULT: 0.001

Source code in quantflow/utils/marginal.py

def mean_from_characteristic(
    self,
    *,
    d: Annotated[
        float,
        Doc("Step size for finite-difference approximation of the derivative."),
    ] = 0.001,
) -> FloatArrayLike:
    """Calculate mean as first derivative of characteristic function at 0"""
    m = -0.5 * 1j * (self.characteristic(d) - self.characteristic(-d)) / d
    return m.real

option_support

option_support(points=101, max_log_strike=1.0)

Compute the x axis.

PARAMETER	DESCRIPTION
points	Number of support points. TYPE: int DEFAULT: 101
max_log_strike	Maximum absolute log-strike. TYPE: float DEFAULT: 1.0

Source code in quantflow/utils/marginal.py

def option_support(
    self,
    points: Annotated[int, Doc("Number of support points.")] = 101,
    max_log_strike: Annotated[float, Doc("Maximum absolute log-strike.")] = 1.0,
) -> FloatArray:
    """
    Compute the x axis.
    """
    return np.linspace(-max_log_strike, max_log_strike, points)

option_time_value

option_time_value(n=128, *, max_frequency=None, max_log_strike=1, alpha=1.1, simpson_rule=False, use_fft=False)

Option time value

PARAMETER	DESCRIPTION
n	Number of discretization points for the transform. TYPE: int DEFAULT: 128
max_frequency	Maximum frequency for the transform grid. Defaults to frequency_range(). TYPE: float | None DEFAULT: None
max_log_strike	Maximum absolute log-strike for the output grid. TYPE: float DEFAULT: 1
alpha	Contour shift parameter controlling the integration strip. TYPE: float DEFAULT: 1.1
simpson_rule	Use Simpson's rule for integration. Default is False. TYPE: bool DEFAULT: False
use_fft	Use FFT for the transform. Default is False. TYPE: bool DEFAULT: False

Source code in quantflow/utils/marginal.py

def option_time_value(
    self,
    n: Annotated[
        int,
        Doc("Number of discretization points for the transform."),
    ] = 128,
    *,
    max_frequency: Annotated[
        float | None,
        Doc(
            "Maximum frequency for the transform grid. "
            "Defaults to frequency_range()."
        ),
    ] = None,
    max_log_strike: Annotated[
        float, Doc("Maximum absolute log-strike for the output grid.")
    ] = 1,
    alpha: Annotated[
        float,
        Doc("Contour shift parameter controlling the integration strip."),
    ] = 1.1,
    simpson_rule: Annotated[
        bool, Doc("Use Simpson's rule for integration. Default is False.")
    ] = False,
    use_fft: Annotated[
        bool, Doc("Use FFT for the transform. Default is False.")
    ] = False,
) -> TransformResult:
    """Option time value"""
    transform = self.get_transform(
        n,
        lambda m: self.option_support(m + 1, max_log_strike=max_log_strike),
        max_frequency=max_frequency,
        simpson_rule=simpson_rule,
        use_fft=use_fft,
    )
    phi = cast(
        np.ndarray,
        self.option_time_value_transform(transform.frequency_domain, alpha),
    )
    result = transform(phi, use_fft=use_fft)
    time_value = result.y / np.sinh(alpha * result.x)
    return TransformResult(x=result.x, y=time_value)

option_time_value_transform

option_time_value_transform(u, alpha=1.1)

Option time value transform

This transform does not require any additional correction since the integrand is already bounded for positive and negative moneyness

PARAMETER	DESCRIPTION
u	Frequency domain points. TYPE: Vector
alpha	Contour shift parameter controlling the integration strip. TYPE: float DEFAULT: 1.1

Source code in quantflow/utils/marginal.py

def option_time_value_transform(
    self,
    u: Annotated[Vector, Doc("Frequency domain points.")],
    alpha: Annotated[
        float, Doc("Contour shift parameter controlling the integration strip.")
    ] = 1.1,
) -> Vector:
    """Option time value transform

    This transform does not require any additional correction since
    the integrand is already bounded for positive and negative moneyness"""
    ia = 1j * alpha
    return 0.5 * (
        self._option_time_value_transform(u - ia)
        - self._option_time_value_transform(u + ia)
    )

pdf

pdf(x)

Computes the probability density (or mass) function of the process.

It has a base implementation that computes the pdf from the cdf method, but a subclass should overload this method if a more optimized way of computing it is available.

PARAMETER	DESCRIPTION
x	Location in the stochastic process domain space. If a numpy array, the output should have the same shape as the input. TYPE: FloatArrayLike

Source code in quantflow/utils/marginal.py

def pdf(
    self,
    x: Annotated[
        FloatArrayLike,
        Doc(
            "Location in the stochastic process domain space. If a numpy array,"
            " the output should have the same shape as the input."
        ),
    ],
) -> FloatArrayLike:
    """
    Computes the probability density (or mass) function of the process.

    It has a base implementation that computes the pdf from the
    [cdf][quantflow.utils.marginal.Marginal1D.cdf] method, but a subclass should
    overload this method if a
    more optimized way of computing it is available.
    """
    raise NotImplementedError("Analytical PDF not available")

pdf_from_characteristic

pdf_from_characteristic(n=None, *, max_frequency=None, simpson_rule=False, use_fft=False, frequency_n=None)

Compute the probability density function from the characteristic function.

PARAMETER	DESCRIPTION
n	Number of discretization points to use in the transform. If None, use 128. TYPE: int | None DEFAULT: None
max_frequency	The maximum frequency to use in the transform. If not provided, the value from the frequency_range method is used. Only needed for special cases/testing. TYPE: float | None DEFAULT: None
simpson_rule	Use Simpson's rule for integration. Default is False. TYPE: bool DEFAULT: False
use_fft	Use FFT for the transform. Default is False. TYPE: bool DEFAULT: False
frequency_n	Number of points for the frequency grid. Overrides n if provided. TYPE: int | None DEFAULT: None

Source code in quantflow/utils/marginal.py

def pdf_from_characteristic(
    self,
    n: Annotated[
        int | None,
        Doc(
            "Number of discretization points to use in the transform."
            " If None, use 128."
        ),
    ] = None,
    *,
    max_frequency: Annotated[
        float | None,
        Doc(
            "The maximum frequency to use in the transform. If not provided,"
            " the value from the [frequency_range]"
            "[quantflow.utils.marginal.Marginal1D.frequency_range] method is used."
            " Only needed for special cases/testing."
        ),
    ] = None,
    simpson_rule: Annotated[
        bool, Doc("Use Simpson's rule for integration. Default is False.")
    ] = False,
    use_fft: Annotated[
        bool, Doc("Use FFT for the transform. Default is False.")
    ] = False,
    frequency_n: Annotated[
        int | None,
        Doc("Number of points for the frequency grid. Overrides n if provided."),
    ] = None,
) -> TransformResult:
    """
    Compute the probability density function from the characteristic function.
    """
    transform = self.get_transform(
        frequency_n or n,
        self.support,
        max_frequency=max_frequency,
        simpson_rule=simpson_rule,
        use_fft=use_fft,
    )
    psi = cast(np.ndarray, self.characteristic(transform.frequency_domain))
    return transform(psi, use_fft=use_fft)

pdf_jacobian

pdf_jacobian(x)

Jacobian of the pdf with respect to the parameters of the process. It has a base implementation that computes it from the cdf_jacobian method, but a subclass should overload this method if a more optimized way of computing it is available.

PARAMETER	DESCRIPTION
x	Location in the state space of the process. TYPE: FloatArrayLike

Source code in quantflow/utils/marginal.py

def pdf_jacobian(
    self,
    x: Annotated[
        FloatArrayLike, Doc("Location in the state space of the process.")
    ],
) -> FloatArrayLike:
    """
    Jacobian of the pdf with respect to the parameters of the process.
    It has a base implementation that computes it from the
    [cdf_jacobian][quantflow.utils.marginal.Marginal1D.cdf_jacobian] method,
    but a subclass should overload this method if a
    more optimized way of computing it is available.
    """
    return self.cdf_jacobian(x) - self.cdf_jacobian(x - 1)

std

std()

Standard deviation at a time horizon

Source code in quantflow/utils/marginal.py

def std(self) -> FloatArrayLike:
    """Standard deviation at a time horizon"""
    return np.sqrt(self.variance())

std_validated

std_validated()

Float standard deviation, raising if it is not finite or non-positive.

Used by the Fourier-based pricing methods to set the log-strike grid range; degenerate parameter samples (e.g. during calibration) can produce a non-finite std and would otherwise crash deep in the transform with a confusing error.

Source code in quantflow/utils/marginal.py

def std_validated(self) -> float:
    """Float standard deviation, raising if it is not finite or non-positive.

    Used by the Fourier-based pricing methods to set the log-strike grid
    range; degenerate parameter samples (e.g. during calibration) can
    produce a non-finite std and would otherwise crash deep in the
    transform with a confusing error.
    """
    std = float(self.std())
    if not np.isfinite(std) or std <= 0:
        raise ValueError(
            f"Marginal std is not finite or non-positive: {std!r}; "
            "model parameters are likely invalid"
        )
    return std

std_from_characteristic

std_from_characteristic()

Calculate standard deviation as square root of variance

Source code in quantflow/utils/marginal.py

def std_from_characteristic(self) -> FloatArrayLike:
    """Calculate standard deviation as square root of variance"""
    return np.sqrt(self.variance_from_characteristic())

support abstractmethod

support(points=100, *, std_mult=3)

Compute the x axis.

PARAMETER	DESCRIPTION
points	Number of support points. TYPE: int DEFAULT: 100
std_mult	Standard deviation multiplier for the support range. TYPE: float DEFAULT: 3

Source code in quantflow/utils/marginal.py

@abstractmethod
def support(
    self,
    points: Annotated[int, Doc("Number of support points.")] = 100,
    *,
    std_mult: Annotated[
        float, Doc("Standard deviation multiplier for the support range.")
    ] = 3,
) -> FloatArray:
    """
    Compute the x axis.
    """

variance

variance()

Variance

By default it uses the variance_from_characteristic method. This should be overloaded if a more efficient/analytical way of computing the variance is available.

Source code in quantflow/utils/marginal.py

def variance(self) -> FloatArrayLike:
    """Variance

    By default it uses the
    [variance_from_characteristic][..variance_from_characteristic] method.
    This should be overloaded if a more efficient/analytical way of computing
    the variance is available.
    """
    return self.variance_from_characteristic()

variance_from_characteristic

variance_from_characteristic(*, d=0.001)

Calculate variance as second derivative of characteristic function at 0

PARAMETER	DESCRIPTION
d	Step size for finite-difference approximation of the derivative. TYPE: float DEFAULT: 0.001

Source code in quantflow/utils/marginal.py

def variance_from_characteristic(
    self,
    *,
    d: Annotated[
        float,
        Doc("Step size for finite-difference approximation of the derivative."),
    ] = 0.001,
) -> FloatArrayLike:
    """Calculate variance as second derivative of characteristic function at 0"""
    c1 = self.characteristic(d)
    c0 = self.characteristic(0)
    c2 = self.characteristic(-d)
    m = -0.5 * 1j * (c1 - c2) / d
    s = -(c1 - 2 * c0 + c2) / (d * d) - m * m
    return s.real

quantflow.utils.distributions.Exponential pydantic-model

Bases: Distribution1D

The Exponential distribution is a continuous probability distribution that describes the time between events in a Poisson process.

It is a special case of the gamma distribution and is given by

\[ f(x) = \lambda e^{-\lambda x}\ \ \forall x \geq 0 \]

Fields:

• decay (float)

decay pydantic-field

decay = 1

exponential decay rate \(\lambda\)

scale property

scale

The scale parameter, it is the inverse of the decay rate

scale2 property

scale2

characteristic

characteristic(u)

The characteristic function of the exponential distribution is given by

\[ \Phi_u = \frac{\lambda}{\lambda - i u} \]

Source code in quantflow/utils/distributions.py

def characteristic(self, u: Vector) -> Vector:
    r"""The characteristic function of the exponential distribution is given by

    $$
    \Phi_u = \frac{\lambda}{\lambda - i u}
    $$
    """
    return self.decay / (self.decay - 1j * u)

mean

mean()

Source code in quantflow/utils/distributions.py

def mean(self) -> float:
    return self.scale

variance

variance()

Source code in quantflow/utils/distributions.py

def variance(self) -> float:
    return self.scale2

sample

sample(n)

Source code in quantflow/utils/distributions.py

def sample(self, n: int) -> np.ndarray:
    return np.random.exponential(scale=self.scale, size=n)

support

support(points=100, *, std_mult=4)

Source code in quantflow/utils/distributions.py

def support(self, points: int = 100, *, std_mult: float = 4) -> FloatArray:
    return np.linspace(0, std_mult * np.max(self.std()), points)

pdf

pdf(x)

The analytical PDF of the exponential distribution as defined above

Source code in quantflow/utils/distributions.py

def pdf(self, x: FloatArrayLike) -> FloatArrayLike:
    """The analytical PDF of the exponential distribution as defined above"""
    return self.decay * np.exp(-self.decay * x)

cdf

cdf(x)

The analytical CDF of the exponential distribution

\[ F(x) = 1 - e^{-\lambda x}\ \ \forall x \geq 0 \]

Source code in quantflow/utils/distributions.py

def cdf(self, x: FloatArrayLike) -> FloatArrayLike:
    r"""The analytical CDF of the exponential distribution

    $$
    F(x) = 1 - e^{-\lambda x}\ \ \forall x \geq 0
    $$
    """
    return 1.0 - np.exp(-self.decay * x)

call_option

call_option(n=None, *, pricing_method=CARR_MADAN, cos_moneyness_std_precision=12, max_moneyness=1.5, max_frequency=None, alpha=None, simpson_rule=False, use_fft=False)

Price call options via one of the available Fourier-based methods, as a function of log-strike

PARAMETER	DESCRIPTION
n	Number of discretization points for the transform. Defaults to 128. TYPE: int | None DEFAULT: None
pricing_method	Method to use for option pricing via Fourier transform of the call option price. Defaults to Lewis. TYPE: OptionPricingMethod DEFAULT: CARR_MADAN
cos_moneyness_std_precision	Truncation parameter for COS: the integration interval is set to [-cos_moneyness_std_precisionstd, cos_moneyness_std_precisionstd]. TYPE: float DEFAULT: 12
max_moneyness	Maximum moneyness to calculate prices. The log-strike grid is set to [-max_moneynessstd, max_moneynessstd]. Used by Lewis and Carr & Madan methods only. TYPE: float DEFAULT: 1.5
max_frequency	Maximum frequency for the transform grid. Defaults to frequency_range(). TYPE: float | None DEFAULT: None
alpha	Damping parameter for integrability of the Carr-Madan integrand, it is ignored if not applicable. TYPE: float | None DEFAULT: None
simpson_rule	Use Simpson's rule for integration. Default is False. TYPE: bool DEFAULT: False
use_fft	Use FFT for the transform rather than FRFT. Default is False. TYPE: bool DEFAULT: False

Source code in quantflow/utils/marginal.py

def call_option(
    self,
    n: Annotated[
        int | None,
        Doc("Number of discretization points for the transform. Defaults to 128."),
    ] = None,
    *,
    pricing_method: Annotated[
        OptionPricingMethod,
        Doc(
            "Method to use for option pricing via Fourier transform of the "
            "call option price. Defaults to Lewis."
        ),
    ] = OptionPricingMethod.CARR_MADAN,
    cos_moneyness_std_precision: Annotated[
        float,
        Doc(
            "Truncation parameter for COS: the integration interval is set to "
            "[-cos_moneyness_std_precision*std, cos_moneyness_std_precision*std]."
        ),
    ] = 12,
    max_moneyness: Annotated[
        float,
        Doc(
            "Maximum moneyness to calculate prices. The log-strike grid is set to "
            "[-max_moneyness*std, max_moneyness*std]. "
            "Used by Lewis and Carr & Madan methods only."
        ),
    ] = 1.5,
    max_frequency: Annotated[
        float | None,
        Doc(
            "Maximum frequency for the transform grid. "
            "Defaults to frequency_range()."
        ),
    ] = None,
    alpha: Annotated[
        float | None,
        Doc(
            "Damping parameter for integrability of the Carr-Madan integrand, "
            "it is ignored if not applicable."
        ),
    ] = None,
    simpson_rule: Annotated[
        bool, Doc("Use Simpson's rule for integration. Default is False.")
    ] = False,
    use_fft: Annotated[
        bool, Doc("Use FFT for the transform rather than FRFT. Default is False.")
    ] = False,
) -> OptionPricingResult:
    """Price call options via one of the available Fourier-based methods,
    as a function of log-strike
    """
    match pricing_method:
        case OptionPricingMethod.COS:
            return self.call_option_cos(
                n,
                moneyness_std_precision=cos_moneyness_std_precision,
            )
        case OptionPricingMethod.CARR_MADAN:
            return self.call_option_carr_madan(
                n,
                max_frequency=max_frequency,
                max_moneyness=max_moneyness,
                alpha=alpha,
                simpson_rule=simpson_rule,
                use_fft=use_fft,
            )
        case _:
            return self.call_option_lewis(
                n,
                max_frequency=max_frequency,
                max_moneyness=max_moneyness,
                simpson_rule=simpson_rule,
                use_fft=use_fft,
            )

call_option_carr_madan

call_option_carr_madan(n=None, *, max_moneyness=1.5, max_frequency=None, alpha=None, simpson_rule=False, use_fft=False)

Call option price via Carr & Madan method

PARAMETER	DESCRIPTION
n	Number of discretization points for the transform. Defaults to 128. TYPE: int | None DEFAULT: None
max_moneyness	Maximum moneyness to calculate prices. The log-strike grid is set to [-max_moneynessstd, max_moneynessstd]. TYPE: float DEFAULT: 1.5
max_frequency	Maximum frequency for the transform grid. Defaults to frequency_range(). TYPE: float | None DEFAULT: None
alpha	Damping parameter for integrability of the Carr-Madan integrand. Defaults to call_option_carr_madan_alpha. TYPE: float | None DEFAULT: None
simpson_rule	Use Simpson's rule for integration. Default is False. TYPE: bool DEFAULT: False
use_fft	Use FFT for the transform. Default is False. TYPE: bool DEFAULT: False

Source code in quantflow/utils/marginal.py

def call_option_carr_madan(
    self,
    n: Annotated[
        int | None,
        Doc("Number of discretization points for the transform. Defaults to 128."),
    ] = None,
    *,
    max_moneyness: Annotated[
        float,
        Doc(
            "Maximum moneyness to calculate prices. The log-strike grid is set to "
            "[-max_moneyness*std, max_moneyness*std]. "
        ),
    ] = 1.5,
    max_frequency: Annotated[
        float | None,
        Doc(
            "Maximum frequency for the transform grid. "
            "Defaults to frequency_range()."
        ),
    ] = None,
    alpha: Annotated[
        float | None,
        Doc(
            "Damping parameter for integrability of the Carr-Madan integrand. "
            "Defaults to "
            "[call_option_carr_madan_alpha][..call_option_carr_madan_alpha]."
        ),
    ] = None,
    simpson_rule: Annotated[
        bool, Doc("Use Simpson's rule for integration. Default is False.")
    ] = False,
    use_fft: Annotated[
        bool, Doc("Use FFT for the transform. Default is False.")
    ] = False,
) -> OptionPricingTransformResult:
    """Call option price via Carr & Madan method"""
    max_log_strike = max_moneyness * self.std_validated()
    transform = self.get_transform(
        n,
        lambda m: self.option_support(m + 1, max_log_strike=max_log_strike),
        max_frequency=max_frequency,
        simpson_rule=simpson_rule,
        use_fft=use_fft,
    )
    alpha = alpha or self.call_option_carr_madan_alpha()
    phi = cast(
        np.ndarray,
        self.call_option_transform(transform.frequency_domain - 1j * alpha),
    )
    result = transform(phi, use_fft=use_fft)
    return OptionPricingTransformResult(
        log_strikes=result.x,
        call_prices=result.y * np.exp(-alpha * result.x),
    )

call_option_carr_madan_alpha

call_option_carr_madan_alpha()

Option alpha to use for Carr & Madan transform to ensure integrability of the call option transform.

Defaults to 1.5, the value suggested in the original Carr & Madan paper.

Source code in quantflow/utils/marginal.py

def call_option_carr_madan_alpha(self) -> float:
    """Option alpha to use for Carr & Madan transform to ensure integrability
    of the call option transform.

    Defaults to 1.5, the value suggested in the original Carr & Madan paper.
    """
    return 1.5

call_option_cos

call_option_cos(n=None, *, moneyness_std_precision=12)

Call option price via the COS method (Fang & Oosterlee 2008).

The call price at log-strike \(k\) is approximated by the cosine series

\[\begin{equation} C(k) \approx e^k \sum_{j=0}^{N-1}{}^{\prime} \mathrm{Re}\!\left[ \phi\!\left(\frac{j\pi}{b-a}\right) e^{-i j\pi (k+a)/(b-a)} \right] V_j \end{equation}\]

where the prime denotes a half weight on the \(j=0\) term, \([a,b]\) is the truncation interval, and \(V_j\) are the cosine payoff coefficients for the normalised call payoff \((e^y - 1)^+\) integrated over \([0, b]\). The \(e^k\) factor converts from strike-normalised to forward-space pricing.

Returns an OptionPricingCosResult with the precomputed coefficient vector. Use call_price to evaluate at arbitrary log-strikes in \(O(N)\) per strike.

PARAMETER	DESCRIPTION
n	Number of cosine series terms. Defaults to 128. TYPE: int | None DEFAULT: None
moneyness_std_precision	Truncation parameter: the integration interval is set to [-moneyness_std_precisionstd, moneyness_std_precisionstd]. TYPE: float DEFAULT: 12

Source code in quantflow/utils/marginal.py

def call_option_cos(
    self,
    n: Annotated[
        int | None,
        Doc("Number of cosine series terms. Defaults to 128."),
    ] = None,
    *,
    moneyness_std_precision: Annotated[
        float,
        Doc(
            "Truncation parameter: the integration interval is set to "
            "[-moneyness_std_precision*std, moneyness_std_precision*std]."
        ),
    ] = 12,
) -> OptionPricingCosResult:
    r"""Call option price via the COS method (Fang & Oosterlee 2008).

    The call price at log-strike $k$ is approximated by the cosine series

    \begin{equation}
        C(k) \approx e^k \sum_{j=0}^{N-1}{}^{\prime}
            \mathrm{Re}\!\left[
                \phi\!\left(\frac{j\pi}{b-a}\right)
                e^{-i j\pi (k+a)/(b-a)}
            \right] V_j
    \end{equation}

    where the prime denotes a half weight on the $j=0$ term, $[a,b]$ is the
    truncation interval, and $V_j$ are the cosine payoff coefficients for the
    normalised call payoff $(e^y - 1)^+$ integrated over $[0, b]$.
    The $e^k$ factor converts from strike-normalised to forward-space pricing.

    Returns an
    [OptionPricingCosResult][quantflow.utils.marginal.OptionPricingCosResult]
    with the precomputed coefficient vector. Use
    [call_price][quantflow.utils.marginal.OptionPricingCosResult.call_price]
    to evaluate at arbitrary log-strikes in $O(N)$ per strike.
    """
    n = n or 128
    std = self.std_validated()
    a = -moneyness_std_precision * std
    b = moneyness_std_precision * std
    bma = b - a

    j = np.arange(n)
    nu = j * np.pi / bma

    phi = np.asarray(self.characteristic_corrected(nu))

    sin_nu_a = np.sin(nu * a)
    cos_nu_a = np.cos(nu * a)
    sign_j = (-1.0) ** j
    safe_nu = np.where(j == 0, 1.0, nu)
    chi = (np.exp(b) * sign_j - cos_nu_a + nu * sin_nu_a) / np.where(
        j == 0, 1.0, 1.0 + nu**2
    )
    psi = np.where(j == 0, b, sin_nu_a / safe_nu)
    V = 2.0 / bma * (chi - psi)
    weights = np.ones(n)
    weights[0] = 0.5
    return OptionPricingCosResult(
        a=a,
        nu=nu,
        coeff=weights * phi * V,
    )

call_option_lewis

call_option_lewis(n=None, *, max_moneyness=1.5, max_frequency=None, simpson_rule=False, use_fft=False)

Call option price via the Lewis (2001) formula

PARAMETER	DESCRIPTION
n	Number of discretization points for the transform. Defaults to 128. TYPE: int | None DEFAULT: None
max_moneyness	Maximum moneyness to calculate prices. The log-strike grid is set to [-max_moneynessstd, max_moneynessstd]. TYPE: float DEFAULT: 1.5
max_frequency	Maximum frequency for the transform grid. Defaults to frequency_range(). TYPE: float | None DEFAULT: None
simpson_rule	Use Simpson's rule for integration. Default is False. TYPE: bool DEFAULT: False
use_fft	Use FFT for the transform. Default is False. TYPE: bool DEFAULT: False

Source code in quantflow/utils/marginal.py

def call_option_lewis(
    self,
    n: Annotated[
        int | None,
        Doc("Number of discretization points for the transform. Defaults to 128."),
    ] = None,
    *,
    max_moneyness: Annotated[
        float,
        Doc(
            "Maximum moneyness to calculate prices. The log-strike grid is set to "
            "[-max_moneyness*std, max_moneyness*std]. "
        ),
    ] = 1.5,
    max_frequency: Annotated[
        float | None,
        Doc(
            "Maximum frequency for the transform grid. "
            "Defaults to frequency_range()."
        ),
    ] = None,
    simpson_rule: Annotated[
        bool, Doc("Use Simpson's rule for integration. Default is False.")
    ] = False,
    use_fft: Annotated[
        bool, Doc("Use FFT for the transform. Default is False.")
    ] = False,
) -> OptionPricingTransformResult:
    """Call option price via the Lewis (2001) formula"""
    max_log_strike = max_moneyness * self.std_validated()
    transform = self.get_transform(
        n,
        lambda m: self.option_support(m + 1, max_log_strike=max_log_strike),
        max_frequency=max_frequency,
        simpson_rule=simpson_rule,
        use_fft=use_fft,
    )
    phi = cast(np.ndarray, self.lewis_transform(transform.frequency_domain))
    result = transform(phi, use_fft=use_fft)
    k = result.x
    return OptionPricingTransformResult(
        log_strikes=k,
        call_prices=1.0 - np.exp(0.5 * k) * result.y,
    )

call_option_transform

call_option_transform(u)

Call option transform

PARAMETER	DESCRIPTION
u	Frequency domain points (possibly complex-shifted). TYPE: Vector

Source code in quantflow/utils/marginal.py

def call_option_transform(
    self,
    u: Annotated[
        Vector, Doc("Frequency domain points (possibly complex-shifted).")
    ],
) -> Vector:
    """Call option transform"""
    uj = 1j * u
    return self.characteristic_corrected(u - 1j) / (uj * uj + uj)

cdf_from_characteristic

cdf_from_characteristic(n=None, *, max_frequency=None, simpson_rule=False, use_fft=False, frequency_n=None)

PARAMETER	DESCRIPTION
n	Number of discretization points for the transform. Defaults to 128. TYPE: int | None DEFAULT: None
max_frequency	Maximum frequency for the transform grid. Defaults to frequency_range(). TYPE: float | None DEFAULT: None
simpson_rule	Use Simpson's rule for integration. Default is False. TYPE: bool DEFAULT: False
use_fft	Use FFT for the transform. Default is False. TYPE: bool DEFAULT: False
frequency_n	Number of points for the frequency grid. Overrides n if provided. TYPE: int | None DEFAULT: None

Source code in quantflow/utils/marginal.py

def cdf_from_characteristic(
    self,
    n: Annotated[
        int | None,
        Doc("Number of discretization points for the transform. Defaults to 128."),
    ] = None,
    *,
    max_frequency: Annotated[
        float | None,
        Doc(
            "Maximum frequency for the transform grid. "
            "Defaults to frequency_range()."
        ),
    ] = None,
    simpson_rule: Annotated[
        bool, Doc("Use Simpson's rule for integration. Default is False.")
    ] = False,
    use_fft: Annotated[
        bool, Doc("Use FFT for the transform. Default is False.")
    ] = False,
    frequency_n: Annotated[
        int | None,
        Doc("Number of points for the frequency grid. Overrides n if provided."),
    ] = None,
) -> TransformResult:
    raise NotImplementedError("CDF not available")

cdf_jacobian

cdf_jacobian(x)

Jacobian of the cdf with respect to the parameters of the process. It is useful for optimization purposes if necessary.

Optional to implement, otherwise raises NotImplementedError if called.

PARAMETER	DESCRIPTION
x	Location in the state space of the process. TYPE: FloatArrayLike

Source code in quantflow/utils/marginal.py

def cdf_jacobian(
    self,
    x: Annotated[
        FloatArrayLike, Doc("Location in the state space of the process.")
    ],
) -> np.ndarray:
    """
    Jacobian of the cdf with respect to the parameters of the process.
    It is useful for optimization purposes if necessary.

    Optional to implement, otherwise raises ``NotImplementedError`` if called.
    """
    raise NotImplementedError("Analytical CDF Jacobian not available")

characteristic_corrected

characteristic_corrected(u)

Characteristic function corrected for the convexity of the log-price distribution

PARAMETER	DESCRIPTION
u	Frequency domain points. TYPE: Vector

Source code in quantflow/utils/marginal.py

def characteristic_corrected(
    self,
    u: Annotated[Vector, Doc("Frequency domain points.")],
) -> Vector:
    """Characteristic function corrected for the convexity of the log-price
    distribution"""
    convexity = np.log(self.characteristic(-1j))
    return self.characteristic(u) * np.exp(-1j * u * convexity)

characteristic_df

characteristic_df(n=None, *, max_frequency=None, simpson_rule=False)

Compute the characteristic function with n discretization points and a max frequency

PARAMETER	DESCRIPTION
n	Number of discretization points for the transform. Defaults to 128. TYPE: int | None DEFAULT: None
max_frequency	Maximum frequency for the transform grid. Defaults to frequency_range(). TYPE: float | None DEFAULT: None
simpson_rule	Use Simpson's rule for integration. Default is False. TYPE: bool DEFAULT: False

Source code in quantflow/utils/marginal.py

def characteristic_df(
    self,
    n: Annotated[
        int | None,
        Doc("Number of discretization points for the transform. Defaults to 128."),
    ] = None,
    *,
    max_frequency: Annotated[
        float | None,
        Doc(
            "Maximum frequency for the transform grid. "
            "Defaults to frequency_range()."
        ),
    ] = None,
    simpson_rule: Annotated[
        bool, Doc("Use Simpson's rule for integration. Default is False.")
    ] = False,
) -> pd.DataFrame:
    """
    Compute the characteristic function with n discretization points
    and a max frequency
    """
    transform = self.get_transform(
        n,
        self.support,
        max_frequency=max_frequency,
        simpson_rule=simpson_rule,
    )
    psi = self.characteristic(transform.frequency_domain)
    return transform.characteristic_df(cast(np.ndarray, psi))

domain_range

domain_range()

The space domain range for the random variable

This should be overloaded if required

Source code in quantflow/utils/marginal.py

def domain_range(self) -> Bounds:
    """The space domain range for the random variable

    This should be overloaded if required
    """
    return default_bounds()

frequency_range

frequency_range(max_frequency=None)

The frequency domain range for the characteristic function

This should be overloaded if required

PARAMETER	DESCRIPTION
max_frequency	Upper bound of the frequency grid. Defaults to 20 if None. TYPE: float | None DEFAULT: None

Source code in quantflow/utils/marginal.py

def frequency_range(
    self,
    max_frequency: Annotated[
        float | None,
        Doc("Upper bound of the frequency grid. Defaults to 20 if None."),
    ] = None,
) -> float:
    """The frequency domain range for the characteristic function

    This should be overloaded if required
    """
    return Bounds(0, max_frequency or 20)

get_transform

get_transform(n, support, *, max_frequency=None, simpson_rule=False, use_fft=False)

PARAMETER	DESCRIPTION
n	Number of discretization points. Defaults to 128. TYPE: int | None
support	Function returning the space domain grid given the number of points. TYPE: Callable[[int], FloatArray]
max_frequency	Maximum frequency for the transform grid. Defaults to frequency_range(). TYPE: float | None DEFAULT: None
simpson_rule	Use Simpson's rule for integration. Default is False. TYPE: bool DEFAULT: False
use_fft	Use FFT for the transform. Default is False. TYPE: bool DEFAULT: False

Source code in quantflow/utils/marginal.py

def get_transform(
    self,
    n: Annotated[
        int | None,
        Doc("Number of discretization points. Defaults to 128."),
    ],
    support: Annotated[
        Callable[[int], FloatArray],
        Doc("Function returning the space domain grid given the number of points."),
    ],
    *,
    max_frequency: Annotated[
        float | None,
        Doc(
            "Maximum frequency for the transform grid. "
            "Defaults to frequency_range()."
        ),
    ] = None,
    simpson_rule: Annotated[
        bool, Doc("Use Simpson's rule for integration. Default is False.")
    ] = False,
    use_fft: Annotated[
        bool, Doc("Use FFT for the transform. Default is False.")
    ] = False,
) -> Transform:
    n = n or 128
    if use_fft:
        bounds = self.domain_range()
    else:
        x = support(n)
        bounds = Bounds(float(np.min(x)), float(np.max(x)))
    return Transform.create(
        n,
        frequency_range=self.frequency_range(max_frequency),
        domain_range=bounds,
        simpson_rule=simpson_rule,
    )

lewis_transform

lewis_transform(u)

Lewis (2001) call option transform - no damping parameter required

PARAMETER	DESCRIPTION
u	Frequency domain points. TYPE: Vector

Source code in quantflow/utils/marginal.py

def lewis_transform(
    self,
    u: Annotated[Vector, Doc("Frequency domain points.")],
) -> Vector:
    """Lewis (2001) call option transform - no damping parameter required"""
    return self.characteristic_corrected(u - 0.5j) / (u * u + 0.25)

mean_from_characteristic

mean_from_characteristic(*, d=0.001)

Calculate mean as first derivative of characteristic function at 0

PARAMETER	DESCRIPTION
d	Step size for finite-difference approximation of the derivative. TYPE: float DEFAULT: 0.001

Source code in quantflow/utils/marginal.py

def mean_from_characteristic(
    self,
    *,
    d: Annotated[
        float,
        Doc("Step size for finite-difference approximation of the derivative."),
    ] = 0.001,
) -> FloatArrayLike:
    """Calculate mean as first derivative of characteristic function at 0"""
    m = -0.5 * 1j * (self.characteristic(d) - self.characteristic(-d)) / d
    return m.real

option_support

option_support(points=101, max_log_strike=1.0)

Compute the x axis.

PARAMETER	DESCRIPTION
points	Number of support points. TYPE: int DEFAULT: 101
max_log_strike	Maximum absolute log-strike. TYPE: float DEFAULT: 1.0

Source code in quantflow/utils/marginal.py

def option_support(
    self,
    points: Annotated[int, Doc("Number of support points.")] = 101,
    max_log_strike: Annotated[float, Doc("Maximum absolute log-strike.")] = 1.0,
) -> FloatArray:
    """
    Compute the x axis.
    """
    return np.linspace(-max_log_strike, max_log_strike, points)

option_time_value

option_time_value(n=128, *, max_frequency=None, max_log_strike=1, alpha=1.1, simpson_rule=False, use_fft=False)

Option time value

PARAMETER	DESCRIPTION
n	Number of discretization points for the transform. TYPE: int DEFAULT: 128
max_frequency	Maximum frequency for the transform grid. Defaults to frequency_range(). TYPE: float | None DEFAULT: None
max_log_strike	Maximum absolute log-strike for the output grid. TYPE: float DEFAULT: 1
alpha	Contour shift parameter controlling the integration strip. TYPE: float DEFAULT: 1.1
simpson_rule	Use Simpson's rule for integration. Default is False. TYPE: bool DEFAULT: False
use_fft	Use FFT for the transform. Default is False. TYPE: bool DEFAULT: False

Source code in quantflow/utils/marginal.py

def option_time_value(
    self,
    n: Annotated[
        int,
        Doc("Number of discretization points for the transform."),
    ] = 128,
    *,
    max_frequency: Annotated[
        float | None,
        Doc(
            "Maximum frequency for the transform grid. "
            "Defaults to frequency_range()."
        ),
    ] = None,
    max_log_strike: Annotated[
        float, Doc("Maximum absolute log-strike for the output grid.")
    ] = 1,
    alpha: Annotated[
        float,
        Doc("Contour shift parameter controlling the integration strip."),
    ] = 1.1,
    simpson_rule: Annotated[
        bool, Doc("Use Simpson's rule for integration. Default is False.")
    ] = False,
    use_fft: Annotated[
        bool, Doc("Use FFT for the transform. Default is False.")
    ] = False,
) -> TransformResult:
    """Option time value"""
    transform = self.get_transform(
        n,
        lambda m: self.option_support(m + 1, max_log_strike=max_log_strike),
        max_frequency=max_frequency,
        simpson_rule=simpson_rule,
        use_fft=use_fft,
    )
    phi = cast(
        np.ndarray,
        self.option_time_value_transform(transform.frequency_domain, alpha),
    )
    result = transform(phi, use_fft=use_fft)
    time_value = result.y / np.sinh(alpha * result.x)
    return TransformResult(x=result.x, y=time_value)

option_time_value_transform

option_time_value_transform(u, alpha=1.1)

Option time value transform

This transform does not require any additional correction since the integrand is already bounded for positive and negative moneyness

PARAMETER	DESCRIPTION
u	Frequency domain points. TYPE: Vector
alpha	Contour shift parameter controlling the integration strip. TYPE: float DEFAULT: 1.1

Source code in quantflow/utils/marginal.py

def option_time_value_transform(
    self,
    u: Annotated[Vector, Doc("Frequency domain points.")],
    alpha: Annotated[
        float, Doc("Contour shift parameter controlling the integration strip.")
    ] = 1.1,
) -> Vector:
    """Option time value transform

    This transform does not require any additional correction since
    the integrand is already bounded for positive and negative moneyness"""
    ia = 1j * alpha
    return 0.5 * (
        self._option_time_value_transform(u - ia)
        - self._option_time_value_transform(u + ia)
    )

pdf_from_characteristic

pdf_from_characteristic(n=None, *, max_frequency=None, simpson_rule=False, use_fft=False, frequency_n=None)

Compute the probability density function from the characteristic function.

PARAMETER	DESCRIPTION
n	Number of discretization points to use in the transform. If None, use 128. TYPE: int | None DEFAULT: None
max_frequency	The maximum frequency to use in the transform. If not provided, the value from the frequency_range method is used. Only needed for special cases/testing. TYPE: float | None DEFAULT: None
simpson_rule	Use Simpson's rule for integration. Default is False. TYPE: bool DEFAULT: False
use_fft	Use FFT for the transform. Default is False. TYPE: bool DEFAULT: False
frequency_n	Number of points for the frequency grid. Overrides n if provided. TYPE: int | None DEFAULT: None

Source code in quantflow/utils/marginal.py

def pdf_from_characteristic(
    self,
    n: Annotated[
        int | None,
        Doc(
            "Number of discretization points to use in the transform."
            " If None, use 128."
        ),
    ] = None,
    *,
    max_frequency: Annotated[
        float | None,
        Doc(
            "The maximum frequency to use in the transform. If not provided,"
            " the value from the [frequency_range]"
            "[quantflow.utils.marginal.Marginal1D.frequency_range] method is used."
            " Only needed for special cases/testing."
        ),
    ] = None,
    simpson_rule: Annotated[
        bool, Doc("Use Simpson's rule for integration. Default is False.")
    ] = False,
    use_fft: Annotated[
        bool, Doc("Use FFT for the transform. Default is False.")
    ] = False,
    frequency_n: Annotated[
        int | None,
        Doc("Number of points for the frequency grid. Overrides n if provided."),
    ] = None,
) -> TransformResult:
    """
    Compute the probability density function from the characteristic function.
    """
    transform = self.get_transform(
        frequency_n or n,
        self.support,
        max_frequency=max_frequency,
        simpson_rule=simpson_rule,
        use_fft=use_fft,
    )
    psi = cast(np.ndarray, self.characteristic(transform.frequency_domain))
    return transform(psi, use_fft=use_fft)

pdf_jacobian

pdf_jacobian(x)

Jacobian of the pdf with respect to the parameters of the process. It has a base implementation that computes it from the cdf_jacobian method, but a subclass should overload this method if a more optimized way of computing it is available.

PARAMETER	DESCRIPTION
x	Location in the state space of the process. TYPE: FloatArrayLike

Source code in quantflow/utils/marginal.py

def pdf_jacobian(
    self,
    x: Annotated[
        FloatArrayLike, Doc("Location in the state space of the process.")
    ],
) -> FloatArrayLike:
    """
    Jacobian of the pdf with respect to the parameters of the process.
    It has a base implementation that computes it from the
    [cdf_jacobian][quantflow.utils.marginal.Marginal1D.cdf_jacobian] method,
    but a subclass should overload this method if a
    more optimized way of computing it is available.
    """
    return self.cdf_jacobian(x) - self.cdf_jacobian(x - 1)

std

std()

Standard deviation at a time horizon

Source code in quantflow/utils/marginal.py

def std(self) -> FloatArrayLike:
    """Standard deviation at a time horizon"""
    return np.sqrt(self.variance())

std_validated

std_validated()

Float standard deviation, raising if it is not finite or non-positive.

Used by the Fourier-based pricing methods to set the log-strike grid range; degenerate parameter samples (e.g. during calibration) can produce a non-finite std and would otherwise crash deep in the transform with a confusing error.

Source code in quantflow/utils/marginal.py

def std_validated(self) -> float:
    """Float standard deviation, raising if it is not finite or non-positive.

    Used by the Fourier-based pricing methods to set the log-strike grid
    range; degenerate parameter samples (e.g. during calibration) can
    produce a non-finite std and would otherwise crash deep in the
    transform with a confusing error.
    """
    std = float(self.std())
    if not np.isfinite(std) or std <= 0:
        raise ValueError(
            f"Marginal std is not finite or non-positive: {std!r}; "
            "model parameters are likely invalid"
        )
    return std

std_from_characteristic

std_from_characteristic()

Calculate standard deviation as square root of variance

Source code in quantflow/utils/marginal.py

def std_from_characteristic(self) -> FloatArrayLike:
    """Calculate standard deviation as square root of variance"""
    return np.sqrt(self.variance_from_characteristic())

variance_from_characteristic

variance_from_characteristic(*, d=0.001)

Calculate variance as second derivative of characteristic function at 0

PARAMETER	DESCRIPTION
d	Step size for finite-difference approximation of the derivative. TYPE: float DEFAULT: 0.001

Source code in quantflow/utils/marginal.py

def variance_from_characteristic(
    self,
    *,
    d: Annotated[
        float,
        Doc("Step size for finite-difference approximation of the derivative."),
    ] = 0.001,
) -> FloatArrayLike:
    """Calculate variance as second derivative of characteristic function at 0"""
    c1 = self.characteristic(d)
    c0 = self.characteristic(0)
    c2 = self.characteristic(-d)
    m = -0.5 * 1j * (c1 - c2) / d
    s = -(c1 - 2 * c0 + c2) / (d * d) - m * m
    return s.real

from_variance_and_asymmetry classmethod

from_variance_and_asymmetry(variance, asymmetry)

Create a distribution from variance and asymmetry

Source code in quantflow/utils/distributions.py

@classmethod
def from_variance_and_asymmetry(cls, variance: float, asymmetry: float) -> Self:
    """Create a distribution from variance and asymmetry"""
    raise NotImplementedError

asymmetry

asymmetry()

Asymmetry of the distribution, 0 for symmetric distributions

Source code in quantflow/utils/distributions.py

def asymmetry(self) -> float:
    """Asymmetry of the distribution, 0 for symmetric distributions"""
    raise NotImplementedError

set_variance

set_variance(variance)

Set the variance of the distribution

Source code in quantflow/utils/distributions.py

def set_variance(self, variance: float) -> None:
    """Set the variance of the distribution"""
    raise NotImplementedError

set_asymmetry

set_asymmetry(asymmetry)

Set the asymmetry of the distribution

Source code in quantflow/utils/distributions.py

def set_asymmetry(self, asymmetry: float) -> None:
    """Set the asymmetry of the distribution"""
    raise NotImplementedError

quantflow.utils.distributions.DoubleExponential pydantic-model

Bases: Exponential

The generalized double exponential distribution

This is also know as the Asymmetric Laplace distribution which is a continuous probability distribution with PDF

\[\begin{align} f(x) &= \frac{\lambda}{\kappa + \frac{1}{\kappa}} e^{-\left(x - m\right) \lambda s(x) \kappa^{s(x)}}\\ s(x) &= {\tt sgn}\left({x - m}\right) \end{align}\]

where \(m\) is the loc parameter, \(\lambda\) is the decay parameter, and \(\kappa\) is the asymmetric parameter.

The Asymmetric Laplace distribution is similar to the Gaussian/normal distribution, but is sharper at the peak, it has fatter tails and allow for skewness. It represents the difference between two independent, exponential random variables.

Fields:

• loc (float)
• decay (float)
• kappa (float)

loc pydantic-field

loc = 0

location parameter \(m\) of the distribution

decay pydantic-field

decay = 1

exponential decay rate \(\lambda\)

kappa pydantic-field

kappa = 1

asymmetric parameter \(\kappa\) - when \(\kappa=1\), the distribution is symmetric

log_kappa property

log_kappa

The log of the kappa parameter

scale property

scale

The scale parameter, it is the inverse of the decay rate

scale2 property

scale2

from_variance_and_asymmetry classmethod

from_variance_and_asymmetry(variance, asymmetry)

Source code in quantflow/utils/distributions.py

@classmethod
def from_variance_and_asymmetry(cls, variance: float, asymmetry: float) -> Self:
    return cls.from_moments(variance=variance, kappa=np.exp(asymmetry))

from_moments classmethod

from_moments(*, mean=0, variance=1, kappa=1)

Create a double exponential distribution from the mean, variance and asymmetry

PARAMETER	DESCRIPTION
mean	The mean of the distribution TYPE: float DEFAULT: 0
variance	The variance of the distribution TYPE: float DEFAULT: 1
kappa	The asymmetry parameter of the distribution, 1 for symmetric TYPE: float DEFAULT: 1

Source code in quantflow/utils/distributions.py

@classmethod
def from_moments(
    cls,
    *,
    mean: Annotated[float, Doc("The mean of the distribution")] = 0,
    variance: Annotated[float, Doc("The variance of the distribution")] = 1,
    kappa: Annotated[
        float, Doc("The asymmetry parameter of the distribution, 1 for symmetric")
    ] = 1,
) -> Self:
    r"""Create a double exponential distribution from the mean, variance
    and asymmetry"""
    k2 = kappa * kappa
    decay = np.sqrt((1 + k2 * k2) / (variance * k2))
    return cls(decay=decay, kappa=kappa, loc=mean - (1 - k2) / (kappa * decay))

characteristic

characteristic(u)

Characteristic function of the double exponential distribution

\[\begin{equation} \Phi(u) = \frac{e^{i u m}}{\left(1 + \frac{i u \kappa}{\lambda}\right) \left(1 - \frac{i u}{\lambda \kappa}\right)} \end{equation}\]

Source code in quantflow/utils/distributions.py

def characteristic(self, u: Vector) -> Vector:
    r"""Characteristic function of the double exponential distribution

    \begin{equation}
    \Phi(u) = \frac{e^{i u m}}{\left(1 + \frac{i u \kappa}{\lambda}\right)
        \left(1 - \frac{i u}{\lambda \kappa}\right)}
    \end{equation}
    """
    den = (1.0 + 1j * u * self.kappa / self.decay) * (
        1.0 - 1j * u / (self.kappa * self.decay)
    )
    return np.exp(1j * u * self.loc) / den

mean

mean()

The mean of the double exponential distribution

Source code in quantflow/utils/distributions.py

def mean(self) -> float:
    r"""The mean of the double exponential distribution"""
    return stats.laplace_asymmetric.mean(self.kappa, loc=self.loc, scale=self.scale)

variance

variance()

Source code in quantflow/utils/distributions.py

def variance(self) -> float:
    return stats.laplace_asymmetric.var(self.kappa, loc=self.loc, scale=self.scale)

pdf

pdf(x)

The analytical PDF as defined above

Source code in quantflow/utils/distributions.py

def pdf(self, x: FloatArrayLike) -> FloatArrayLike:
    """The analytical PDF as defined above"""
    return stats.laplace_asymmetric.pdf(
        x, self.kappa, loc=self.loc, scale=self.scale
    )

sample

sample(n)

Sample from the double exponential distribution

Source code in quantflow/utils/distributions.py

def sample(self, n: int) -> np.ndarray:
    """Sample from the double exponential distribution"""
    return stats.laplace_asymmetric.rvs(
        self.kappa, loc=self.loc, scale=self.scale, size=n
    )

support

support(points=100, *, std_mult=4)

Source code in quantflow/utils/distributions.py

def support(self, points: int = 100, *, std_mult: float = 4) -> FloatArray:
    return np.linspace(
        self.mean() - std_mult * self.std(),
        self.mean() + std_mult * self.std(),
        points,
    )

asymmetry

asymmetry()

Asymmetry of the distribution

Source code in quantflow/utils/distributions.py

def asymmetry(self) -> float:
    """Asymmetry of the distribution"""
    return np.log(self.kappa)

set_variance

set_variance(variance)

Set the variance of the distribution

Source code in quantflow/utils/distributions.py

def set_variance(self, variance: float) -> None:
    """Set the variance of the distribution"""
    k2 = self.kappa * self.kappa
    self.decay = np.sqrt((1 + k2 * k2) / (variance * k2))

set_asymmetry

set_asymmetry(asymmetry)

Source code in quantflow/utils/distributions.py

def set_asymmetry(self, asymmetry: float) -> None:
    self.kappa = np.exp(asymmetry)

call_option

call_option(n=None, *, pricing_method=CARR_MADAN, cos_moneyness_std_precision=12, max_moneyness=1.5, max_frequency=None, alpha=None, simpson_rule=False, use_fft=False)

Price call options via one of the available Fourier-based methods, as a function of log-strike

PARAMETER	DESCRIPTION
n	Number of discretization points for the transform. Defaults to 128. TYPE: int | None DEFAULT: None
pricing_method	Method to use for option pricing via Fourier transform of the call option price. Defaults to Lewis. TYPE: OptionPricingMethod DEFAULT: CARR_MADAN
cos_moneyness_std_precision	Truncation parameter for COS: the integration interval is set to [-cos_moneyness_std_precisionstd, cos_moneyness_std_precisionstd]. TYPE: float DEFAULT: 12
max_moneyness	Maximum moneyness to calculate prices. The log-strike grid is set to [-max_moneynessstd, max_moneynessstd]. Used by Lewis and Carr & Madan methods only. TYPE: float DEFAULT: 1.5
max_frequency	Maximum frequency for the transform grid. Defaults to frequency_range(). TYPE: float | None DEFAULT: None
alpha	Damping parameter for integrability of the Carr-Madan integrand, it is ignored if not applicable. TYPE: float | None DEFAULT: None
simpson_rule	Use Simpson's rule for integration. Default is False. TYPE: bool DEFAULT: False
use_fft	Use FFT for the transform rather than FRFT. Default is False. TYPE: bool DEFAULT: False

Source code in quantflow/utils/marginal.py

def call_option(
    self,
    n: Annotated[
        int | None,
        Doc("Number of discretization points for the transform. Defaults to 128."),
    ] = None,
    *,
    pricing_method: Annotated[
        OptionPricingMethod,
        Doc(
            "Method to use for option pricing via Fourier transform of the "
            "call option price. Defaults to Lewis."
        ),
    ] = OptionPricingMethod.CARR_MADAN,
    cos_moneyness_std_precision: Annotated[
        float,
        Doc(
            "Truncation parameter for COS: the integration interval is set to "
            "[-cos_moneyness_std_precision*std, cos_moneyness_std_precision*std]."
        ),
    ] = 12,
    max_moneyness: Annotated[
        float,
        Doc(
            "Maximum moneyness to calculate prices. The log-strike grid is set to "
            "[-max_moneyness*std, max_moneyness*std]. "
            "Used by Lewis and Carr & Madan methods only."
        ),
    ] = 1.5,
    max_frequency: Annotated[
        float | None,
        Doc(
            "Maximum frequency for the transform grid. "
            "Defaults to frequency_range()."
        ),
    ] = None,
    alpha: Annotated[
        float | None,
        Doc(
            "Damping parameter for integrability of the Carr-Madan integrand, "
            "it is ignored if not applicable."
        ),
    ] = None,
    simpson_rule: Annotated[
        bool, Doc("Use Simpson's rule for integration. Default is False.")
    ] = False,
    use_fft: Annotated[
        bool, Doc("Use FFT for the transform rather than FRFT. Default is False.")
    ] = False,
) -> OptionPricingResult:
    """Price call options via one of the available Fourier-based methods,
    as a function of log-strike
    """
    match pricing_method:
        case OptionPricingMethod.COS:
            return self.call_option_cos(
                n,
                moneyness_std_precision=cos_moneyness_std_precision,
            )
        case OptionPricingMethod.CARR_MADAN:
            return self.call_option_carr_madan(
                n,
                max_frequency=max_frequency,
                max_moneyness=max_moneyness,
                alpha=alpha,
                simpson_rule=simpson_rule,
                use_fft=use_fft,
            )
        case _:
            return self.call_option_lewis(
                n,
                max_frequency=max_frequency,
                max_moneyness=max_moneyness,
                simpson_rule=simpson_rule,
                use_fft=use_fft,
            )

call_option_carr_madan

call_option_carr_madan(n=None, *, max_moneyness=1.5, max_frequency=None, alpha=None, simpson_rule=False, use_fft=False)

Call option price via Carr & Madan method

PARAMETER	DESCRIPTION
n	Number of discretization points for the transform. Defaults to 128. TYPE: int | None DEFAULT: None
max_moneyness	Maximum moneyness to calculate prices. The log-strike grid is set to [-max_moneynessstd, max_moneynessstd]. TYPE: float DEFAULT: 1.5
max_frequency	Maximum frequency for the transform grid. Defaults to frequency_range(). TYPE: float | None DEFAULT: None
alpha	Damping parameter for integrability of the Carr-Madan integrand. Defaults to call_option_carr_madan_alpha. TYPE: float | None DEFAULT: None
simpson_rule	Use Simpson's rule for integration. Default is False. TYPE: bool DEFAULT: False
use_fft	Use FFT for the transform. Default is False. TYPE: bool DEFAULT: False

Source code in quantflow/utils/marginal.py

def call_option_carr_madan(
    self,
    n: Annotated[
        int | None,
        Doc("Number of discretization points for the transform. Defaults to 128."),
    ] = None,
    *,
    max_moneyness: Annotated[
        float,
        Doc(
            "Maximum moneyness to calculate prices. The log-strike grid is set to "
            "[-max_moneyness*std, max_moneyness*std]. "
        ),
    ] = 1.5,
    max_frequency: Annotated[
        float | None,
        Doc(
            "Maximum frequency for the transform grid. "
            "Defaults to frequency_range()."
        ),
    ] = None,
    alpha: Annotated[
        float | None,
        Doc(
            "Damping parameter for integrability of the Carr-Madan integrand. "
            "Defaults to "
            "[call_option_carr_madan_alpha][..call_option_carr_madan_alpha]."
        ),
    ] = None,
    simpson_rule: Annotated[
        bool, Doc("Use Simpson's rule for integration. Default is False.")
    ] = False,
    use_fft: Annotated[
        bool, Doc("Use FFT for the transform. Default is False.")
    ] = False,
) -> OptionPricingTransformResult:
    """Call option price via Carr & Madan method"""
    max_log_strike = max_moneyness * self.std_validated()
    transform = self.get_transform(
        n,
        lambda m: self.option_support(m + 1, max_log_strike=max_log_strike),
        max_frequency=max_frequency,
        simpson_rule=simpson_rule,
        use_fft=use_fft,
    )
    alpha = alpha or self.call_option_carr_madan_alpha()
    phi = cast(
        np.ndarray,
        self.call_option_transform(transform.frequency_domain - 1j * alpha),
    )
    result = transform(phi, use_fft=use_fft)
    return OptionPricingTransformResult(
        log_strikes=result.x,
        call_prices=result.y * np.exp(-alpha * result.x),
    )

call_option_carr_madan_alpha

call_option_carr_madan_alpha()

Option alpha to use for Carr & Madan transform to ensure integrability of the call option transform.

Defaults to 1.5, the value suggested in the original Carr & Madan paper.

Source code in quantflow/utils/marginal.py

def call_option_carr_madan_alpha(self) -> float:
    """Option alpha to use for Carr & Madan transform to ensure integrability
    of the call option transform.

    Defaults to 1.5, the value suggested in the original Carr & Madan paper.
    """
    return 1.5

call_option_cos

call_option_cos(n=None, *, moneyness_std_precision=12)

Call option price via the COS method (Fang & Oosterlee 2008).

The call price at log-strike \(k\) is approximated by the cosine series

\[\begin{equation} C(k) \approx e^k \sum_{j=0}^{N-1}{}^{\prime} \mathrm{Re}\!\left[ \phi\!\left(\frac{j\pi}{b-a}\right) e^{-i j\pi (k+a)/(b-a)} \right] V_j \end{equation}\]

where the prime denotes a half weight on the \(j=0\) term, \([a,b]\) is the truncation interval, and \(V_j\) are the cosine payoff coefficients for the normalised call payoff \((e^y - 1)^+\) integrated over \([0, b]\). The \(e^k\) factor converts from strike-normalised to forward-space pricing.

Returns an OptionPricingCosResult with the precomputed coefficient vector. Use call_price to evaluate at arbitrary log-strikes in \(O(N)\) per strike.

PARAMETER	DESCRIPTION
n	Number of cosine series terms. Defaults to 128. TYPE: int | None DEFAULT: None
moneyness_std_precision	Truncation parameter: the integration interval is set to [-moneyness_std_precisionstd, moneyness_std_precisionstd]. TYPE: float DEFAULT: 12

Source code in quantflow/utils/marginal.py

def call_option_cos(
    self,
    n: Annotated[
        int | None,
        Doc("Number of cosine series terms. Defaults to 128."),
    ] = None,
    *,
    moneyness_std_precision: Annotated[
        float,
        Doc(
            "Truncation parameter: the integration interval is set to "
            "[-moneyness_std_precision*std, moneyness_std_precision*std]."
        ),
    ] = 12,
) -> OptionPricingCosResult:
    r"""Call option price via the COS method (Fang & Oosterlee 2008).

    The call price at log-strike $k$ is approximated by the cosine series

    \begin{equation}
        C(k) \approx e^k \sum_{j=0}^{N-1}{}^{\prime}
            \mathrm{Re}\!\left[
                \phi\!\left(\frac{j\pi}{b-a}\right)
                e^{-i j\pi (k+a)/(b-a)}
            \right] V_j
    \end{equation}

    where the prime denotes a half weight on the $j=0$ term, $[a,b]$ is the
    truncation interval, and $V_j$ are the cosine payoff coefficients for the
    normalised call payoff $(e^y - 1)^+$ integrated over $[0, b]$.
    The $e^k$ factor converts from strike-normalised to forward-space pricing.

    Returns an
    [OptionPricingCosResult][quantflow.utils.marginal.OptionPricingCosResult]
    with the precomputed coefficient vector. Use
    [call_price][quantflow.utils.marginal.OptionPricingCosResult.call_price]
    to evaluate at arbitrary log-strikes in $O(N)$ per strike.
    """
    n = n or 128
    std = self.std_validated()
    a = -moneyness_std_precision * std
    b = moneyness_std_precision * std
    bma = b - a

    j = np.arange(n)
    nu = j * np.pi / bma

    phi = np.asarray(self.characteristic_corrected(nu))

    sin_nu_a = np.sin(nu * a)
    cos_nu_a = np.cos(nu * a)
    sign_j = (-1.0) ** j
    safe_nu = np.where(j == 0, 1.0, nu)
    chi = (np.exp(b) * sign_j - cos_nu_a + nu * sin_nu_a) / np.where(
        j == 0, 1.0, 1.0 + nu**2
    )
    psi = np.where(j == 0, b, sin_nu_a / safe_nu)
    V = 2.0 / bma * (chi - psi)
    weights = np.ones(n)
    weights[0] = 0.5
    return OptionPricingCosResult(
        a=a,
        nu=nu,
        coeff=weights * phi * V,
    )

call_option_lewis

call_option_lewis(n=None, *, max_moneyness=1.5, max_frequency=None, simpson_rule=False, use_fft=False)

Call option price via the Lewis (2001) formula

PARAMETER	DESCRIPTION
n	Number of discretization points for the transform. Defaults to 128. TYPE: int | None DEFAULT: None
max_moneyness	Maximum moneyness to calculate prices. The log-strike grid is set to [-max_moneynessstd, max_moneynessstd]. TYPE: float DEFAULT: 1.5
max_frequency	Maximum frequency for the transform grid. Defaults to frequency_range(). TYPE: float | None DEFAULT: None
simpson_rule	Use Simpson's rule for integration. Default is False. TYPE: bool DEFAULT: False
use_fft	Use FFT for the transform. Default is False. TYPE: bool DEFAULT: False

Source code in quantflow/utils/marginal.py

def call_option_lewis(
    self,
    n: Annotated[
        int | None,
        Doc("Number of discretization points for the transform. Defaults to 128."),
    ] = None,
    *,
    max_moneyness: Annotated[
        float,
        Doc(
            "Maximum moneyness to calculate prices. The log-strike grid is set to "
            "[-max_moneyness*std, max_moneyness*std]. "
        ),
    ] = 1.5,
    max_frequency: Annotated[
        float | None,
        Doc(
            "Maximum frequency for the transform grid. "
            "Defaults to frequency_range()."
        ),
    ] = None,
    simpson_rule: Annotated[
        bool, Doc("Use Simpson's rule for integration. Default is False.")
    ] = False,
    use_fft: Annotated[
        bool, Doc("Use FFT for the transform. Default is False.")
    ] = False,
) -> OptionPricingTransformResult:
    """Call option price via the Lewis (2001) formula"""
    max_log_strike = max_moneyness * self.std_validated()
    transform = self.get_transform(
        n,
        lambda m: self.option_support(m + 1, max_log_strike=max_log_strike),
        max_frequency=max_frequency,
        simpson_rule=simpson_rule,
        use_fft=use_fft,
    )
    phi = cast(np.ndarray, self.lewis_transform(transform.frequency_domain))
    result = transform(phi, use_fft=use_fft)
    k = result.x
    return OptionPricingTransformResult(
        log_strikes=k,
        call_prices=1.0 - np.exp(0.5 * k) * result.y,
    )

call_option_transform

call_option_transform(u)

Call option transform

PARAMETER	DESCRIPTION
u	Frequency domain points (possibly complex-shifted). TYPE: Vector

Source code in quantflow/utils/marginal.py

def call_option_transform(
    self,
    u: Annotated[
        Vector, Doc("Frequency domain points (possibly complex-shifted).")
    ],
) -> Vector:
    """Call option transform"""
    uj = 1j * u
    return self.characteristic_corrected(u - 1j) / (uj * uj + uj)

cdf

cdf(x)

The analytical CDF of the exponential distribution

\[ F(x) = 1 - e^{-\lambda x}\ \ \forall x \geq 0 \]

Source code in quantflow/utils/distributions.py

def cdf(self, x: FloatArrayLike) -> FloatArrayLike:
    r"""The analytical CDF of the exponential distribution

    $$
    F(x) = 1 - e^{-\lambda x}\ \ \forall x \geq 0
    $$
    """
    return 1.0 - np.exp(-self.decay * x)

cdf_from_characteristic

cdf_from_characteristic(n=None, *, max_frequency=None, simpson_rule=False, use_fft=False, frequency_n=None)

PARAMETER	DESCRIPTION
n	Number of discretization points for the transform. Defaults to 128. TYPE: int | None DEFAULT: None
max_frequency	Maximum frequency for the transform grid. Defaults to frequency_range(). TYPE: float | None DEFAULT: None
simpson_rule	Use Simpson's rule for integration. Default is False. TYPE: bool DEFAULT: False
use_fft	Use FFT for the transform. Default is False. TYPE: bool DEFAULT: False
frequency_n	Number of points for the frequency grid. Overrides n if provided. TYPE: int | None DEFAULT: None

Source code in quantflow/utils/marginal.py

def cdf_from_characteristic(
    self,
    n: Annotated[
        int | None,
        Doc("Number of discretization points for the transform. Defaults to 128."),
    ] = None,
    *,
    max_frequency: Annotated[
        float | None,
        Doc(
            "Maximum frequency for the transform grid. "
            "Defaults to frequency_range()."
        ),
    ] = None,
    simpson_rule: Annotated[
        bool, Doc("Use Simpson's rule for integration. Default is False.")
    ] = False,
    use_fft: Annotated[
        bool, Doc("Use FFT for the transform. Default is False.")
    ] = False,
    frequency_n: Annotated[
        int | None,
        Doc("Number of points for the frequency grid. Overrides n if provided."),
    ] = None,
) -> TransformResult:
    raise NotImplementedError("CDF not available")

cdf_jacobian

cdf_jacobian(x)

Jacobian of the cdf with respect to the parameters of the process. It is useful for optimization purposes if necessary.

Optional to implement, otherwise raises NotImplementedError if called.

PARAMETER	DESCRIPTION
x	Location in the state space of the process. TYPE: FloatArrayLike

Source code in quantflow/utils/marginal.py

def cdf_jacobian(
    self,
    x: Annotated[
        FloatArrayLike, Doc("Location in the state space of the process.")
    ],
) -> np.ndarray:
    """
    Jacobian of the cdf with respect to the parameters of the process.
    It is useful for optimization purposes if necessary.

    Optional to implement, otherwise raises ``NotImplementedError`` if called.
    """
    raise NotImplementedError("Analytical CDF Jacobian not available")

characteristic_corrected

characteristic_corrected(u)

Characteristic function corrected for the convexity of the log-price distribution

PARAMETER	DESCRIPTION
u	Frequency domain points. TYPE: Vector

Source code in quantflow/utils/marginal.py

def characteristic_corrected(
    self,
    u: Annotated[Vector, Doc("Frequency domain points.")],
) -> Vector:
    """Characteristic function corrected for the convexity of the log-price
    distribution"""
    convexity = np.log(self.characteristic(-1j))
    return self.characteristic(u) * np.exp(-1j * u * convexity)

characteristic_df

characteristic_df(n=None, *, max_frequency=None, simpson_rule=False)

Compute the characteristic function with n discretization points and a max frequency

PARAMETER	DESCRIPTION
n	Number of discretization points for the transform. Defaults to 128. TYPE: int | None DEFAULT: None
max_frequency	Maximum frequency for the transform grid. Defaults to frequency_range(). TYPE: float | None DEFAULT: None
simpson_rule	Use Simpson's rule for integration. Default is False. TYPE: bool DEFAULT: False

Source code in quantflow/utils/marginal.py

def characteristic_df(
    self,
    n: Annotated[
        int | None,
        Doc("Number of discretization points for the transform. Defaults to 128."),
    ] = None,
    *,
    max_frequency: Annotated[
        float | None,
        Doc(
            "Maximum frequency for the transform grid. "
            "Defaults to frequency_range()."
        ),
    ] = None,
    simpson_rule: Annotated[
        bool, Doc("Use Simpson's rule for integration. Default is False.")
    ] = False,
) -> pd.DataFrame:
    """
    Compute the characteristic function with n discretization points
    and a max frequency
    """
    transform = self.get_transform(
        n,
        self.support,
        max_frequency=max_frequency,
        simpson_rule=simpson_rule,
    )
    psi = self.characteristic(transform.frequency_domain)
    return transform.characteristic_df(cast(np.ndarray, psi))

domain_range

domain_range()

The space domain range for the random variable

This should be overloaded if required

Source code in quantflow/utils/marginal.py

def domain_range(self) -> Bounds:
    """The space domain range for the random variable

    This should be overloaded if required
    """
    return default_bounds()

frequency_range

frequency_range(max_frequency=None)

The frequency domain range for the characteristic function

This should be overloaded if required

PARAMETER	DESCRIPTION
max_frequency	Upper bound of the frequency grid. Defaults to 20 if None. TYPE: float | None DEFAULT: None

Source code in quantflow/utils/marginal.py

def frequency_range(
    self,
    max_frequency: Annotated[
        float | None,
        Doc("Upper bound of the frequency grid. Defaults to 20 if None."),
    ] = None,
) -> float:
    """The frequency domain range for the characteristic function

    This should be overloaded if required
    """
    return Bounds(0, max_frequency or 20)

get_transform

get_transform(n, support, *, max_frequency=None, simpson_rule=False, use_fft=False)

PARAMETER	DESCRIPTION
n	Number of discretization points. Defaults to 128. TYPE: int | None
support	Function returning the space domain grid given the number of points. TYPE: Callable[[int], FloatArray]
max_frequency	Maximum frequency for the transform grid. Defaults to frequency_range(). TYPE: float | None DEFAULT: None
simpson_rule	Use Simpson's rule for integration. Default is False. TYPE: bool DEFAULT: False
use_fft	Use FFT for the transform. Default is False. TYPE: bool DEFAULT: False

Source code in quantflow/utils/marginal.py

def get_transform(
    self,
    n: Annotated[
        int | None,
        Doc("Number of discretization points. Defaults to 128."),
    ],
    support: Annotated[
        Callable[[int], FloatArray],
        Doc("Function returning the space domain grid given the number of points."),
    ],
    *,
    max_frequency: Annotated[
        float | None,
        Doc(
            "Maximum frequency for the transform grid. "
            "Defaults to frequency_range()."
        ),
    ] = None,
    simpson_rule: Annotated[
        bool, Doc("Use Simpson's rule for integration. Default is False.")
    ] = False,
    use_fft: Annotated[
        bool, Doc("Use FFT for the transform. Default is False.")
    ] = False,
) -> Transform:
    n = n or 128
    if use_fft:
        bounds = self.domain_range()
    else:
        x = support(n)
        bounds = Bounds(float(np.min(x)), float(np.max(x)))
    return Transform.create(
        n,
        frequency_range=self.frequency_range(max_frequency),
        domain_range=bounds,
        simpson_rule=simpson_rule,
    )

lewis_transform

lewis_transform(u)

Lewis (2001) call option transform - no damping parameter required

PARAMETER	DESCRIPTION
u	Frequency domain points. TYPE: Vector

Source code in quantflow/utils/marginal.py

def lewis_transform(
    self,
    u: Annotated[Vector, Doc("Frequency domain points.")],
) -> Vector:
    """Lewis (2001) call option transform - no damping parameter required"""
    return self.characteristic_corrected(u - 0.5j) / (u * u + 0.25)

mean_from_characteristic

mean_from_characteristic(*, d=0.001)

Calculate mean as first derivative of characteristic function at 0

PARAMETER	DESCRIPTION
d	Step size for finite-difference approximation of the derivative. TYPE: float DEFAULT: 0.001

Source code in quantflow/utils/marginal.py

def mean_from_characteristic(
    self,
    *,
    d: Annotated[
        float,
        Doc("Step size for finite-difference approximation of the derivative."),
    ] = 0.001,
) -> FloatArrayLike:
    """Calculate mean as first derivative of characteristic function at 0"""
    m = -0.5 * 1j * (self.characteristic(d) - self.characteristic(-d)) / d
    return m.real

option_support

option_support(points=101, max_log_strike=1.0)

Compute the x axis.

PARAMETER	DESCRIPTION
points	Number of support points. TYPE: int DEFAULT: 101
max_log_strike	Maximum absolute log-strike. TYPE: float DEFAULT: 1.0

Source code in quantflow/utils/marginal.py

def option_support(
    self,
    points: Annotated[int, Doc("Number of support points.")] = 101,
    max_log_strike: Annotated[float, Doc("Maximum absolute log-strike.")] = 1.0,
) -> FloatArray:
    """
    Compute the x axis.
    """
    return np.linspace(-max_log_strike, max_log_strike, points)

option_time_value

option_time_value(n=128, *, max_frequency=None, max_log_strike=1, alpha=1.1, simpson_rule=False, use_fft=False)

Option time value

PARAMETER	DESCRIPTION
n	Number of discretization points for the transform. TYPE: int DEFAULT: 128
max_frequency	Maximum frequency for the transform grid. Defaults to frequency_range(). TYPE: float | None DEFAULT: None
max_log_strike	Maximum absolute log-strike for the output grid. TYPE: float DEFAULT: 1
alpha	Contour shift parameter controlling the integration strip. TYPE: float DEFAULT: 1.1
simpson_rule	Use Simpson's rule for integration. Default is False. TYPE: bool DEFAULT: False
use_fft	Use FFT for the transform. Default is False. TYPE: bool DEFAULT: False

Source code in quantflow/utils/marginal.py

def option_time_value(
    self,
    n: Annotated[
        int,
        Doc("Number of discretization points for the transform."),
    ] = 128,
    *,
    max_frequency: Annotated[
        float | None,
        Doc(
            "Maximum frequency for the transform grid. "
            "Defaults to frequency_range()."
        ),
    ] = None,
    max_log_strike: Annotated[
        float, Doc("Maximum absolute log-strike for the output grid.")
    ] = 1,
    alpha: Annotated[
        float,
        Doc("Contour shift parameter controlling the integration strip."),
    ] = 1.1,
    simpson_rule: Annotated[
        bool, Doc("Use Simpson's rule for integration. Default is False.")
    ] = False,
    use_fft: Annotated[
        bool, Doc("Use FFT for the transform. Default is False.")
    ] = False,
) -> TransformResult:
    """Option time value"""
    transform = self.get_transform(
        n,
        lambda m: self.option_support(m + 1, max_log_strike=max_log_strike),
        max_frequency=max_frequency,
        simpson_rule=simpson_rule,
        use_fft=use_fft,
    )
    phi = cast(
        np.ndarray,
        self.option_time_value_transform(transform.frequency_domain, alpha),
    )
    result = transform(phi, use_fft=use_fft)
    time_value = result.y / np.sinh(alpha * result.x)
    return TransformResult(x=result.x, y=time_value)

option_time_value_transform

option_time_value_transform(u, alpha=1.1)

Option time value transform

This transform does not require any additional correction since the integrand is already bounded for positive and negative moneyness

PARAMETER	DESCRIPTION
u	Frequency domain points. TYPE: Vector
alpha	Contour shift parameter controlling the integration strip. TYPE: float DEFAULT: 1.1

Source code in quantflow/utils/marginal.py

def option_time_value_transform(
    self,
    u: Annotated[Vector, Doc("Frequency domain points.")],
    alpha: Annotated[
        float, Doc("Contour shift parameter controlling the integration strip.")
    ] = 1.1,
) -> Vector:
    """Option time value transform

    This transform does not require any additional correction since
    the integrand is already bounded for positive and negative moneyness"""
    ia = 1j * alpha
    return 0.5 * (
        self._option_time_value_transform(u - ia)
        - self._option_time_value_transform(u + ia)
    )

pdf_from_characteristic

pdf_from_characteristic(n=None, *, max_frequency=None, simpson_rule=False, use_fft=False, frequency_n=None)

Compute the probability density function from the characteristic function.

PARAMETER	DESCRIPTION
n	Number of discretization points to use in the transform. If None, use 128. TYPE: int | None DEFAULT: None
max_frequency	The maximum frequency to use in the transform. If not provided, the value from the frequency_range method is used. Only needed for special cases/testing. TYPE: float | None DEFAULT: None
simpson_rule	Use Simpson's rule for integration. Default is False. TYPE: bool DEFAULT: False
use_fft	Use FFT for the transform. Default is False. TYPE: bool DEFAULT: False
frequency_n	Number of points for the frequency grid. Overrides n if provided. TYPE: int | None DEFAULT: None

Source code in quantflow/utils/marginal.py

def pdf_from_characteristic(
    self,
    n: Annotated[
        int | None,
        Doc(
            "Number of discretization points to use in the transform."
            " If None, use 128."
        ),
    ] = None,
    *,
    max_frequency: Annotated[
        float | None,
        Doc(
            "The maximum frequency to use in the transform. If not provided,"
            " the value from the [frequency_range]"
            "[quantflow.utils.marginal.Marginal1D.frequency_range] method is used."
            " Only needed for special cases/testing."
        ),
    ] = None,
    simpson_rule: Annotated[
        bool, Doc("Use Simpson's rule for integration. Default is False.")
    ] = False,
    use_fft: Annotated[
        bool, Doc("Use FFT for the transform. Default is False.")
    ] = False,
    frequency_n: Annotated[
        int | None,
        Doc("Number of points for the frequency grid. Overrides n if provided."),
    ] = None,
) -> TransformResult:
    """
    Compute the probability density function from the characteristic function.
    """
    transform = self.get_transform(
        frequency_n or n,
        self.support,
        max_frequency=max_frequency,
        simpson_rule=simpson_rule,
        use_fft=use_fft,
    )
    psi = cast(np.ndarray, self.characteristic(transform.frequency_domain))
    return transform(psi, use_fft=use_fft)

pdf_jacobian

pdf_jacobian(x)

Jacobian of the pdf with respect to the parameters of the process. It has a base implementation that computes it from the cdf_jacobian method, but a subclass should overload this method if a more optimized way of computing it is available.

PARAMETER	DESCRIPTION
x	Location in the state space of the process. TYPE: FloatArrayLike

Source code in quantflow/utils/marginal.py

def pdf_jacobian(
    self,
    x: Annotated[
        FloatArrayLike, Doc("Location in the state space of the process.")
    ],
) -> FloatArrayLike:
    """
    Jacobian of the pdf with respect to the parameters of the process.
    It has a base implementation that computes it from the
    [cdf_jacobian][quantflow.utils.marginal.Marginal1D.cdf_jacobian] method,
    but a subclass should overload this method if a
    more optimized way of computing it is available.
    """
    return self.cdf_jacobian(x) - self.cdf_jacobian(x - 1)

std

std()

Standard deviation at a time horizon

Source code in quantflow/utils/marginal.py

def std(self) -> FloatArrayLike:
    """Standard deviation at a time horizon"""
    return np.sqrt(self.variance())

std_validated

std_validated()

Float standard deviation, raising if it is not finite or non-positive.

Used by the Fourier-based pricing methods to set the log-strike grid range; degenerate parameter samples (e.g. during calibration) can produce a non-finite std and would otherwise crash deep in the transform with a confusing error.

Source code in quantflow/utils/marginal.py

def std_validated(self) -> float:
    """Float standard deviation, raising if it is not finite or non-positive.

    Used by the Fourier-based pricing methods to set the log-strike grid
    range; degenerate parameter samples (e.g. during calibration) can
    produce a non-finite std and would otherwise crash deep in the
    transform with a confusing error.
    """
    std = float(self.std())
    if not np.isfinite(std) or std <= 0:
        raise ValueError(
            f"Marginal std is not finite or non-positive: {std!r}; "
            "model parameters are likely invalid"
        )
    return std

std_from_characteristic

std_from_characteristic()

Calculate standard deviation as square root of variance

Source code in quantflow/utils/marginal.py

def std_from_characteristic(self) -> FloatArrayLike:
    """Calculate standard deviation as square root of variance"""
    return np.sqrt(self.variance_from_characteristic())

variance_from_characteristic

variance_from_characteristic(*, d=0.001)

Calculate variance as second derivative of characteristic function at 0

PARAMETER	DESCRIPTION
d	Step size for finite-difference approximation of the derivative. TYPE: float DEFAULT: 0.001

Source code in quantflow/utils/marginal.py

def variance_from_characteristic(
    self,
    *,
    d: Annotated[
        float,
        Doc("Step size for finite-difference approximation of the derivative."),
    ] = 0.001,
) -> FloatArrayLike:
    """Calculate variance as second derivative of characteristic function at 0"""
    c1 = self.characteristic(d)
    c0 = self.characteristic(0)
    c2 = self.characteristic(-d)
    m = -0.5 * 1j * (c1 - c2) / d
    s = -(c1 - 2 * c0 + c2) / (d * d) - m * m
    return s.real

quantflow.utils.distributions.Normal pydantic-model

Bases: Distribution1D

The normal distribution is a continuous probability distribution with PDF given by

\[\begin{equation} f(x) = \frac{e^{-\frac{\left(x - \mu\right)^2}{2\sigma^2}}}{\sqrt{2\pi\sigma^2}} \end{equation}\]

Fields:

• mu (float)
• sigma (float)

mu pydantic-field

mu = 0

mean of the normal distribution

sigma pydantic-field

sigma = 1

standard deviation of the normal distribution

sigma2 property

sigma2

from_variance_and_asymmetry classmethod

from_variance_and_asymmetry(variance, asymmetry)

The normal distribution is symmetric, so the asymmetry is ignored

Source code in quantflow/utils/distributions.py

@classmethod
def from_variance_and_asymmetry(cls, variance: float, asymmetry: float) -> Self:
    """The normal distribution is symmetric, so the asymmetry is ignored"""
    return cls(mu=0, sigma=np.sqrt(variance))

characteristic

characteristic(u)

Source code in quantflow/utils/distributions.py

def characteristic(self, u: Vector) -> Vector:
    return np.exp(complex(0, 1) * u * self.mu - self.sigma2 * u**2 / 2)

mean

mean()

Source code in quantflow/utils/distributions.py

def mean(self) -> float:
    return self.mu

variance

variance()

Source code in quantflow/utils/distributions.py

def variance(self) -> float:
    return self.sigma2

sample

sample(n)

Source code in quantflow/utils/distributions.py

def sample(self, n: int) -> np.ndarray:
    return np.random.normal(loc=self.mu, scale=self.sigma, size=n)

support

support(points=100, *, std_mult=4)

Source code in quantflow/utils/distributions.py

def support(self, points: int = 100, *, std_mult: float = 4) -> FloatArray:
    return np.linspace(
        self.mu - std_mult * self.sigma,
        self.mu + std_mult * self.sigma,
        points,
    )

set_variance

set_variance(variance)

Set the variance of the distribution

Source code in quantflow/utils/distributions.py

def set_variance(self, variance: float) -> None:
    """Set the variance of the distribution"""
    self.sigma = np.sqrt(variance)

call_option

call_option(n=None, *, pricing_method=CARR_MADAN, cos_moneyness_std_precision=12, max_moneyness=1.5, max_frequency=None, alpha=None, simpson_rule=False, use_fft=False)

Price call options via one of the available Fourier-based methods, as a function of log-strike

PARAMETER	DESCRIPTION
n	Number of discretization points for the transform. Defaults to 128. TYPE: int | None DEFAULT: None
pricing_method	Method to use for option pricing via Fourier transform of the call option price. Defaults to Lewis. TYPE: OptionPricingMethod DEFAULT: CARR_MADAN
cos_moneyness_std_precision	Truncation parameter for COS: the integration interval is set to [-cos_moneyness_std_precisionstd, cos_moneyness_std_precisionstd]. TYPE: float DEFAULT: 12
max_moneyness	Maximum moneyness to calculate prices. The log-strike grid is set to [-max_moneynessstd, max_moneynessstd]. Used by Lewis and Carr & Madan methods only. TYPE: float DEFAULT: 1.5
max_frequency	Maximum frequency for the transform grid. Defaults to frequency_range(). TYPE: float | None DEFAULT: None
alpha	Damping parameter for integrability of the Carr-Madan integrand, it is ignored if not applicable. TYPE: float | None DEFAULT: None
simpson_rule	Use Simpson's rule for integration. Default is False. TYPE: bool DEFAULT: False
use_fft	Use FFT for the transform rather than FRFT. Default is False. TYPE: bool DEFAULT: False

Source code in quantflow/utils/marginal.py

def call_option(
    self,
    n: Annotated[
        int | None,
        Doc("Number of discretization points for the transform. Defaults to 128."),
    ] = None,
    *,
    pricing_method: Annotated[
        OptionPricingMethod,
        Doc(
            "Method to use for option pricing via Fourier transform of the "
            "call option price. Defaults to Lewis."
        ),
    ] = OptionPricingMethod.CARR_MADAN,
    cos_moneyness_std_precision: Annotated[
        float,
        Doc(
            "Truncation parameter for COS: the integration interval is set to "
            "[-cos_moneyness_std_precision*std, cos_moneyness_std_precision*std]."
        ),
    ] = 12,
    max_moneyness: Annotated[
        float,
        Doc(
            "Maximum moneyness to calculate prices. The log-strike grid is set to "
            "[-max_moneyness*std, max_moneyness*std]. "
            "Used by Lewis and Carr & Madan methods only."
        ),
    ] = 1.5,
    max_frequency: Annotated[
        float | None,
        Doc(
            "Maximum frequency for the transform grid. "
            "Defaults to frequency_range()."
        ),
    ] = None,
    alpha: Annotated[
        float | None,
        Doc(
            "Damping parameter for integrability of the Carr-Madan integrand, "
            "it is ignored if not applicable."
        ),
    ] = None,
    simpson_rule: Annotated[
        bool, Doc("Use Simpson's rule for integration. Default is False.")
    ] = False,
    use_fft: Annotated[
        bool, Doc("Use FFT for the transform rather than FRFT. Default is False.")
    ] = False,
) -> OptionPricingResult:
    """Price call options via one of the available Fourier-based methods,
    as a function of log-strike
    """
    match pricing_method:
        case OptionPricingMethod.COS:
            return self.call_option_cos(
                n,
                moneyness_std_precision=cos_moneyness_std_precision,
            )
        case OptionPricingMethod.CARR_MADAN:
            return self.call_option_carr_madan(
                n,
                max_frequency=max_frequency,
                max_moneyness=max_moneyness,
                alpha=alpha,
                simpson_rule=simpson_rule,
                use_fft=use_fft,
            )
        case _:
            return self.call_option_lewis(
                n,
                max_frequency=max_frequency,
                max_moneyness=max_moneyness,
                simpson_rule=simpson_rule,
                use_fft=use_fft,
            )

call_option_carr_madan

call_option_carr_madan(n=None, *, max_moneyness=1.5, max_frequency=None, alpha=None, simpson_rule=False, use_fft=False)

Call option price via Carr & Madan method

PARAMETER	DESCRIPTION
n	Number of discretization points for the transform. Defaults to 128. TYPE: int | None DEFAULT: None
max_moneyness	Maximum moneyness to calculate prices. The log-strike grid is set to [-max_moneynessstd, max_moneynessstd]. TYPE: float DEFAULT: 1.5
max_frequency	Maximum frequency for the transform grid. Defaults to frequency_range(). TYPE: float | None DEFAULT: None
alpha	Damping parameter for integrability of the Carr-Madan integrand. Defaults to call_option_carr_madan_alpha. TYPE: float | None DEFAULT: None
simpson_rule	Use Simpson's rule for integration. Default is False. TYPE: bool DEFAULT: False
use_fft	Use FFT for the transform. Default is False. TYPE: bool DEFAULT: False

Source code in quantflow/utils/marginal.py

def call_option_carr_madan(
    self,
    n: Annotated[
        int | None,
        Doc("Number of discretization points for the transform. Defaults to 128."),
    ] = None,
    *,
    max_moneyness: Annotated[
        float,
        Doc(
            "Maximum moneyness to calculate prices. The log-strike grid is set to "
            "[-max_moneyness*std, max_moneyness*std]. "
        ),
    ] = 1.5,
    max_frequency: Annotated[
        float | None,
        Doc(
            "Maximum frequency for the transform grid. "
            "Defaults to frequency_range()."
        ),
    ] = None,
    alpha: Annotated[
        float | None,
        Doc(
            "Damping parameter for integrability of the Carr-Madan integrand. "
            "Defaults to "
            "[call_option_carr_madan_alpha][..call_option_carr_madan_alpha]."
        ),
    ] = None,
    simpson_rule: Annotated[
        bool, Doc("Use Simpson's rule for integration. Default is False.")
    ] = False,
    use_fft: Annotated[
        bool, Doc("Use FFT for the transform. Default is False.")
    ] = False,
) -> OptionPricingTransformResult:
    """Call option price via Carr & Madan method"""
    max_log_strike = max_moneyness * self.std_validated()
    transform = self.get_transform(
        n,
        lambda m: self.option_support(m + 1, max_log_strike=max_log_strike),
        max_frequency=max_frequency,
        simpson_rule=simpson_rule,
        use_fft=use_fft,
    )
    alpha = alpha or self.call_option_carr_madan_alpha()
    phi = cast(
        np.ndarray,
        self.call_option_transform(transform.frequency_domain - 1j * alpha),
    )
    result = transform(phi, use_fft=use_fft)
    return OptionPricingTransformResult(
        log_strikes=result.x,
        call_prices=result.y * np.exp(-alpha * result.x),
    )

call_option_carr_madan_alpha

call_option_carr_madan_alpha()

Option alpha to use for Carr & Madan transform to ensure integrability of the call option transform.

Defaults to 1.5, the value suggested in the original Carr & Madan paper.

Source code in quantflow/utils/marginal.py

def call_option_carr_madan_alpha(self) -> float:
    """Option alpha to use for Carr & Madan transform to ensure integrability
    of the call option transform.

    Defaults to 1.5, the value suggested in the original Carr & Madan paper.
    """
    return 1.5

call_option_cos

call_option_cos(n=None, *, moneyness_std_precision=12)

Call option price via the COS method (Fang & Oosterlee 2008).

The call price at log-strike \(k\) is approximated by the cosine series

\[\begin{equation} C(k) \approx e^k \sum_{j=0}^{N-1}{}^{\prime} \mathrm{Re}\!\left[ \phi\!\left(\frac{j\pi}{b-a}\right) e^{-i j\pi (k+a)/(b-a)} \right] V_j \end{equation}\]

where the prime denotes a half weight on the \(j=0\) term, \([a,b]\) is the truncation interval, and \(V_j\) are the cosine payoff coefficients for the normalised call payoff \((e^y - 1)^+\) integrated over \([0, b]\). The \(e^k\) factor converts from strike-normalised to forward-space pricing.

Returns an OptionPricingCosResult with the precomputed coefficient vector. Use call_price to evaluate at arbitrary log-strikes in \(O(N)\) per strike.

PARAMETER	DESCRIPTION
n	Number of cosine series terms. Defaults to 128. TYPE: int | None DEFAULT: None
moneyness_std_precision	Truncation parameter: the integration interval is set to [-moneyness_std_precisionstd, moneyness_std_precisionstd]. TYPE: float DEFAULT: 12

Source code in quantflow/utils/marginal.py

def call_option_cos(
    self,
    n: Annotated[
        int | None,
        Doc("Number of cosine series terms. Defaults to 128."),
    ] = None,
    *,
    moneyness_std_precision: Annotated[
        float,
        Doc(
            "Truncation parameter: the integration interval is set to "
            "[-moneyness_std_precision*std, moneyness_std_precision*std]."
        ),
    ] = 12,
) -> OptionPricingCosResult:
    r"""Call option price via the COS method (Fang & Oosterlee 2008).

    The call price at log-strike $k$ is approximated by the cosine series

    \begin{equation}
        C(k) \approx e^k \sum_{j=0}^{N-1}{}^{\prime}
            \mathrm{Re}\!\left[
                \phi\!\left(\frac{j\pi}{b-a}\right)
                e^{-i j\pi (k+a)/(b-a)}
            \right] V_j
    \end{equation}

    where the prime denotes a half weight on the $j=0$ term, $[a,b]$ is the
    truncation interval, and $V_j$ are the cosine payoff coefficients for the
    normalised call payoff $(e^y - 1)^+$ integrated over $[0, b]$.
    The $e^k$ factor converts from strike-normalised to forward-space pricing.

    Returns an
    [OptionPricingCosResult][quantflow.utils.marginal.OptionPricingCosResult]
    with the precomputed coefficient vector. Use
    [call_price][quantflow.utils.marginal.OptionPricingCosResult.call_price]
    to evaluate at arbitrary log-strikes in $O(N)$ per strike.
    """
    n = n or 128
    std = self.std_validated()
    a = -moneyness_std_precision * std
    b = moneyness_std_precision * std
    bma = b - a

    j = np.arange(n)
    nu = j * np.pi / bma

    phi = np.asarray(self.characteristic_corrected(nu))

    sin_nu_a = np.sin(nu * a)
    cos_nu_a = np.cos(nu * a)
    sign_j = (-1.0) ** j
    safe_nu = np.where(j == 0, 1.0, nu)
    chi = (np.exp(b) * sign_j - cos_nu_a + nu * sin_nu_a) / np.where(
        j == 0, 1.0, 1.0 + nu**2
    )
    psi = np.where(j == 0, b, sin_nu_a / safe_nu)
    V = 2.0 / bma * (chi - psi)
    weights = np.ones(n)
    weights[0] = 0.5
    return OptionPricingCosResult(
        a=a,
        nu=nu,
        coeff=weights * phi * V,
    )

call_option_lewis

call_option_lewis(n=None, *, max_moneyness=1.5, max_frequency=None, simpson_rule=False, use_fft=False)

Call option price via the Lewis (2001) formula

PARAMETER	DESCRIPTION
n	Number of discretization points for the transform. Defaults to 128. TYPE: int | None DEFAULT: None
max_moneyness	Maximum moneyness to calculate prices. The log-strike grid is set to [-max_moneynessstd, max_moneynessstd]. TYPE: float DEFAULT: 1.5
max_frequency	Maximum frequency for the transform grid. Defaults to frequency_range(). TYPE: float | None DEFAULT: None
simpson_rule	Use Simpson's rule for integration. Default is False. TYPE: bool DEFAULT: False
use_fft	Use FFT for the transform. Default is False. TYPE: bool DEFAULT: False

Source code in quantflow/utils/marginal.py

def call_option_lewis(
    self,
    n: Annotated[
        int | None,
        Doc("Number of discretization points for the transform. Defaults to 128."),
    ] = None,
    *,
    max_moneyness: Annotated[
        float,
        Doc(
            "Maximum moneyness to calculate prices. The log-strike grid is set to "
            "[-max_moneyness*std, max_moneyness*std]. "
        ),
    ] = 1.5,
    max_frequency: Annotated[
        float | None,
        Doc(
            "Maximum frequency for the transform grid. "
            "Defaults to frequency_range()."
        ),
    ] = None,
    simpson_rule: Annotated[
        bool, Doc("Use Simpson's rule for integration. Default is False.")
    ] = False,
    use_fft: Annotated[
        bool, Doc("Use FFT for the transform. Default is False.")
    ] = False,
) -> OptionPricingTransformResult:
    """Call option price via the Lewis (2001) formula"""
    max_log_strike = max_moneyness * self.std_validated()
    transform = self.get_transform(
        n,
        lambda m: self.option_support(m + 1, max_log_strike=max_log_strike),
        max_frequency=max_frequency,
        simpson_rule=simpson_rule,
        use_fft=use_fft,
    )
    phi = cast(np.ndarray, self.lewis_transform(transform.frequency_domain))
    result = transform(phi, use_fft=use_fft)
    k = result.x
    return OptionPricingTransformResult(
        log_strikes=k,
        call_prices=1.0 - np.exp(0.5 * k) * result.y,
    )

call_option_transform

call_option_transform(u)

Call option transform

PARAMETER	DESCRIPTION
u	Frequency domain points (possibly complex-shifted). TYPE: Vector

Source code in quantflow/utils/marginal.py

def call_option_transform(
    self,
    u: Annotated[
        Vector, Doc("Frequency domain points (possibly complex-shifted).")
    ],
) -> Vector:
    """Call option transform"""
    uj = 1j * u
    return self.characteristic_corrected(u - 1j) / (uj * uj + uj)

cdf

cdf(x)

Compute the cumulative distribution function

PARAMETER	DESCRIPTION
x	Location in the stochastic process domain space. If a numpy array, the output should have the same shape as the input. TYPE: FloatArrayLike

Source code in quantflow/utils/marginal.py

def cdf(
    self,
    x: Annotated[
        FloatArrayLike,
        Doc(
            "Location in the stochastic process domain space. If a numpy array,"
            " the output should have the same shape as the input."
        ),
    ],
) -> FloatArrayLike:
    """
    Compute the cumulative distribution function
    """
    raise NotImplementedError("Analytical CDF not available")

cdf_from_characteristic

cdf_from_characteristic(n=None, *, max_frequency=None, simpson_rule=False, use_fft=False, frequency_n=None)

PARAMETER	DESCRIPTION
n	Number of discretization points for the transform. Defaults to 128. TYPE: int | None DEFAULT: None
max_frequency	Maximum frequency for the transform grid. Defaults to frequency_range(). TYPE: float | None DEFAULT: None
simpson_rule	Use Simpson's rule for integration. Default is False. TYPE: bool DEFAULT: False
use_fft	Use FFT for the transform. Default is False. TYPE: bool DEFAULT: False
frequency_n	Number of points for the frequency grid. Overrides n if provided. TYPE: int | None DEFAULT: None

Source code in quantflow/utils/marginal.py

def cdf_from_characteristic(
    self,
    n: Annotated[
        int | None,
        Doc("Number of discretization points for the transform. Defaults to 128."),
    ] = None,
    *,
    max_frequency: Annotated[
        float | None,
        Doc(
            "Maximum frequency for the transform grid. "
            "Defaults to frequency_range()."
        ),
    ] = None,
    simpson_rule: Annotated[
        bool, Doc("Use Simpson's rule for integration. Default is False.")
    ] = False,
    use_fft: Annotated[
        bool, Doc("Use FFT for the transform. Default is False.")
    ] = False,
    frequency_n: Annotated[
        int | None,
        Doc("Number of points for the frequency grid. Overrides n if provided."),
    ] = None,
) -> TransformResult:
    raise NotImplementedError("CDF not available")

cdf_jacobian

cdf_jacobian(x)

Jacobian of the cdf with respect to the parameters of the process. It is useful for optimization purposes if necessary.

Optional to implement, otherwise raises NotImplementedError if called.

PARAMETER	DESCRIPTION
x	Location in the state space of the process. TYPE: FloatArrayLike

Source code in quantflow/utils/marginal.py

def cdf_jacobian(
    self,
    x: Annotated[
        FloatArrayLike, Doc("Location in the state space of the process.")
    ],
) -> np.ndarray:
    """
    Jacobian of the cdf with respect to the parameters of the process.
    It is useful for optimization purposes if necessary.

    Optional to implement, otherwise raises ``NotImplementedError`` if called.
    """
    raise NotImplementedError("Analytical CDF Jacobian not available")

characteristic_corrected

characteristic_corrected(u)

Characteristic function corrected for the convexity of the log-price distribution

PARAMETER	DESCRIPTION
u	Frequency domain points. TYPE: Vector

Source code in quantflow/utils/marginal.py

def characteristic_corrected(
    self,
    u: Annotated[Vector, Doc("Frequency domain points.")],
) -> Vector:
    """Characteristic function corrected for the convexity of the log-price
    distribution"""
    convexity = np.log(self.characteristic(-1j))
    return self.characteristic(u) * np.exp(-1j * u * convexity)

characteristic_df

characteristic_df(n=None, *, max_frequency=None, simpson_rule=False)

Compute the characteristic function with n discretization points and a max frequency

PARAMETER	DESCRIPTION
n	Number of discretization points for the transform. Defaults to 128. TYPE: int | None DEFAULT: None
max_frequency	Maximum frequency for the transform grid. Defaults to frequency_range(). TYPE: float | None DEFAULT: None
simpson_rule	Use Simpson's rule for integration. Default is False. TYPE: bool DEFAULT: False

Source code in quantflow/utils/marginal.py

def characteristic_df(
    self,
    n: Annotated[
        int | None,
        Doc("Number of discretization points for the transform. Defaults to 128."),
    ] = None,
    *,
    max_frequency: Annotated[
        float | None,
        Doc(
            "Maximum frequency for the transform grid. "
            "Defaults to frequency_range()."
        ),
    ] = None,
    simpson_rule: Annotated[
        bool, Doc("Use Simpson's rule for integration. Default is False.")
    ] = False,
) -> pd.DataFrame:
    """
    Compute the characteristic function with n discretization points
    and a max frequency
    """
    transform = self.get_transform(
        n,
        self.support,
        max_frequency=max_frequency,
        simpson_rule=simpson_rule,
    )
    psi = self.characteristic(transform.frequency_domain)
    return transform.characteristic_df(cast(np.ndarray, psi))

domain_range

domain_range()

The space domain range for the random variable

This should be overloaded if required

Source code in quantflow/utils/marginal.py

def domain_range(self) -> Bounds:
    """The space domain range for the random variable

    This should be overloaded if required
    """
    return default_bounds()

frequency_range

frequency_range(max_frequency=None)

The frequency domain range for the characteristic function

This should be overloaded if required

PARAMETER	DESCRIPTION
max_frequency	Upper bound of the frequency grid. Defaults to 20 if None. TYPE: float | None DEFAULT: None

Source code in quantflow/utils/marginal.py

def frequency_range(
    self,
    max_frequency: Annotated[
        float | None,
        Doc("Upper bound of the frequency grid. Defaults to 20 if None."),
    ] = None,
) -> float:
    """The frequency domain range for the characteristic function

    This should be overloaded if required
    """
    return Bounds(0, max_frequency or 20)

get_transform

get_transform(n, support, *, max_frequency=None, simpson_rule=False, use_fft=False)

PARAMETER	DESCRIPTION
n	Number of discretization points. Defaults to 128. TYPE: int | None
support	Function returning the space domain grid given the number of points. TYPE: Callable[[int], FloatArray]
max_frequency	Maximum frequency for the transform grid. Defaults to frequency_range(). TYPE: float | None DEFAULT: None
simpson_rule	Use Simpson's rule for integration. Default is False. TYPE: bool DEFAULT: False
use_fft	Use FFT for the transform. Default is False. TYPE: bool DEFAULT: False

Source code in quantflow/utils/marginal.py

def get_transform(
    self,
    n: Annotated[
        int | None,
        Doc("Number of discretization points. Defaults to 128."),
    ],
    support: Annotated[
        Callable[[int], FloatArray],
        Doc("Function returning the space domain grid given the number of points."),
    ],
    *,
    max_frequency: Annotated[
        float | None,
        Doc(
            "Maximum frequency for the transform grid. "
            "Defaults to frequency_range()."
        ),
    ] = None,
    simpson_rule: Annotated[
        bool, Doc("Use Simpson's rule for integration. Default is False.")
    ] = False,
    use_fft: Annotated[
        bool, Doc("Use FFT for the transform. Default is False.")
    ] = False,
) -> Transform:
    n = n or 128
    if use_fft:
        bounds = self.domain_range()
    else:
        x = support(n)
        bounds = Bounds(float(np.min(x)), float(np.max(x)))
    return Transform.create(
        n,
        frequency_range=self.frequency_range(max_frequency),
        domain_range=bounds,
        simpson_rule=simpson_rule,
    )

lewis_transform

lewis_transform(u)

Lewis (2001) call option transform - no damping parameter required

PARAMETER	DESCRIPTION
u	Frequency domain points. TYPE: Vector

Source code in quantflow/utils/marginal.py

def lewis_transform(
    self,
    u: Annotated[Vector, Doc("Frequency domain points.")],
) -> Vector:
    """Lewis (2001) call option transform - no damping parameter required"""
    return self.characteristic_corrected(u - 0.5j) / (u * u + 0.25)

mean_from_characteristic

mean_from_characteristic(*, d=0.001)

Calculate mean as first derivative of characteristic function at 0

PARAMETER	DESCRIPTION
d	Step size for finite-difference approximation of the derivative. TYPE: float DEFAULT: 0.001

Source code in quantflow/utils/marginal.py

def mean_from_characteristic(
    self,
    *,
    d: Annotated[
        float,
        Doc("Step size for finite-difference approximation of the derivative."),
    ] = 0.001,
) -> FloatArrayLike:
    """Calculate mean as first derivative of characteristic function at 0"""
    m = -0.5 * 1j * (self.characteristic(d) - self.characteristic(-d)) / d
    return m.real

option_support

option_support(points=101, max_log_strike=1.0)

Compute the x axis.

PARAMETER	DESCRIPTION
points	Number of support points. TYPE: int DEFAULT: 101
max_log_strike	Maximum absolute log-strike. TYPE: float DEFAULT: 1.0

Source code in quantflow/utils/marginal.py

def option_support(
    self,
    points: Annotated[int, Doc("Number of support points.")] = 101,
    max_log_strike: Annotated[float, Doc("Maximum absolute log-strike.")] = 1.0,
) -> FloatArray:
    """
    Compute the x axis.
    """
    return np.linspace(-max_log_strike, max_log_strike, points)

option_time_value

option_time_value(n=128, *, max_frequency=None, max_log_strike=1, alpha=1.1, simpson_rule=False, use_fft=False)

Option time value

PARAMETER	DESCRIPTION
n	Number of discretization points for the transform. TYPE: int DEFAULT: 128
max_frequency	Maximum frequency for the transform grid. Defaults to frequency_range(). TYPE: float | None DEFAULT: None
max_log_strike	Maximum absolute log-strike for the output grid. TYPE: float DEFAULT: 1
alpha	Contour shift parameter controlling the integration strip. TYPE: float DEFAULT: 1.1
simpson_rule	Use Simpson's rule for integration. Default is False. TYPE: bool DEFAULT: False
use_fft	Use FFT for the transform. Default is False. TYPE: bool DEFAULT: False

Source code in quantflow/utils/marginal.py

def option_time_value(
    self,
    n: Annotated[
        int,
        Doc("Number of discretization points for the transform."),
    ] = 128,
    *,
    max_frequency: Annotated[
        float | None,
        Doc(
            "Maximum frequency for the transform grid. "
            "Defaults to frequency_range()."
        ),
    ] = None,
    max_log_strike: Annotated[
        float, Doc("Maximum absolute log-strike for the output grid.")
    ] = 1,
    alpha: Annotated[
        float,
        Doc("Contour shift parameter controlling the integration strip."),
    ] = 1.1,
    simpson_rule: Annotated[
        bool, Doc("Use Simpson's rule for integration. Default is False.")
    ] = False,
    use_fft: Annotated[
        bool, Doc("Use FFT for the transform. Default is False.")
    ] = False,
) -> TransformResult:
    """Option time value"""
    transform = self.get_transform(
        n,
        lambda m: self.option_support(m + 1, max_log_strike=max_log_strike),
        max_frequency=max_frequency,
        simpson_rule=simpson_rule,
        use_fft=use_fft,
    )
    phi = cast(
        np.ndarray,
        self.option_time_value_transform(transform.frequency_domain, alpha),
    )
    result = transform(phi, use_fft=use_fft)
    time_value = result.y / np.sinh(alpha * result.x)
    return TransformResult(x=result.x, y=time_value)

option_time_value_transform

option_time_value_transform(u, alpha=1.1)

Option time value transform

This transform does not require any additional correction since the integrand is already bounded for positive and negative moneyness

PARAMETER	DESCRIPTION
u	Frequency domain points. TYPE: Vector
alpha	Contour shift parameter controlling the integration strip. TYPE: float DEFAULT: 1.1

Source code in quantflow/utils/marginal.py

def option_time_value_transform(
    self,
    u: Annotated[Vector, Doc("Frequency domain points.")],
    alpha: Annotated[
        float, Doc("Contour shift parameter controlling the integration strip.")
    ] = 1.1,
) -> Vector:
    """Option time value transform

    This transform does not require any additional correction since
    the integrand is already bounded for positive and negative moneyness"""
    ia = 1j * alpha
    return 0.5 * (
        self._option_time_value_transform(u - ia)
        - self._option_time_value_transform(u + ia)
    )

pdf

pdf(x)

Computes the probability density (or mass) function of the process.

It has a base implementation that computes the pdf from the cdf method, but a subclass should overload this method if a more optimized way of computing it is available.

PARAMETER	DESCRIPTION
x	Location in the stochastic process domain space. If a numpy array, the output should have the same shape as the input. TYPE: FloatArrayLike

Source code in quantflow/utils/marginal.py

def pdf(
    self,
    x: Annotated[
        FloatArrayLike,
        Doc(
            "Location in the stochastic process domain space. If a numpy array,"
            " the output should have the same shape as the input."
        ),
    ],
) -> FloatArrayLike:
    """
    Computes the probability density (or mass) function of the process.

    It has a base implementation that computes the pdf from the
    [cdf][quantflow.utils.marginal.Marginal1D.cdf] method, but a subclass should
    overload this method if a
    more optimized way of computing it is available.
    """
    raise NotImplementedError("Analytical PDF not available")

pdf_from_characteristic

pdf_from_characteristic(n=None, *, max_frequency=None, simpson_rule=False, use_fft=False, frequency_n=None)

Compute the probability density function from the characteristic function.

PARAMETER	DESCRIPTION
n	Number of discretization points to use in the transform. If None, use 128. TYPE: int | None DEFAULT: None
max_frequency	The maximum frequency to use in the transform. If not provided, the value from the frequency_range method is used. Only needed for special cases/testing. TYPE: float | None DEFAULT: None
simpson_rule	Use Simpson's rule for integration. Default is False. TYPE: bool DEFAULT: False
use_fft	Use FFT for the transform. Default is False. TYPE: bool DEFAULT: False
frequency_n	Number of points for the frequency grid. Overrides n if provided. TYPE: int | None DEFAULT: None

Source code in quantflow/utils/marginal.py

def pdf_from_characteristic(
    self,
    n: Annotated[
        int | None,
        Doc(
            "Number of discretization points to use in the transform."
            " If None, use 128."
        ),
    ] = None,
    *,
    max_frequency: Annotated[
        float | None,
        Doc(
            "The maximum frequency to use in the transform. If not provided,"
            " the value from the [frequency_range]"
            "[quantflow.utils.marginal.Marginal1D.frequency_range] method is used."
            " Only needed for special cases/testing."
        ),
    ] = None,
    simpson_rule: Annotated[
        bool, Doc("Use Simpson's rule for integration. Default is False.")
    ] = False,
    use_fft: Annotated[
        bool, Doc("Use FFT for the transform. Default is False.")
    ] = False,
    frequency_n: Annotated[
        int | None,
        Doc("Number of points for the frequency grid. Overrides n if provided."),
    ] = None,
) -> TransformResult:
    """
    Compute the probability density function from the characteristic function.
    """
    transform = self.get_transform(
        frequency_n or n,
        self.support,
        max_frequency=max_frequency,
        simpson_rule=simpson_rule,
        use_fft=use_fft,
    )
    psi = cast(np.ndarray, self.characteristic(transform.frequency_domain))
    return transform(psi, use_fft=use_fft)

pdf_jacobian

pdf_jacobian(x)

Jacobian of the pdf with respect to the parameters of the process. It has a base implementation that computes it from the cdf_jacobian method, but a subclass should overload this method if a more optimized way of computing it is available.

PARAMETER	DESCRIPTION
x	Location in the state space of the process. TYPE: FloatArrayLike

Source code in quantflow/utils/marginal.py

def pdf_jacobian(
    self,
    x: Annotated[
        FloatArrayLike, Doc("Location in the state space of the process.")
    ],
) -> FloatArrayLike:
    """
    Jacobian of the pdf with respect to the parameters of the process.
    It has a base implementation that computes it from the
    [cdf_jacobian][quantflow.utils.marginal.Marginal1D.cdf_jacobian] method,
    but a subclass should overload this method if a
    more optimized way of computing it is available.
    """
    return self.cdf_jacobian(x) - self.cdf_jacobian(x - 1)

std

std()

Standard deviation at a time horizon

Source code in quantflow/utils/marginal.py

def std(self) -> FloatArrayLike:
    """Standard deviation at a time horizon"""
    return np.sqrt(self.variance())

std_validated

std_validated()

Float standard deviation, raising if it is not finite or non-positive.

Used by the Fourier-based pricing methods to set the log-strike grid range; degenerate parameter samples (e.g. during calibration) can produce a non-finite std and would otherwise crash deep in the transform with a confusing error.

Source code in quantflow/utils/marginal.py

def std_validated(self) -> float:
    """Float standard deviation, raising if it is not finite or non-positive.

    Used by the Fourier-based pricing methods to set the log-strike grid
    range; degenerate parameter samples (e.g. during calibration) can
    produce a non-finite std and would otherwise crash deep in the
    transform with a confusing error.
    """
    std = float(self.std())
    if not np.isfinite(std) or std <= 0:
        raise ValueError(
            f"Marginal std is not finite or non-positive: {std!r}; "
            "model parameters are likely invalid"
        )
    return std

std_from_characteristic

std_from_characteristic()

Calculate standard deviation as square root of variance

Source code in quantflow/utils/marginal.py

def std_from_characteristic(self) -> FloatArrayLike:
    """Calculate standard deviation as square root of variance"""
    return np.sqrt(self.variance_from_characteristic())

variance_from_characteristic

variance_from_characteristic(*, d=0.001)

Calculate variance as second derivative of characteristic function at 0

PARAMETER	DESCRIPTION
d	Step size for finite-difference approximation of the derivative. TYPE: float DEFAULT: 0.001

Source code in quantflow/utils/marginal.py

def variance_from_characteristic(
    self,
    *,
    d: Annotated[
        float,
        Doc("Step size for finite-difference approximation of the derivative."),
    ] = 0.001,
) -> FloatArrayLike:
    """Calculate variance as second derivative of characteristic function at 0"""
    c1 = self.characteristic(d)
    c0 = self.characteristic(0)
    c2 = self.characteristic(-d)
    m = -0.5 * 1j * (c1 - c2) / d
    s = -(c1 - 2 * c0 + c2) / (d * d) - m * m
    return s.real

asymmetry

asymmetry()

Asymmetry of the distribution, 0 for symmetric distributions

Source code in quantflow/utils/distributions.py

def asymmetry(self) -> float:
    """Asymmetry of the distribution, 0 for symmetric distributions"""
    raise NotImplementedError

set_asymmetry

set_asymmetry(asymmetry)

Set the asymmetry of the distribution

Source code in quantflow/utils/distributions.py

def set_asymmetry(self, asymmetry: float) -> None:
    """Set the asymmetry of the distribution"""
    raise NotImplementedError