Marginal 1D¶

quantflow.utils.marginal.OptionPricingMethod ¶

Bases: StrEnum

Method to use for option pricing via Fourier transform of the call option price

CARR_MADAN `class-attribute` `instance-attribute` ¶

CARR_MADAN = auto()

LEWIS `class-attribute` `instance-attribute` ¶

LEWIS = auto()

COS `class-attribute` `instance-attribute` ¶

COS = auto()

quantflow.utils.marginal.OptionPricingResult `pydantic-model` ¶

Bases: BaseModel, ABC

call_price `abstractmethod` ¶

call_price(log_strikes)

Evaluate call prices at arbitrary log-strikes.

Source code in quantflow/utils/marginal.py

@abstractmethod
def call_price(self, log_strikes: FloatArrayLike) -> FloatArray:
    """Evaluate call prices at arbitrary log-strikes."""

call_greeks `abstractmethod` ¶

call_greeks(log_strike)

Evaluate option price, delta & gamma at a given log-strike.

Source code in quantflow/utils/marginal.py

@abstractmethod
def call_greeks(self, log_strike: float) -> Greeks:
    """Evaluate option price, delta & gamma at a given log-strike."""

quantflow.utils.marginal.OptionPricingCosResult `pydantic-model` ¶

Bases: OptionPricingResult

Result of call option pricing via the COS method.

Stores the precomputed coefficient vector, enabling \(O(N)\) evaluation at any log-strike without recomputing the characteristic function.

Fields:

a (float)
nu (FloatArray)
coeff (ndarray)

a `pydantic-field` ¶

Left endpoint of the truncation interval [a, b]

nu `pydantic-field` ¶

nu

Cosine frequency grid j*pi/(b-a)

coeff `pydantic-field` ¶

coeff

Complex coefficient vector w_j * phi(nu_j) * V_j

call_price ¶

call_price(log_strikes)

Evaluate call prices at arbitrary log-strikes in \(O(N)\) per strike.

PARAMETER	DESCRIPTION
`log_strikes`	Log-strikes at which to evaluate the call price TYPE: `FloatArrayLike`

Source code in quantflow/utils/marginal.py

def call_price(
    self,
    log_strikes: Annotated[
        FloatArrayLike, Doc("Log-strikes at which to evaluate the call price")
    ],
) -> FloatArray:
    """Evaluate call prices at arbitrary log-strikes in $O(N)$ per strike."""
    k = np.asarray(log_strikes)
    return np.real(
        np.exp(-1j * np.outer(k + self.a, self.nu)) @ self.coeff
    ) * np.exp(k)

call_greeks ¶

call_greeks(log_strike)

Analytical price, delta and gamma at a single log-strike.

The COS expansion in \(k\) admits closed-form \(k\)-derivatives: each cosine term picks up an extra factor of \(-i\nu_j\) per derivative. In forward space (\(F=1\), \(k=\log(K/F)\)), the forward-delta and forward-gamma are \(c-c_k\) and \(c_{kk}-c_k\) respectively.

Source code in quantflow/utils/marginal.py

def call_greeks(self, log_strike: float) -> Greeks:
    r"""Analytical price, delta and gamma at a single log-strike.

    The COS expansion in $k$ admits closed-form $k$-derivatives:
    each cosine term picks up an extra factor of $-i\nu_j$ per
    derivative. In forward space ($F=1$, $k=\log(K/F)$), the
    forward-delta and forward-gamma are
    $c-c_k$ and $c_{kk}-c_k$ respectively.
    """
    k = float(log_strike)
    z = np.exp(-1j * self.nu * (k + self.a)) * self.coeff
    s0 = np.real(z.sum())
    s1 = np.real((-1j * self.nu * z).sum())
    s2 = np.real((-(self.nu * self.nu) * z).sum())
    ek = np.exp(k)
    return Greeks(
        price=float(ek * s0),
        delta=float(-ek * s1),
        gamma=float(ek * (s1 + s2)),
    )

quantflow.utils.marginal.Marginal1D `pydantic-model` ¶

Bases: BaseModel, ABC

Abstract 1D marginal distribution with Fourier-based option pricing.

This class represents the marginal distribution. It provides methods to compute the pdf, cdf, and option prices from the characteristic function of the underlying process, as well as their Jacobians with respect to the parameters of the process.

Option prices are computed via Fourier inversion of the characteristic function, using two supported formulas:

Carr & Madan: uses a damping parameter \(\alpha\) to ensure integrability. The integrand is evaluated along a contour shifted by \(\alpha\) in the imaginary direction. See call_option_carr_madan.
Lewis: no damping parameter required. The contour is fixed at imaginary part \(1/2\), giving an integrand that is naturally bounded for all real \(u\). See call_option_lewis.
COS method: uses a Fourier-cosine expansion of the option payoff, with coefficients that depend on the characteristic function evaluated at a cosine frequency grid. See call_option_cos.

It is the base class for the StochasticProcess1DMarginal.

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

Marginal 1D¶

quantflow.utils.marginal.OptionPricingMethod ¶

CARR_MADAN class-attribute instance-attribute ¶

LEWIS class-attribute instance-attribute ¶

COS class-attribute instance-attribute ¶

quantflow.utils.marginal.OptionPricingResult pydantic-model ¶

call_price abstractmethod ¶

call_greeks abstractmethod ¶

quantflow.utils.marginal.OptionPricingCosResult pydantic-model ¶

a pydantic-field ¶

nu pydantic-field ¶

coeff pydantic-field ¶

call_price ¶

call_greeks ¶

quantflow.utils.marginal.Marginal1D pydantic-model ¶

call_option ¶

call_option_carr_madan ¶

call_option_carr_madan_alpha ¶

call_option_cos ¶

call_option_lewis ¶

call_option_transform ¶

cdf ¶

cdf_from_characteristic ¶

cdf_jacobian ¶

characteristic abstractmethod ¶

characteristic_corrected ¶

characteristic_df ¶

domain_range ¶

frequency_range ¶

get_transform ¶

lewis_transform ¶

mean ¶

mean_from_characteristic ¶

option_support ¶

option_time_value ¶

option_time_value_transform ¶

pdf ¶

pdf_from_characteristic ¶

pdf_jacobian ¶

std ¶

std_validated ¶

std_from_characteristic ¶

support abstractmethod ¶

variance ¶

variance_from_characteristic ¶

CARR_MADAN `class-attribute` `instance-attribute` ¶

LEWIS `class-attribute` `instance-attribute` ¶

COS `class-attribute` `instance-attribute` ¶

quantflow.utils.marginal.OptionPricingResult `pydantic-model` ¶

call_price `abstractmethod` ¶

call_greeks `abstractmethod` ¶

quantflow.utils.marginal.OptionPricingCosResult `pydantic-model` ¶

a `pydantic-field` ¶

nu `pydantic-field` ¶

coeff `pydantic-field` ¶

quantflow.utils.marginal.Marginal1D `pydantic-model` ¶

characteristic `abstractmethod` ¶

support `abstractmethod` ¶