Vol Surface

quantflow.options.surface.VolSurface pydantic-model

Bases: BaseModel, Generic[S]

Represents a volatility surface, which captures the implied volatility of an option for different strikes and maturities.

Key Concepts:

• Implied Volatility: The market's expectation of future volatility, derived from the price of an option using a pricing model (e.g., Black-Scholes).
• Strike Price: The price at which the underlying asset can be bought (call option) or sold (put option) at the option's expiry.
• Time to Maturity: The time remaining until the option's expiration date.
• Volatility Smile/Skew: The often-observed phenomenon where implied volatility varies across different strike prices for the same maturity. Typically, it forms a "smile" or "skew" shape.

This class provides a structure for storing and manipulating volatility surface data. It can be used for various tasks, such as:

• Option pricing and risk management: Using the surface to determine the appropriate volatility input for pricing models.
• Volatility arbitrage: Identifying mispricings in options by comparing market prices to model prices derived from the surface.
• Market analysis: Understanding market sentiment and expectations of future volatility.

Fields:

• ref_date (datetime)
• asset (str)
• spot (SpotPrice[S])
• maturities (tuple[VolCrossSection[S], ...])
• day_counter (DayCounter)
• tick_size_forwards (DecimalNumber | None)
• tick_size_options (DecimalNumber | None)

ref_date pydantic-field

ref_date

Reference date for the volatility surface

asset pydantic-field

asset

Underlying asset of the volatility surface

spot pydantic-field

spot

Spot price of the underlying asset

maturities pydantic-field

maturities

Sorted tuple of VolCrossSection, each containing the forward price and option prices for that maturity

day_counter pydantic-field

day_counter = default_day_counter

Day counter for time to maturity calculations, by default it uses Act/Act

tick_size_forwards pydantic-field

tick_size_forwards = None

Tick size for rounding forward and spot prices - optional

tick_size_options pydantic-field

tick_size_options = None

Tick size for rounding option prices - optional

securities

securities(*, select=all, index=None, converged=False)

Iterator over securities in the volatility surface

PARAMETER	DESCRIPTION
select	Option selection method TYPE: OptionSelection DEFAULT: all
index	Index of the cross section to use, if None use all TYPE: int | None DEFAULT: None
converged	Include the spot, forwards and options with implied volatility converged only if True, otherwise include all securities regardless of convergence TYPE: bool DEFAULT: False

Source code in quantflow/options/surface.py

def securities(
    self,
    *,
    select: Annotated[
        OptionSelection, Doc("Option selection method")
    ] = OptionSelection.all,
    index: Annotated[
        int | None, Doc("Index of the cross section to use, if None use all")
    ] = None,
    converged: Annotated[
        bool,
        Doc(
            "Include the spot, forwards and options with implied volatility "
            "converged only if True, otherwise include all securities regardless "
            "of convergence"
        ),
    ] = False,
) -> Iterator[SpotPrice[S] | FwdPrice[S] | OptionPrices[S]]:
    """Iterator over securities in the volatility surface"""
    yield self.spot
    if index is not None:
        yield from self.maturities[index].securities(
            select=select, converged=converged
        )
    else:
        for maturity in self.maturities:
            yield from maturity.securities(select=select, converged=converged)

inputs

inputs(*, select=all, index=None, converged=False)

Convert the volatility surface to a VolSurfaceInputs instance

PARAMETER	DESCRIPTION
select	Option selection method TYPE: OptionSelection DEFAULT: all
index	Index of the cross section to use, if None use all TYPE: int | None DEFAULT: None
converged	Include spot, forwards and options with implied volatility converged only if True, otherwise include all securities regardless of convergence TYPE: bool DEFAULT: False

Source code in quantflow/options/surface.py

def inputs(
    self,
    *,
    select: Annotated[
        OptionSelection, Doc("Option selection method")
    ] = OptionSelection.all,
    index: Annotated[
        int | None, Doc("Index of the cross section to use, if None use all")
    ] = None,
    converged: Annotated[
        bool,
        Doc(
            "Include spot, forwards and options with implied volatility "
            "converged only if True, otherwise include all securities regardless "
            "of convergence"
        ),
    ] = False,
) -> VolSurfaceInputs:
    """Convert the volatility surface to a
    [VolSurfaceInputs][quantflow.options.inputs.VolSurfaceInputs] instance"""
    return VolSurfaceInputs(
        asset=self.asset,
        ref_date=self.ref_date,
        inputs=list(
            s.inputs()
            for s in self.securities(
                select=select, converged=converged, index=index
            )
        ),
    )

term_structure

term_structure()

Return the term structure of the volatility surface as a DataFrame

Source code in quantflow/options/surface.py

def term_structure(self) -> pd.DataFrame:
    """Return the term structure of the volatility surface as a DataFrame"""
    return pd.DataFrame(
        cross.info_dict(self.ref_date, self.spot) for cross in self.maturities
    )

trim

trim(num_maturities)

Create a new volatility surface with the last num_maturities maturities

Source code in quantflow/options/surface.py

def trim(self, num_maturities: int) -> Self:
    """Create a new volatility surface with the last `num_maturities` maturities"""
    return self.model_copy(
        update=dict(maturities=self.maturities[-num_maturities:])
    )

option_prices

option_prices(*, select=best, index=None, initial_vol=INITIAL_VOL, converged=False)

Iterator over selected option prices in the surface

PARAMETER	DESCRIPTION
select	Option selection method TYPE: OptionSelection DEFAULT: best
index	Index of the cross section to use, if None use all TYPE: int | None DEFAULT: None
initial_vol	Initial volatility for the root finding algorithm TYPE: float DEFAULT: INITIAL_VOL
converged	Include options with implied volatility converged only if True, otherwise include all options regardless of convergence TYPE: bool DEFAULT: False

Source code in quantflow/options/surface.py

def option_prices(
    self,
    *,
    select: Annotated[
        OptionSelection, Doc("Option selection method")
    ] = OptionSelection.best,
    index: Annotated[
        int | None, Doc("Index of the cross section to use, if None use all")
    ] = None,
    initial_vol: Annotated[
        float, Doc("Initial volatility for the root finding algorithm")
    ] = INITIAL_VOL,
    converged: Annotated[
        bool,
        Doc(
            "Include options with implied volatility "
            "converged only if True, otherwise include all options regardless "
            "of convergence"
        ),
    ] = False,
) -> Iterator[OptionPrice]:
    """Iterator over selected option prices in the surface"""
    if index is not None:
        yield from self.maturities[index].option_prices(
            self.ref_date,
            select=select,
            initial_vol=initial_vol,
            converged=converged,
        )
    else:
        for maturity in self.maturities:
            yield from maturity.option_prices(
                self.ref_date,
                select=select,
                initial_vol=initial_vol,
                converged=converged,
            )

option_list

option_list(*, select=best, index=None, converged=False)

List of selected option prices in the surface

PARAMETER	DESCRIPTION
select	Option selection method TYPE: OptionSelection DEFAULT: best
index	Index of the cross section to use, if None use all TYPE: int | None DEFAULT: None
converged	Include options with implied volatility converged only if True, otherwise include all options regardless of convergence TYPE: bool DEFAULT: False

Source code in quantflow/options/surface.py

def option_list(
    self,
    *,
    select: Annotated[
        OptionSelection, Doc("Option selection method")
    ] = OptionSelection.best,
    index: Annotated[
        int | None, Doc("Index of the cross section to use, if None use all")
    ] = None,
    converged: Annotated[
        bool,
        Doc(
            "Include options with implied volatility "
            "converged only if True, otherwise include all options regardless "
            "of convergence"
        ),
    ] = False,
) -> list[OptionPrice]:
    "List of selected option prices in the surface"
    return list(self.option_prices(select=select, index=index, converged=converged))

bs

bs(*, select=best, index=None, initial_vol=INITIAL_VOL)

Calculate Black-Scholes implied volatility for options in the surface. For some option prices, the implied volatility calculation may not converge, in this case the implied volatility is not calculated correctly and the option is marked as not converged.

PARAMETER	DESCRIPTION
select	Option selection method TYPE: OptionSelection DEFAULT: best
index	Index of the cross section to use, if None use all TYPE: int | None DEFAULT: None
initial_vol	Initial volatility for the root finding algorithm TYPE: float DEFAULT: INITIAL_VOL

Source code in quantflow/options/surface.py

def bs(
    self,
    *,
    select: Annotated[
        OptionSelection, Doc("Option selection method")
    ] = OptionSelection.best,
    index: Annotated[
        int | None, Doc("Index of the cross section to use, if None use all")
    ] = None,
    initial_vol: Annotated[
        float, Doc("Initial volatility for the root finding algorithm")
    ] = INITIAL_VOL,
) -> list[OptionPrice]:
    """Calculate Black-Scholes implied volatility for options
    in the surface.
    For some option prices, the implied volatility calculation may not converge,
    in this case the implied volatility is not
    calculated correctly and the option is marked as not converged.
    """
    self.reset_convergence()
    d = self.as_array(
        select=select,
        index=index,
        initial_vol=initial_vol,
    )
    with warnings.catch_warnings():
        warnings.simplefilter("ignore")
        result = implied_black_volatility(
            k=d.log_strike,
            price=d.price,
            ttm=d.ttm,
            initial_sigma=d.implied_vol,
            call_put=d.call_put,
        )
    for option, implied_vol, converged in zip(
        d.options, result.values, result.converged
    ):
        option.implied_vol = float(implied_vol)
        option.converged = converged and not np.isnan(implied_vol)
    return d.options

calc_bs_prices

calc_bs_prices(*, select=best, index=None)

calculate Black-Scholes prices for all options in the surface

It uses options with a converged implied volatility calculation only, otherwise the price calculation won't be correct.

PARAMETER	DESCRIPTION
select	Option selection method TYPE: OptionSelection DEFAULT: best
index	Index of the cross section to use, if None use all TYPE: int | None DEFAULT: None

Source code in quantflow/options/surface.py

def calc_bs_prices(
    self,
    *,
    select: Annotated[
        OptionSelection, Doc("Option selection method")
    ] = OptionSelection.best,
    index: Annotated[
        int | None, Doc("Index of the cross section to use, if None use all")
    ] = None,
) -> FloatArray:
    """calculate Black-Scholes prices for all options in the surface

    It uses options with a converged implied volatility calculation only,
    otherwise the price calculation won't be correct.
    """
    d = self.as_array(select=select, index=index, converged=True)
    return black_price(k=d.log_strike, sigma=d.implied_vol, ttm=d.ttm, s=d.call_put)

options_df

options_df(*, select=best, index=None, initial_vol=INITIAL_VOL, converged=False)

Time frame of Black-Scholes call input data

PARAMETER	DESCRIPTION
select	Option selection method TYPE: OptionSelection DEFAULT: best
index	Index of the cross section to use, if None use all TYPE: int | None DEFAULT: None
initial_vol	Initial volatility for the root finding algorithm TYPE: float DEFAULT: INITIAL_VOL
converged	Whether the calculation has converged TYPE: bool DEFAULT: False

Source code in quantflow/options/surface.py

def options_df(
    self,
    *,
    select: Annotated[
        OptionSelection, Doc("Option selection method")
    ] = OptionSelection.best,
    index: Annotated[
        int | None, Doc("Index of the cross section to use, if None use all")
    ] = None,
    initial_vol: Annotated[
        float, Doc("Initial volatility for the root finding algorithm")
    ] = INITIAL_VOL,
    converged: Annotated[
        bool, Doc("Whether the calculation has converged")
    ] = False,
) -> pd.DataFrame:
    """Time frame of Black-Scholes call input data"""
    data = self.option_prices(
        select=select,
        index=index,
        initial_vol=initial_vol,
        converged=converged,
    )
    return pd.DataFrame([d.info_dict() for d in data])

as_array

as_array(*, select=best, index=None, initial_vol=INITIAL_VOL, converged=False)

Organize option prices in a numpy arrays for Black volatility and price calculation

It returns an OptionArrays instance, which contains the option prices and their corresponding log strikes, time to maturity and implied volatility in numpy arrays for efficient calculations.

PARAMETER	DESCRIPTION
select	Option selection method TYPE: OptionSelection DEFAULT: best
index	Index of the cross section to use, if None use all TYPE: int | None DEFAULT: None
initial_vol	Initial volatility for the root finding algorithm TYPE: float DEFAULT: INITIAL_VOL
converged	If True, include only options for which the calculation has converged TYPE: bool DEFAULT: False

Source code in quantflow/options/surface.py

def as_array(
    self,
    *,
    select: Annotated[
        OptionSelection, Doc("Option selection method")
    ] = OptionSelection.best,
    index: Annotated[
        int | None, Doc("Index of the cross section to use, if None use all")
    ] = None,
    initial_vol: Annotated[
        float, Doc("Initial volatility for the root finding algorithm")
    ] = INITIAL_VOL,
    converged: Annotated[
        bool,
        Doc(
            "If True, include only options for which the calculation has converged"
        ),
    ] = False,
) -> OptionArrays:
    """Organize option prices in a numpy arrays for Black volatility
    and price calculation

    It returns an [OptionArrays][quantflow.options.surface.OptionArrays] instance,
    which contains the option prices and their corresponding log strikes,
    time to maturity and implied volatility in numpy arrays
    for efficient calculations.
    """
    options = list(
        self.option_prices(
            select=select,
            index=index,
            initial_vol=initial_vol,
            converged=converged,
        )
    )
    log_strike = []
    ttm = []
    price = []
    vol = []
    call_put = []
    for option in options:
        log_strike.append(float(option.log_strike))
        price.append(float(option.price_in_forward_space))
        ttm.append(float(option.ttm))
        vol.append(float(option.implied_vol))
        call_put.append(1 if option.option_type.is_call() else -1)
    return OptionArrays(
        options=options,
        log_strike=np.array(log_strike),
        price=np.array(price),
        ttm=np.array(ttm),
        implied_vol=np.array(vol),
        call_put=np.array(call_put),
    )

reset_convergence

reset_convergence()

Reset the convergence flag for all options in the surface

Source code in quantflow/options/surface.py

def reset_convergence(self) -> None:
    """Reset the convergence flag for all options in the surface"""
    for option in self.option_prices(select=OptionSelection.all):
        option.converged = False

disable_outliers

disable_outliers(*, bid_ask_spread_fraction=0.2, svi_residual_fraction=0.2, repeat=2)

Disable outlier options across all maturities in the surface.

Calls VolCrossSection.disable_outliers on each maturity with the same parameters.

PARAMETER	DESCRIPTION
bid_ask_spread_fraction	Maximum allowed bid/ask spread as a fraction of the mid implied volatility. A value of 0.2 means options with a spread greater than 20% of the mid vol are disabled. TYPE: float DEFAULT: 0.2
svi_residual_fraction	Maximum allowed SVI residual as a fraction of the mid implied volatility. A value of 0.2 means options whose mid vol deviates from the SVI fit by more than 20% of their mid vol are disabled. TYPE: float DEFAULT: 0.2
repeat	Number of times to repeat the outlier removal process TYPE: int DEFAULT: 2

Source code in quantflow/options/surface.py

def disable_outliers(
    self,
    *,
    bid_ask_spread_fraction: Annotated[
        float,
        Doc(
            "Maximum allowed bid/ask spread as a fraction of the mid implied "
            "volatility. A value of 0.2 means options with a spread greater than "
            "20% of the mid vol are disabled."
        ),
    ] = 0.2,
    svi_residual_fraction: Annotated[
        float,
        Doc(
            "Maximum allowed SVI residual as a fraction of the mid implied "
            "volatility. A value of 0.2 means options whose mid vol deviates "
            "from the SVI fit by more than 20% of their mid vol are disabled."
        ),
    ] = 0.2,
    repeat: Annotated[
        int, Doc("Number of times to repeat the outlier removal process")
    ] = 2,
) -> Self:
    """Disable outlier options across all maturities in the surface.

    Calls
    [VolCrossSection.disable_outliers]
    [quantflow.options.surface.VolCrossSection.disable_outliers]
    on each maturity with the same parameters.
    """
    for maturity in self.maturities:
        maturity.disable_outliers(
            ttm=maturity.ttm(self.ref_date),
            bid_ask_spread_fraction=bid_ask_spread_fraction,
            svi_residual_fraction=svi_residual_fraction,
            repeat=repeat,
        )
    return self

plot

plot(*, index=None, select=best, **kwargs)

Plot the volatility surface

PARAMETER	DESCRIPTION
index	Index of the cross section to use, if None use all TYPE: int | None DEFAULT: None
select	Option selection method TYPE: OptionSelection DEFAULT: best

Source code in quantflow/options/surface.py

def plot(
    self,
    *,
    index: Annotated[
        int | None, Doc("Index of the cross section to use, if None use all")
    ] = None,
    select: Annotated[
        OptionSelection, Doc("Option selection method")
    ] = OptionSelection.best,
    **kwargs: Any,
) -> Any:
    """Plot the volatility surface"""
    df = self.options_df(index=index, select=select, converged=True)
    return plot.plot_vol_surface(df, **kwargs)

plot3d

plot3d(*, select=best, index=None, dragmode='turntable', **kwargs)

Plot the volatility surface

PARAMETER	DESCRIPTION
select	Option selection method TYPE: OptionSelection DEFAULT: best
index	Index of the cross section to use, if None use all TYPE: int | None DEFAULT: None
dragmode	Drag interaction mode for the 3D scene TYPE: str DEFAULT: 'turntable'

Source code in quantflow/options/surface.py

def plot3d(
    self,
    *,
    select: Annotated[
        OptionSelection, Doc("Option selection method")
    ] = OptionSelection.best,
    index: Annotated[
        int | None, Doc("Index of the cross section to use, if None use all")
    ] = None,
    dragmode: Annotated[
        str, Doc("Drag interaction mode for the 3D scene")
    ] = "turntable",
    **kwargs: Any,
) -> Any:
    """Plot the volatility surface"""
    df = self.options_df(select=select, index=index, converged=True)
    return plot.plot_vol_surface_3d(df, dragmode=dragmode, **kwargs)

quantflow.options.surface.VolCrossSection pydantic-model

Bases: BaseModel, Generic[S]

Represents a cross section of a volatility surface at a specific maturity.

Fields:

• maturity (datetime)
• forward (FwdPrice[S])
• strikes (tuple[Strike[S], ...])
• day_counter (DayCounter)

maturity pydantic-field

maturity

Maturity date of the cross section

forward pydantic-field

forward

Forward price of the underlying asset at the time of the cross section

strikes pydantic-field

strikes

Tuple of sorted strikes and their corresponding option prices

day_counter pydantic-field

day_counter = default_day_counter

Day counter for time to maturity calculations - by default it uses Act/Act

ttm

ttm(ref_date)

Time to maturity in years

Source code in quantflow/options/surface.py

def ttm(self, ref_date: datetime) -> float:
    """Time to maturity in years"""
    return self.day_counter.dcf(ref_date, self.maturity)

forward_rate

forward_rate(ref_date, spot)

Compute the implied continuous rate from spot and forward mid

Source code in quantflow/options/surface.py

def forward_rate(self, ref_date: datetime, spot: SpotPrice[S]) -> Rate:
    """Compute the implied continuous rate from spot and forward mid"""
    return Rate.from_spot_and_forward(
        spot.mid,
        self.forward.mid,
        ref_date,
        self.maturity,
        day_counter=self.day_counter,
    )

forward_spread_fraction

forward_spread_fraction()

Bid-ask spread of the forward as a fraction of its mid price

Source code in quantflow/options/surface.py

def forward_spread_fraction(self) -> Decimal:
    """Bid-ask spread of the forward as a fraction of its mid price"""
    mid = self.forward.mid
    if mid <= ZERO:
        return Decimal("Inf")
    return (self.forward.ask - self.forward.bid) / mid

info_dict

info_dict(ref_date, spot)

Return a dictionary with information about the cross section

Source code in quantflow/options/surface.py

def info_dict(self, ref_date: datetime, spot: SpotPrice[S]) -> dict:
    """Return a dictionary with information about the cross section"""
    return dict(
        maturity=self.maturity,
        ttm=self.ttm(ref_date),
        forward=self.forward.mid,
        bid_ask_spread=self.forward.spread,
        basis=self.forward.mid - spot.mid,
        rate_percent=self.forward_rate(ref_date, spot).percent,
        fwd_spread_pct=round(100 * self.forward_spread_fraction(), 4),
        open_interest=self.forward.open_interest,
        volume=self.forward.volume,
    )

option_prices

option_prices(ref_date, *, select=best, initial_vol=INITIAL_VOL, converged=False)

Iterator over option prices in the cross section

PARAMETER	DESCRIPTION
ref_date	Reference date for time to maturity calculation TYPE: datetime
select	Option selection method TYPE: OptionSelection DEFAULT: best
initial_vol	Initial volatility for the root finding algorithm TYPE: float DEFAULT: INITIAL_VOL
converged	Whether the calculation has converged TYPE: bool DEFAULT: False

Source code in quantflow/options/surface.py

def option_prices(
    self,
    ref_date: Annotated[
        datetime, Doc("Reference date for time to maturity calculation")
    ],
    *,
    select: Annotated[
        OptionSelection, Doc("Option selection method")
    ] = OptionSelection.best,
    initial_vol: Annotated[
        float, Doc("Initial volatility for the root finding algorithm")
    ] = INITIAL_VOL,
    converged: Annotated[
        bool, Doc("Whether the calculation has converged")
    ] = False,
) -> Iterator[OptionPrice]:
    """Iterator over option prices in the cross section"""
    for s in self.strikes:
        yield from s.option_prices(
            self.forward.mid,
            self.ttm(ref_date),
            select=select,
            initial_vol=initial_vol,
            converged=converged,
        )

securities

securities(*, select=all, converged=False)

Iterator over all securities in the cross section

PARAMETER	DESCRIPTION
select	Option selection method TYPE: OptionSelection DEFAULT: all
converged	Include the forward and options with implied volatility converged only if True, otherwise include all securities regardless of convergence TYPE: bool DEFAULT: False

Source code in quantflow/options/surface.py

def securities(
    self,
    *,
    select: Annotated[
        OptionSelection, Doc("Option selection method")
    ] = OptionSelection.all,
    converged: Annotated[
        bool,
        Doc(
            "Include the forward and options with implied volatility "
            "converged only if `True`, otherwise include all securities regardless "
            "of convergence"
        ),
    ] = False,
) -> Iterator[FwdPrice[S] | OptionPrices[S]]:
    """Iterator over all securities in the cross section"""
    yield self.forward
    yield from self.option_securities(select=select, converged=converged)

option_securities

option_securities(*, select=all, converged=False)

Iterator over all option securities in the cross section

PARAMETER	DESCRIPTION
select	Option selection method TYPE: OptionSelection DEFAULT: all
converged	Include the forward and options with implied volatility converged only if True, otherwise include all securities regardless of convergence TYPE: bool DEFAULT: False

Source code in quantflow/options/surface.py

def option_securities(
    self,
    *,
    select: Annotated[
        OptionSelection, Doc("Option selection method")
    ] = OptionSelection.all,
    converged: Annotated[
        bool,
        Doc(
            "Include the forward and options with implied volatility "
            "converged only if `True`, otherwise include all securities regardless "
            "of convergence"
        ),
    ] = False,
) -> Iterator[OptionPrices[S]]:
    """Iterator over all option securities in the cross section"""
    for strike in self.strikes:
        yield from strike.securities(
            self.forward.mid,
            select=select,
            converged=converged,
        )

disable_outliers

disable_outliers(*, ttm, bid_ask_spread_fraction=0.2, svi_residual_fraction=0.2, repeat=2)

Disable outlier options in the cross section by marking them as not converged.

Two passes are applied:

First pass: options where the bid/ask spread in implied vol space exceeds bid_ask_spread_fraction of the mid implied vol are disabled. For example, a value of 0.2 disables options where the spread is more than 20% of the mid vol. Options with a zero mid vol are also disabled.

Second pass: an SVI smile is fitted to the surviving options (mid implied vol vs log-strike). Options whose residual from the SVI fit exceeds svi_residual_fraction of their mid implied vol are disabled. This is repeated up to repeat times, refitting after each removal. The loop stops early if no outliers are found or fewer than 5 options remain.

PARAMETER	DESCRIPTION
ttm	Time to maturity in years, used for SVI fitting TYPE: float
bid_ask_spread_fraction	Maximum allowed bid/ask spread as a fraction of the mid implied volatility. A value of 0.2 means options with a spread greater than 20% of the mid vol are disabled. TYPE: float DEFAULT: 0.2
svi_residual_fraction	Maximum allowed SVI residual as a fraction of the mid implied volatility. A value of 0.2 means options whose mid vol deviates from the SVI fit by more than 20% of their mid vol are disabled. TYPE: float DEFAULT: 0.2
repeat	Number of times to repeat the outlier removal process TYPE: int DEFAULT: 2

Source code in quantflow/options/surface.py

def disable_outliers(
    self,
    *,
    ttm: Annotated[float, Doc("Time to maturity in years, used for SVI fitting")],
    bid_ask_spread_fraction: Annotated[
        float,
        Doc(
            "Maximum allowed bid/ask spread as a fraction of the mid implied "
            "volatility. A value of 0.2 means options with a spread greater than "
            "20% of the mid vol are disabled."
        ),
    ] = 0.2,
    svi_residual_fraction: Annotated[
        float,
        Doc(
            "Maximum allowed SVI residual as a fraction of the mid implied "
            "volatility. A value of 0.2 means options whose mid vol deviates "
            "from the SVI fit by more than 20% of their mid vol are disabled."
        ),
    ] = 0.2,
    repeat: Annotated[
        int, Doc("Number of times to repeat the outlier removal process")
    ] = 2,
) -> None:
    """Disable outlier options in the cross section by marking them as not
    converged.

    Two passes are applied:

    First pass: options where the bid/ask spread in implied vol space exceeds
    `bid_ask_spread_fraction` of the mid implied vol are disabled.
    For example, a value of 0.2 disables options where the spread is more
    than 20% of the mid vol. Options with a zero mid vol are also disabled.

    Second pass: an [SVI][quantflow.options.svi.SVI] smile is fitted to the
    surviving options (mid implied vol vs log-strike). Options whose
    residual from the SVI fit exceeds `svi_residual_fraction` of their mid
    implied vol are disabled. This is repeated up to `repeat` times,
    refitting after each removal. The loop stops early if no outliers are
    found or fewer than 5 options remain.
    """
    options = list(self.option_securities(converged=True))
    # first remove options with high bid/offer spread
    for option in options:
        spread = option.iv_bid_ask_spread()
        mid = option.iv_mid()
        if mid > 0:
            if spread / mid > bid_ask_spread_fraction:
                option.disable()
        else:
            option.disable()
    # remove outliers based on residuals from an SVI smile fit
    forward = float(self.forward.mid)
    for _ in range(repeat):
        options = list(self.option_securities(converged=True))
        if len(options) < 5:
            break
        log_m = np.array([np.log(float(o.meta.strike) / forward) for o in options])
        iv_mid = np.array([o.iv_mid() for o in options])
        try:
            svi = SVI.fit(log_m, iv_mid, ttm)
        except Exception:
            break
        iv_fit = svi.implied_vol(log_m, ttm)
        residuals = np.abs(iv_mid - iv_fit) / iv_mid
        found = False
        for option, residual in zip(options, residuals):
            if residual > svi_residual_fraction:
                option.disable()
                found = True
        if not found:
            break

quantflow.options.surface.GenericVolSurfaceLoader pydantic-model

Bases: BaseModel, Generic[S]

Helper class to build a volatility surface from a list of securities

Use this class to add spot, forward and option securities with their prices and then call the surface method to build a VolSurface instance from the provided data.

Fields:

• asset (str)
• spot (SpotPrice[S] | None)
• maturities (dict[datetime, VolCrossSectionLoader[S]])
• day_counter (DayCounter)
• tick_size_forwards (DecimalNumber | None)
• tick_size_options (DecimalNumber | None)
• exclude_open_interest (DecimalNumber | None)
• exclude_volume (DecimalNumber | None)

asset pydantic-field

asset = ''

Name of the underlying asset

spot pydantic-field

spot = None

Spot price of the underlying asset

maturities pydantic-field

maturities

Dictionary of maturities and their corresponding cross section loaders

day_counter pydantic-field

day_counter = default_day_counter

Day counter for time to maturity calculations by default it uses Act/Act

tick_size_forwards pydantic-field

tick_size_forwards = None

Tick size for rounding forward and spot prices - optional

tick_size_options pydantic-field

tick_size_options = None

Tick size for rounding option prices - optional

exclude_open_interest pydantic-field

exclude_open_interest = None

Exclude options with open interest at or below this value

exclude_volume pydantic-field

exclude_volume = None

Exclude options with volume at or below this value

get_or_create_maturity

get_or_create_maturity(maturity)

Get or create a VolCrossSectionLoader for a given maturity

PARAMETER	DESCRIPTION
maturity	Maturity date for the options TYPE: datetime

Source code in quantflow/options/surface.py

def get_or_create_maturity(
    self, maturity: Annotated[datetime, Doc("Maturity date for the options")]
) -> VolCrossSectionLoader[S]:
    """Get or create a
    [VolCrossSectionLoader][quantflow.options.surface.VolCrossSectionLoader]
    for a given maturity"""
    if maturity not in self.maturities:
        self.maturities[maturity] = VolCrossSectionLoader(
            maturity=maturity,
            day_counter=self.day_counter,
        )
    return self.maturities[maturity]

add_spot

add_spot(security, bid=ZERO, ask=ZERO, open_interest=ZERO, volume=ZERO)

Add a spot to the volatility surface loader

PARAMETER	DESCRIPTION
security	Security for the spot price TYPE: S
bid	Bid price for the spot TYPE: Decimal DEFAULT: ZERO
ask	Ask price for the spot TYPE: Decimal DEFAULT: ZERO
open_interest	Open interest for the spot TYPE: Decimal DEFAULT: ZERO
volume	Volume for the spot TYPE: Decimal DEFAULT: ZERO

Source code in quantflow/options/surface.py

def add_spot(
    self,
    security: Annotated[S, Doc("Security for the spot price")],
    bid: Annotated[Decimal, Doc("Bid price for the spot")] = ZERO,
    ask: Annotated[Decimal, Doc("Ask price for the spot")] = ZERO,
    open_interest: Annotated[Decimal, Doc("Open interest for the spot")] = ZERO,
    volume: Annotated[Decimal, Doc("Volume for the spot")] = ZERO,
) -> None:
    """Add a spot to the volatility surface loader"""
    if security.vol_surface_type() != VolSecurityType.spot:
        raise ValueError("Security is not a spot")
    self.spot = SpotPrice(
        security=security,
        bid=normalize_decimal(bid),
        ask=normalize_decimal(ask),
        open_interest=normalize_decimal(open_interest),
        volume=normalize_decimal(volume),
    )

add_forward

add_forward(security, maturity, bid=ZERO, ask=ZERO, open_interest=ZERO, volume=ZERO)

Add a forward to the volatility surface loader

PARAMETER	DESCRIPTION
security	Security for the forward price TYPE: S
maturity	Maturity date for the forward price TYPE: datetime
bid	Bid price for the forward TYPE: Decimal DEFAULT: ZERO
ask	Ask price for the forward TYPE: Decimal DEFAULT: ZERO
open_interest	Open interest for the forward TYPE: Decimal DEFAULT: ZERO
volume	Volume for the forward TYPE: Decimal DEFAULT: ZERO

Source code in quantflow/options/surface.py

def add_forward(
    self,
    security: Annotated[S, Doc("Security for the forward price")],
    maturity: Annotated[datetime, Doc("Maturity date for the forward price")],
    bid: Annotated[Decimal, Doc("Bid price for the forward")] = ZERO,
    ask: Annotated[Decimal, Doc("Ask price for the forward")] = ZERO,
    open_interest: Annotated[Decimal, Doc("Open interest for the forward")] = ZERO,
    volume: Annotated[Decimal, Doc("Volume for the forward")] = ZERO,
) -> None:
    """Add a forward to the volatility surface loader"""
    if security.vol_surface_type() != VolSecurityType.forward:
        raise ValueError("Security is not a forward")
    self.get_or_create_maturity(maturity=maturity).forward = FwdPrice(
        security=security,
        bid=normalize_decimal(bid),
        ask=normalize_decimal(ask),
        maturity=maturity,
        open_interest=normalize_decimal(open_interest),
        volume=normalize_decimal(volume),
    )

add_option

add_option(security, strike, maturity, option_type, bid=ZERO, ask=ZERO, open_interest=ZERO, volume=ZERO, inverse=True)

Add an option to the volatility surface loader

PARAMETER	DESCRIPTION
security	Security for the option TYPE: S
strike	Strike price for the option TYPE: Decimal
maturity	Maturity date for the option TYPE: datetime
option_type	Type of the option (call or put) TYPE: OptionType
bid	Bid price for the option TYPE: Decimal DEFAULT: ZERO
ask	Ask price for the option TYPE: Decimal DEFAULT: ZERO
open_interest	Open interest for the option TYPE: Decimal DEFAULT: ZERO
volume	Volume for the option TYPE: Decimal DEFAULT: ZERO
inverse	Whether the option is an inverse option TYPE: bool DEFAULT: True

Source code in quantflow/options/surface.py

def add_option(
    self,
    security: Annotated[S, Doc("Security for the option")],
    strike: Annotated[Decimal, Doc("Strike price for the option")],
    maturity: Annotated[datetime, Doc("Maturity date for the option")],
    option_type: Annotated[OptionType, Doc("Type of the option (call or put)")],
    bid: Annotated[Decimal, Doc("Bid price for the option")] = ZERO,
    ask: Annotated[Decimal, Doc("Ask price for the option")] = ZERO,
    open_interest: Annotated[Decimal, Doc("Open interest for the option")] = ZERO,
    volume: Annotated[Decimal, Doc("Volume for the option")] = ZERO,
    inverse: Annotated[bool, Doc("Whether the option is an inverse option")] = True,
) -> None:
    """Add an option to the volatility surface loader"""
    if security.vol_surface_type() != VolSecurityType.option:
        raise ValueError("Security is not an option")
    if self.exclude_volume is not None and volume <= self.exclude_volume:
        return
    if (
        self.exclude_open_interest is not None
        and open_interest <= self.exclude_open_interest
    ):
        return
    self.get_or_create_maturity(maturity=maturity).add_option(
        security=security,
        strike=strike,
        option_type=option_type,
        bid=bid,
        ask=ask,
        open_interest=open_interest,
        volume=volume,
        inverse=inverse,
    )

surface

surface(ref_date=None)

Build a volatility surface from the provided data

PARAMETER	DESCRIPTION
ref_date	Reference date for the volatility surface TYPE: datetime | None DEFAULT: None

Source code in quantflow/options/surface.py

def surface(
    self,
    ref_date: Annotated[
        datetime | None, Doc("Reference date for the volatility surface")
    ] = None,
) -> VolSurface[S]:
    """Build a volatility surface from the provided data"""
    if not self.spot or self.spot.mid == ZERO:
        raise ValueError("No spot price provided")
    maturities = []
    ref_date = ref_date or utcnow()
    previous_forward = self.spot.mid
    for maturity in sorted(self.maturities):
        if section := self.maturities[maturity].cross_section(
            ref_date=ref_date,
            previous_forward=previous_forward,
            tick_size=self.tick_size_forwards,
        ):
            previous_forward = section.forward.mid
            maturities.append(section)
    return VolSurface(
        asset=self.asset,
        ref_date=ref_date,
        spot=self.spot,
        maturities=tuple(maturities),
        day_counter=self.day_counter,
        tick_size_forwards=self.tick_size_forwards,
        tick_size_options=self.tick_size_options,
    )

quantflow.options.surface.VolSurfaceLoader pydantic-model

Bases: GenericVolSurfaceLoader[DefaultVolSecurity]

Helper class to build a volatility surface from a list of securities

Use this class to add spot, forward and option securities with their prices and then call the surface method to build a VolSurface instance from the provided data.

Fields:

• asset (str)
• spot (SpotPrice[S] | None)
• maturities (dict[datetime, VolCrossSectionLoader[S]])
• day_counter (DayCounter)
• tick_size_forwards (DecimalNumber | None)
• tick_size_options (DecimalNumber | None)
• exclude_open_interest (DecimalNumber | None)
• exclude_volume (DecimalNumber | None)

asset pydantic-field

asset = ''

Name of the underlying asset

spot pydantic-field

spot = None

Spot price of the underlying asset

maturities pydantic-field

maturities

Dictionary of maturities and their corresponding cross section loaders

day_counter pydantic-field

day_counter = default_day_counter

Day counter for time to maturity calculations by default it uses Act/Act

tick_size_forwards pydantic-field

tick_size_forwards = None

Tick size for rounding forward and spot prices - optional

tick_size_options pydantic-field

tick_size_options = None

Tick size for rounding option prices - optional

exclude_open_interest pydantic-field

exclude_open_interest = None

Exclude options with open interest at or below this value

exclude_volume pydantic-field

exclude_volume = None

Exclude options with volume at or below this value

add

add(input)

Add a volatility security input to the loader

PARAMETER	DESCRIPTION
input	Volatility surface input data TYPE: VolSurfaceInput

Source code in quantflow/options/surface.py

def add(
    self, input: Annotated[VolSurfaceInput, Doc("Volatility surface input data")]
) -> None:
    """Add a volatility security input to the loader"""
    if isinstance(input, SpotInput):
        self.add_spot(
            DefaultVolSecurity.spot(),
            bid=input.bid,
            ask=input.ask,
            open_interest=input.open_interest,
            volume=input.volume,
        )
    elif isinstance(input, ForwardInput):
        self.add_forward(
            DefaultVolSecurity.forward(),
            maturity=input.maturity,
            bid=input.bid,
            ask=input.ask,
            open_interest=input.open_interest,
            volume=input.volume,
        )
    elif isinstance(input, OptionInput):
        self.add_option(
            DefaultVolSecurity.option(),
            strike=input.strike,
            option_type=input.option_type,
            maturity=input.maturity,
            bid=input.bid,
            ask=input.ask,
            open_interest=input.open_interest,
            volume=input.volume,
            inverse=input.inverse,
        )
    else:
        raise ValueError(f"Unknown input type {type(input)}")

get_or_create_maturity

get_or_create_maturity(maturity)

Get or create a VolCrossSectionLoader for a given maturity

PARAMETER	DESCRIPTION
maturity	Maturity date for the options TYPE: datetime

Source code in quantflow/options/surface.py

def get_or_create_maturity(
    self, maturity: Annotated[datetime, Doc("Maturity date for the options")]
) -> VolCrossSectionLoader[S]:
    """Get or create a
    [VolCrossSectionLoader][quantflow.options.surface.VolCrossSectionLoader]
    for a given maturity"""
    if maturity not in self.maturities:
        self.maturities[maturity] = VolCrossSectionLoader(
            maturity=maturity,
            day_counter=self.day_counter,
        )
    return self.maturities[maturity]

add_spot

add_spot(security, bid=ZERO, ask=ZERO, open_interest=ZERO, volume=ZERO)

Add a spot to the volatility surface loader

PARAMETER	DESCRIPTION
security	Security for the spot price TYPE: S
bid	Bid price for the spot TYPE: Decimal DEFAULT: ZERO
ask	Ask price for the spot TYPE: Decimal DEFAULT: ZERO
open_interest	Open interest for the spot TYPE: Decimal DEFAULT: ZERO
volume	Volume for the spot TYPE: Decimal DEFAULT: ZERO

Source code in quantflow/options/surface.py

def add_spot(
    self,
    security: Annotated[S, Doc("Security for the spot price")],
    bid: Annotated[Decimal, Doc("Bid price for the spot")] = ZERO,
    ask: Annotated[Decimal, Doc("Ask price for the spot")] = ZERO,
    open_interest: Annotated[Decimal, Doc("Open interest for the spot")] = ZERO,
    volume: Annotated[Decimal, Doc("Volume for the spot")] = ZERO,
) -> None:
    """Add a spot to the volatility surface loader"""
    if security.vol_surface_type() != VolSecurityType.spot:
        raise ValueError("Security is not a spot")
    self.spot = SpotPrice(
        security=security,
        bid=normalize_decimal(bid),
        ask=normalize_decimal(ask),
        open_interest=normalize_decimal(open_interest),
        volume=normalize_decimal(volume),
    )

add_forward

add_forward(security, maturity, bid=ZERO, ask=ZERO, open_interest=ZERO, volume=ZERO)

Add a forward to the volatility surface loader

PARAMETER	DESCRIPTION
security	Security for the forward price TYPE: S
maturity	Maturity date for the forward price TYPE: datetime
bid	Bid price for the forward TYPE: Decimal DEFAULT: ZERO
ask	Ask price for the forward TYPE: Decimal DEFAULT: ZERO
open_interest	Open interest for the forward TYPE: Decimal DEFAULT: ZERO
volume	Volume for the forward TYPE: Decimal DEFAULT: ZERO

Source code in quantflow/options/surface.py

def add_forward(
    self,
    security: Annotated[S, Doc("Security for the forward price")],
    maturity: Annotated[datetime, Doc("Maturity date for the forward price")],
    bid: Annotated[Decimal, Doc("Bid price for the forward")] = ZERO,
    ask: Annotated[Decimal, Doc("Ask price for the forward")] = ZERO,
    open_interest: Annotated[Decimal, Doc("Open interest for the forward")] = ZERO,
    volume: Annotated[Decimal, Doc("Volume for the forward")] = ZERO,
) -> None:
    """Add a forward to the volatility surface loader"""
    if security.vol_surface_type() != VolSecurityType.forward:
        raise ValueError("Security is not a forward")
    self.get_or_create_maturity(maturity=maturity).forward = FwdPrice(
        security=security,
        bid=normalize_decimal(bid),
        ask=normalize_decimal(ask),
        maturity=maturity,
        open_interest=normalize_decimal(open_interest),
        volume=normalize_decimal(volume),
    )

add_option

add_option(security, strike, maturity, option_type, bid=ZERO, ask=ZERO, open_interest=ZERO, volume=ZERO, inverse=True)

Add an option to the volatility surface loader

PARAMETER	DESCRIPTION
security	Security for the option TYPE: S
strike	Strike price for the option TYPE: Decimal
maturity	Maturity date for the option TYPE: datetime
option_type	Type of the option (call or put) TYPE: OptionType
bid	Bid price for the option TYPE: Decimal DEFAULT: ZERO
ask	Ask price for the option TYPE: Decimal DEFAULT: ZERO
open_interest	Open interest for the option TYPE: Decimal DEFAULT: ZERO
volume	Volume for the option TYPE: Decimal DEFAULT: ZERO
inverse	Whether the option is an inverse option TYPE: bool DEFAULT: True

Source code in quantflow/options/surface.py

def add_option(
    self,
    security: Annotated[S, Doc("Security for the option")],
    strike: Annotated[Decimal, Doc("Strike price for the option")],
    maturity: Annotated[datetime, Doc("Maturity date for the option")],
    option_type: Annotated[OptionType, Doc("Type of the option (call or put)")],
    bid: Annotated[Decimal, Doc("Bid price for the option")] = ZERO,
    ask: Annotated[Decimal, Doc("Ask price for the option")] = ZERO,
    open_interest: Annotated[Decimal, Doc("Open interest for the option")] = ZERO,
    volume: Annotated[Decimal, Doc("Volume for the option")] = ZERO,
    inverse: Annotated[bool, Doc("Whether the option is an inverse option")] = True,
) -> None:
    """Add an option to the volatility surface loader"""
    if security.vol_surface_type() != VolSecurityType.option:
        raise ValueError("Security is not an option")
    if self.exclude_volume is not None and volume <= self.exclude_volume:
        return
    if (
        self.exclude_open_interest is not None
        and open_interest <= self.exclude_open_interest
    ):
        return
    self.get_or_create_maturity(maturity=maturity).add_option(
        security=security,
        strike=strike,
        option_type=option_type,
        bid=bid,
        ask=ask,
        open_interest=open_interest,
        volume=volume,
        inverse=inverse,
    )

surface

surface(ref_date=None)

Build a volatility surface from the provided data

PARAMETER	DESCRIPTION
ref_date	Reference date for the volatility surface TYPE: datetime | None DEFAULT: None

Source code in quantflow/options/surface.py

def surface(
    self,
    ref_date: Annotated[
        datetime | None, Doc("Reference date for the volatility surface")
    ] = None,
) -> VolSurface[S]:
    """Build a volatility surface from the provided data"""
    if not self.spot or self.spot.mid == ZERO:
        raise ValueError("No spot price provided")
    maturities = []
    ref_date = ref_date or utcnow()
    previous_forward = self.spot.mid
    for maturity in sorted(self.maturities):
        if section := self.maturities[maturity].cross_section(
            ref_date=ref_date,
            previous_forward=previous_forward,
            tick_size=self.tick_size_forwards,
        ):
            previous_forward = section.forward.mid
            maturities.append(section)
    return VolSurface(
        asset=self.asset,
        ref_date=ref_date,
        spot=self.spot,
        maturities=tuple(maturities),
        day_counter=self.day_counter,
        tick_size_forwards=self.tick_size_forwards,
        tick_size_options=self.tick_size_options,
    )

quantflow.options.surface.VolCrossSectionLoader pydantic-model

Bases: BaseModel, Generic[S]

Fields:

• maturity (datetime)
• forward (FwdPrice[S] | None)
• strikes (dict[Decimal, Strike[S]])
• day_counter (DayCounter)

maturity pydantic-field

maturity

Maturity date of the cross section

forward pydantic-field

forward = None

Forward price of the underlying asset at the time of the cross section

strikes pydantic-field

strikes

Dictionary of strikes and their corresponding option prices

day_counter pydantic-field

day_counter = default_day_counter

Day counter for time to maturity calculations - by default it uses Act/Act

add_option

add_option(security, strike, option_type, bid=ZERO, ask=ZERO, open_interest=ZERO, volume=ZERO, inverse=True)

Add an option to the cross section loader

PARAMETER	DESCRIPTION
security	Security for the option TYPE: S
strike	Strike price for the option TYPE: Decimal
option_type	Type of the option (call or put) TYPE: OptionType
bid	Bid price for the option TYPE: Decimal DEFAULT: ZERO
ask	Ask price for the option TYPE: Decimal DEFAULT: ZERO
open_interest	Open interest for the option TYPE: Decimal DEFAULT: ZERO
volume	Volume for the option TYPE: Decimal DEFAULT: ZERO
inverse	Whether the option is an inverse option TYPE: bool DEFAULT: True

Source code in quantflow/options/surface.py

def add_option(
    self,
    security: Annotated[S, Doc("Security for the option")],
    strike: Annotated[Decimal, Doc("Strike price for the option")],
    option_type: Annotated[OptionType, Doc("Type of the option (call or put)")],
    bid: Annotated[Decimal, Doc("Bid price for the option")] = ZERO,
    ask: Annotated[Decimal, Doc("Ask price for the option")] = ZERO,
    open_interest: Annotated[Decimal, Doc("Open interest for the option")] = ZERO,
    volume: Annotated[Decimal, Doc("Volume for the option")] = ZERO,
    inverse: Annotated[bool, Doc("Whether the option is an inverse option")] = True,
) -> None:
    """Add an option to the cross section loader"""
    strike = normalize_decimal(strike)
    if strike not in self.strikes:
        self.strikes[strike] = Strike(strike=strike)
    meta = OptionMetadata(
        strike=strike,
        option_type=option_type,
        maturity=self.maturity,
        open_interest=normalize_decimal(open_interest),
        volume=normalize_decimal(volume),
        inverse=inverse,
    )
    option = OptionPrices(
        security=security,
        meta=meta,
        bid=OptionPrice(price=normalize_decimal(bid), meta=meta, side=Side.bid),
        ask=OptionPrice(price=normalize_decimal(ask), meta=meta, side=Side.ask),
    )
    if option_type.is_call():
        self.strikes[strike].call = option
    else:
        self.strikes[strike].put = option

cross_section

cross_section(ref_date=None, previous_forward=None, tick_size=None)

PARAMETER	DESCRIPTION
ref_date	Reference date for the volatility surface TYPE: datetime | None DEFAULT: None
previous_forward	Previous forward price for the volatility surface Usaed by the implied forward calculation to replace missing or unreliable forwards TYPE: Decimal | None DEFAULT: None
tick_size	Tick size for rounding implied forward bid/ask prices TYPE: Decimal | None DEFAULT: None

Source code in quantflow/options/surface.py

def cross_section(
    self,
    ref_date: Annotated[
        datetime | None, Doc("Reference date for the volatility surface")
    ] = None,
    previous_forward: Annotated[
        Decimal | None,
        Doc(
            "Previous forward price for the volatility surface "
            "Usaed by the implied forward calculation to replace missing "
            "or unreliable forwards"
        ),
    ] = None,
    tick_size: Annotated[
        Decimal | None,
        Doc("Tick size for rounding implied forward bid/ask prices"),
    ] = None,
) -> VolCrossSection[S] | None:
    strikes = []
    implied_forwards = []
    for strike in sorted(self.strikes):
        sk = self.strikes[strike]
        if sk.call is None and sk.put is None:
            continue
        if implied_forward := sk.implied_forward(tick_size=tick_size):
            implied_forwards.append(implied_forward)
        strikes.append(sk)
    forward = self.forward
    if implied_forwards:
        ttm = self.day_counter.dcf(ref_date or utcnow(), self.maturity)
        forward = ImpliedFwdPrice.aggregate(
            implied_forwards,
            ttm,
            default=self.forward,
            previous_forward=previous_forward,
            tick_size=tick_size,
        )
    if forward is None or not forward.is_valid():
        return None
    return (
        VolCrossSection(
            maturity=self.maturity,
            forward=forward,
            strikes=tuple(strikes),
            day_counter=self.day_counter,
        )
        if strikes
        else None
    )

Bid/Ask Prices

quantflow.options.surface.Price pydantic-model

Bases: BaseModel, Generic[S]

Represents the bid/ask price of a security, which can be a spot price, forward price or option price

Fields:

• security (S)
• bid (DecimalNumber)
• ask (DecimalNumber)

security pydantic-field

security

The underlying security of the price

bid pydantic-field

bid

Bid price

ask pydantic-field

ask

Ask price

mid property

mid

Calculate the mid price by averaging the bid and ask prices

spread property

spread

Calculate the bid-ask spread

bp_spread property

bp_spread

Bid-ask spread in basis points, calculated as spread divided by mid price and multiplied by 10000

quantflow.options.surface.SpotPrice pydantic-model

Bases: Price[S]

Represents the spot bid/ask price of an underlying asset

Fields:

• security (S)
• bid (DecimalNumber)
• ask (DecimalNumber)
• open_interest (DecimalNumber)
• volume (DecimalNumber)

open_interest pydantic-field

open_interest = ZERO

Open interest of the spot price

volume pydantic-field

volume = ZERO

Total volume traded

security pydantic-field

security

The underlying security of the price

bid pydantic-field

bid

Bid price

ask pydantic-field

ask

Ask price

mid property

mid

Calculate the mid price by averaging the bid and ask prices

spread property

spread

Calculate the bid-ask spread

bp_spread property

bp_spread

Bid-ask spread in basis points, calculated as spread divided by mid price and multiplied by 10000

inputs

inputs()

Source code in quantflow/options/surface.py

def inputs(self) -> SpotInput:
    return SpotInput(
        bid=self.bid,
        ask=self.ask,
        open_interest=self.open_interest,
        volume=self.volume,
    )

quantflow.options.surface.FwdPrice pydantic-model

Bases: Price[S]

Represents the forward bid/ask price of an underlying asset at a specific maturity

Fields:

• security (S)
• bid (DecimalNumber)
• ask (DecimalNumber)
• maturity (datetime)
• open_interest (DecimalNumber)
• volume (DecimalNumber)

maturity pydantic-field

maturity

Maturity date of the forward price

open_interest pydantic-field

open_interest = ZERO

Open interest of the forward price

volume pydantic-field

volume = ZERO

Total volume traded

security pydantic-field

security

The underlying security of the price

bid pydantic-field

bid

Bid price

ask pydantic-field

ask

Ask price

mid property

mid

Calculate the mid price by averaging the bid and ask prices

spread property

spread

Calculate the bid-ask spread

bp_spread property

bp_spread

Bid-ask spread in basis points, calculated as spread divided by mid price and multiplied by 10000

inputs

inputs()

Source code in quantflow/options/surface.py

def inputs(self) -> ForwardInput:
    return ForwardInput(
        bid=self.bid,
        ask=self.ask,
        maturity=self.maturity,
        open_interest=self.open_interest,
        volume=self.volume,
    )

is_valid

is_valid()

Check if the forward price is valid, which means that the bid and ask are positive and the bid is less than or equal to the ask

Source code in quantflow/options/surface.py

def is_valid(self) -> bool:
    """Check if the forward price is valid, which means that the bid and ask
    are positive and the bid is less than or equal to the ask"""
    return self.bid > ZERO and self.ask > ZERO and self.bid <= self.ask

quantflow.options.surface.ImpliedFwdPrice pydantic-model

Bases: FwdPrice[S]

Represents the implied forward price of an underlying asset at a specific maturity, extracted from option prices via put-call parity

Fields:

• security (S)
• bid (DecimalNumber)
• ask (DecimalNumber)
• maturity (datetime)
• open_interest (DecimalNumber)
• volume (DecimalNumber)
• strike (DecimalNumber)

strike pydantic-field

strike

Strike price of the options used to extract the forward price

security pydantic-field

security

The underlying security of the price

bid pydantic-field

bid

Bid price

ask pydantic-field

ask

Ask price

mid property

mid

Calculate the mid price by averaging the bid and ask prices

spread property

spread

Calculate the bid-ask spread

bp_spread property

bp_spread

Bid-ask spread in basis points, calculated as spread divided by mid price and multiplied by 10000

maturity pydantic-field

maturity

Maturity date of the forward price

open_interest pydantic-field

open_interest = ZERO

Open interest of the forward price

volume pydantic-field

volume = ZERO

Total volume traded

moneyness

moneyness(ttm)

Moneyness of the implied forward

Source code in quantflow/options/surface.py

def moneyness(self, ttm: float) -> float:
    """Moneyness of the implied forward"""
    return math.log(float(self.strike / self.mid)) / math.sqrt(ttm)

aggregate classmethod

aggregate(implied_forwards, ttm, default=None, previous_forward=None, tick_size=None)

Aggregate implied forward prices extracted from put-call parity into a single best-estimate forward price.

Selection: valid implied forwards are sorted by bid-ask spread in basis points and the tightest 5 are retained as candidates. Let \(c\) denote the tightest bp spread among the candidates.

Default priority: if a default forward is provided and its bp spread is tighter than \(c\), it is returned immediately as the most reliable price.

Default inclusion: if the default's bp spread is wider than \(c\) but narrower than the worst candidate, it is appended to the candidate pool and weighted on equal footing with the implied forwards.

Weighting: each candidate \(i\) receives weight

\[\begin{equation} w_i = w^{\text{spread}}_i \cdot w^{\text{proximity}}_i \end{equation}\]

where the spread weight is a Gaussian on the normalised distance from the best spread \(c\) and the proximity weight, applied only when previous_forward is provided. The result is the weighted average of the candidate mid prices, with the bid/ask spread computed as the weighted average of candidate spreads. When tick_size is provided the output bid is rounded down and the ask is rounded up to the nearest tick.

PARAMETER	DESCRIPTION
implied_forwards	Implied forward prices from put-call parity TYPE: list[Self]
ttm	Time to maturity in years TYPE: float
default	Market forward (e.g. from futures) used as fallback or for blending TYPE: FwdPrice[S] | None DEFAULT: None
previous_forward	Anchor forward for proximity weighting, typically the previous maturity TYPE: Decimal | None DEFAULT: None
tick_size	Tick size for rounding the implied forward bid/ask TYPE: Decimal | None DEFAULT: None

Source code in quantflow/options/surface.py

@classmethod
def aggregate(
    cls,
    implied_forwards: Annotated[
        list[Self], Doc("Implied forward prices from put-call parity")
    ],
    ttm: Annotated[float, Doc("Time to maturity in years")],
    default: Annotated[
        FwdPrice[S] | None,
        Doc("Market forward (e.g. from futures) used as fallback or for blending"),
    ] = None,
    previous_forward: Annotated[
        Decimal | None,
        Doc(
            "Anchor forward for proximity weighting, "
            "typically the previous maturity"
        ),
    ] = None,
    tick_size: Annotated[
        Decimal | None, Doc("Tick size for rounding the implied forward bid/ask")
    ] = None,
) -> FwdPrice[S] | None:
    r"""Aggregate implied forward prices extracted from put-call parity into a
    single best-estimate forward price.

    **Selection**: valid implied forwards are sorted by bid-ask spread in basis
    points and the tightest 5 are retained as candidates. Let $c$ denote the
    tightest bp spread among the candidates.

    **Default priority**: if a default forward is provided and its bp spread is
    tighter than $c$, it is returned immediately as the most reliable price.

    **Default inclusion**: if the default's bp spread is wider than $c$ but
    narrower than the worst candidate, it is appended to the candidate pool
    and weighted on equal footing with the implied forwards.

    **Weighting**: each candidate $i$ receives weight

    \begin{equation}
        w_i = w^{\text{spread}}_i \cdot w^{\text{proximity}}_i
    \end{equation}

    where the spread weight is a Gaussian on the normalised distance from the
    best spread $c$ and the proximity weight, applied only when
    `previous_forward` is provided.
    The result is the weighted average of the candidate mid prices, with the
    bid/ask spread computed as the weighted average of candidate spreads.
    When `tick_size` is provided the output bid is rounded down and the ask
    is rounded up to the nearest tick.
    """
    forwards: list[FwdPrice[S]] = [f for f in implied_forwards if f.is_valid()]
    if not forwards:
        return default
    forwards = sorted(forwards, key=lambda f: f.bp_spread)[:5]
    best_bp_spread = forwards[0].bp_spread
    if (
        default is not None
        and default.is_valid()
        and default.bp_spread < best_bp_spread
    ):
        return default
    weights = 0.0
    values = 0.0
    spreads = 0.0
    worse_bp_spread = forwards[-1].bp_spread
    if (
        default is not None
        and default.is_valid()
        and default.bp_spread < worse_bp_spread
    ):
        forwards.append(default)
    for forward in forwards:
        s = (forward.bp_spread - best_bp_spread) / best_bp_spread
        weight = math.exp(-s * s)
        if previous_forward is not None:
            d = (forward.mid - previous_forward) / previous_forward
            weight *= math.exp(-d * d)
        weights += weight
        values += weight * float(forward.mid)
        spreads += weight * float(forward.spread)
    mid = to_decimal(values / weights)
    spread = to_decimal(spreads / weights)
    bid = mid - spread / 2
    ask = mid + spread / 2
    if tick_size is not None:
        bid = round_to_step(bid, tick_size, Rounding.DOWN)
        ask = round_to_step(ask, tick_size, Rounding.UP)
    return FwdPrice(
        security=forwards[0].security.forward(),
        bid=bid,
        ask=ask,
        maturity=forwards[0].maturity,
    )

inputs

inputs()

Source code in quantflow/options/surface.py

def inputs(self) -> ForwardInput:
    return ForwardInput(
        bid=self.bid,
        ask=self.ask,
        maturity=self.maturity,
        open_interest=self.open_interest,
        volume=self.volume,
    )

is_valid

is_valid()

Check if the forward price is valid, which means that the bid and ask are positive and the bid is less than or equal to the ask

Source code in quantflow/options/surface.py

def is_valid(self) -> bool:
    """Check if the forward price is valid, which means that the bid and ask
    are positive and the bid is less than or equal to the ask"""
    return self.bid > ZERO and self.ask > ZERO and self.bid <= self.ask

quantflow.options.surface.Strike pydantic-model

Bases: BaseModel, Generic[S]

Option prices for a single strike

Fields:

• strike (DecimalNumber)
• call (OptionPrices[S] | None)
• put (OptionPrices[S] | None)

strike pydantic-field

strike

Strike price of the options

call pydantic-field

call = None

Call option prices for the strike

put pydantic-field

put = None

Put option prices for the strike

implied_forward

implied_forward(tick_size=None)

Extract the implied forward price from put-call parity.

Requires both a call and a put at this strike. Uses bid/ask prices to construct the bid/ask of the implied forward. When tick_size is provided, bid is rounded down and ask is rounded up to the nearest tick.

For inverse options (prices quoted in the underlying currency) put-call parity reads

\[\begin{equation} F = \frac{K}{1 - c + p} \end{equation}\]

For non-inverse options (prices quoted in the quote currency)

\[\begin{equation} F = K + C - P \end{equation}\]

Returns None when the strike does not have both a call and a put, or when the denominator is non-positive (arbitrage condition violated).

PARAMETER	DESCRIPTION
tick_size	Tick size for rounding the implied forward bid/ask TYPE: Decimal | None DEFAULT: None

Source code in quantflow/options/surface.py

def implied_forward(
    self,
    tick_size: Annotated[
        Decimal | None, Doc("Tick size for rounding the implied forward bid/ask")
    ] = None,
) -> ImpliedFwdPrice[S] | None:
    r"""Extract the implied forward price from put-call parity.

    Requires both a call and a put at this strike. Uses bid/ask prices
    to construct the bid/ask of the implied forward. When `tick_size` is
    provided, bid is rounded down and ask is rounded up to the nearest tick.

    For inverse options (prices quoted in the underlying currency)
    put-call parity reads

    \begin{equation}
        F = \frac{K}{1 - c + p}
    \end{equation}

    For non-inverse options (prices quoted in the quote currency)

    \begin{equation}
        F = K + C - P
    \end{equation}

    Returns None when the strike does not have both a call and a put,
    or when the denominator is non-positive (arbitrage condition violated).
    """
    if self.call is None or self.put is None:
        return None
    cp_bid = self.call.bid.price - self.put.ask.price
    cp_ask = self.call.ask.price - self.put.bid.price
    if self.call.meta.inverse:
        d_bid = 1 - cp_bid
        d_ask = 1 - cp_ask
        if d_bid <= ZERO or d_ask <= ZERO:
            return None
        bid = self.strike / d_bid
        ask = self.strike / d_ask
    else:
        bid = self.strike + cp_bid
        ask = self.strike + cp_ask
        if bid <= ZERO or ask <= ZERO:
            return None
    if bid > ask:
        return None
    if tick_size is not None:
        bid = round_to_step(bid, tick_size, Rounding.DOWN)
        ask = round_to_step(ask, tick_size, Rounding.UP)
    return ImpliedFwdPrice(
        security=self.call.security.forward(),
        strike=self.strike,
        maturity=self.call.meta.maturity,
        bid=bid,
        ask=ask,
    )

options_iter

options_iter(forward, *, select=all)

Iterator over option prices for the strike

It uses the select parameter to determine which options to include in the iteration. The forward price is used to determine the moneyness of the options when the best selection method is used, in which case only the Out of the Money options are included in the iteration.

PARAMETER	DESCRIPTION
forward	Forward price of the underlying asset TYPE: Decimal
select	Option selection method TYPE: OptionSelection DEFAULT: all

Source code in quantflow/options/surface.py

def options_iter(
    self,
    forward: Annotated[Decimal, Doc("Forward price of the underlying asset")],
    *,
    select: Annotated[
        OptionSelection, Doc("Option selection method")
    ] = OptionSelection.all,
) -> Iterator[OptionPrices[S]]:
    """Iterator over option prices for the strike

    It uses the `select` parameter to determine which options to include in
    the iteration. The forward price is used to determine the moneyness of
    the options when the `best` selection method is used, in which
    case only the Out of the Money options are included in the iteration.
    """
    match select:
        case OptionSelection.best:
            if self.call and not self.call.is_in_the_money(forward):
                yield self.call
            elif self.put and not self.put.is_in_the_money(forward):
                yield self.put
        case OptionSelection.call:
            if self.call:
                yield self.call
        case OptionSelection.put:
            if self.put:
                yield self.put
        case OptionSelection.all:
            if self.call:
                yield self.call
            if self.put:
                yield self.put

securities

securities(forward, *, select=all, converged=False)

Iterator over option prices for the strike

PARAMETER	DESCRIPTION
forward	Forward price of the underlying asset TYPE: Decimal
select	Option selection method TYPE: OptionSelection DEFAULT: all
converged	Include options with implied volatility converged only if True, otherwise include all options regardless of convergence TYPE: bool DEFAULT: False

Source code in quantflow/options/surface.py

def securities(
    self,
    forward: Annotated[Decimal, Doc("Forward price of the underlying asset")],
    *,
    select: Annotated[
        OptionSelection, Doc("Option selection method")
    ] = OptionSelection.all,
    converged: Annotated[
        bool,
        Doc(
            "Include options with implied volatility converged only if True, "
            "otherwise include all options regardless of convergence"
        ),
    ] = False,
) -> Iterator[OptionPrices[S]]:
    """Iterator over option prices for the strike"""
    for option in self.options_iter(forward, select=select):
        if not converged or option.converged:
            yield option

option_prices

option_prices(forward, ttm, *, select=best, initial_vol=INITIAL_VOL, converged=False)

PARAMETER	DESCRIPTION
forward	Forward price of the underlying asset TYPE: Decimal
ttm	Time to maturity in years TYPE: float
select	Option selection method TYPE: OptionSelection DEFAULT: best
initial_vol	Initial volatility for the root finding algorithm TYPE: float DEFAULT: INITIAL_VOL
converged	Include options with implied volatility converged only if True, otherwise include all options regardless of convergence TYPE: bool DEFAULT: False

Source code in quantflow/options/surface.py

def option_prices(
    self,
    forward: Annotated[Decimal, Doc("Forward price of the underlying asset")],
    ttm: Annotated[float, Doc("Time to maturity in years")],
    *,
    select: Annotated[
        OptionSelection, Doc("Option selection method")
    ] = OptionSelection.best,
    initial_vol: Annotated[
        float, Doc("Initial volatility for the root finding algorithm")
    ] = INITIAL_VOL,
    converged: Annotated[
        bool,
        Doc(
            "Include options with implied volatility converged only if True, "
            "otherwise include all options regardless of convergence"
        ),
    ] = False,
) -> Iterator[OptionPrice]:
    for option in self.options_iter(forward, select=select):
        if not converged or option.converged:
            yield from option.prices(
                forward,
                ttm,
                initial_vol=initial_vol,
            )

quantflow.options.surface.OptionArrays

Bases: NamedTuple

Represents the option data in array form for efficient calculations via vectorized operations

options instance-attribute

options

List of option prices corresponding to the arrays below

log_strike instance-attribute

log_strike

The log strike of the options, calculated as log(strike/forward)

price instance-attribute

price

The option prices

ttm instance-attribute

ttm

Time to maturity of the options

implied_vol instance-attribute

implied_vol

Implied volatility of the options

call_put instance-attribute

call_put

Indicator for call (1) or put (-1) options

quantflow.options.surface.OptionMetadata pydantic-model

Bases: BaseModel

Represents the metadata of an option, including its strike, type, maturity, and other relevant information.

Fields:

• strike (DecimalNumber)
• option_type (OptionType)
• maturity (datetime)
• forward (DecimalNumber)
• ttm (float)
• open_interest (DecimalNumber)
• volume (DecimalNumber)
• inverse (bool)

strike pydantic-field

strike

Strike price of the option

option_type pydantic-field

option_type

Type of the option, call or put

maturity pydantic-field

maturity

Maturity date of the option

forward pydantic-field

forward = ZERO

Forward price of the underlying

ttm pydantic-field

ttm = 0

Time to maturity in years

open_interest pydantic-field

open_interest = ZERO

Open interest of the option

volume pydantic-field

volume = ZERO

Total volume traded

inverse pydantic-field

inverse = True

Whether the option is an inverse option (i.e. quoted in terms of the underlying) or not (i.e. quoted in terms of the quote currency)

is_in_the_money

is_in_the_money(forward)

Check if the option is in the money given the forward price

Source code in quantflow/options/surface.py

def is_in_the_money(self, forward: Decimal) -> bool:
    """Check if the option is in the money given the forward price"""
    if self.option_type.is_call():
        return self.strike < forward
    else:
        return self.strike > forward

quantflow.options.surface.OptionPrice pydantic-model

Bases: BaseModel

Represents the price of an option quoted in the market along with its metadata and implied volatility information.

Fields:

• price (DecimalNumber)
• meta (OptionMetadata)
• implied_vol (float)
• side (Side)
• converged (bool)

price pydantic-field

price

Price of the option as a percentage of the forward price

meta pydantic-field

meta

Metadata of the option price

implied_vol pydantic-field

implied_vol = 0

Implied volatility of the option

side pydantic-field

side = bid

Side of the market for the option price

converged pydantic-field

converged = False

Flag indicating if implied vol calculation converged

strike property

strike

forward property

forward

Forward price of the underlying asset at the time of the option price

maturity property

maturity

ttm property

ttm

option_type property

option_type

open_interest property

open_interest

volume property

volume

log_strike property

log_strike

Log strike of the option, calculated as log(strike/forward)

price_in_forward_space property

price_in_forward_space

Price of the option as a percentage of the forward price

price_bp property

price_bp

Price of the option in basis points, calculated as price in forward space multiplied by 10000

price_in_quote property

price_in_quote

Price of the option in quote currency

price_intrinsic property

price_intrinsic

Intrinsic price of the option in forward space, which is the price if the option had zero time value

price_time property

price_time

Time value of the option in forward space, which is the price minus its intrinsic value

call_price property

call_price

call price in forward space

use put-call parity to calculate the call price if a put

put_price property

put_price

put price in forward space

use put-call parity to calculate the put price if a call

create classmethod

create(strike, *, price=ZERO, implied_vol=INITIAL_VOL, forward=None, ref_date=None, maturity=None, day_counter=None, option_type=call, open_interest=ZERO, volume=ZERO, inverse=True)

Create an option price

mainly used for testing

Source code in quantflow/options/surface.py

@classmethod
def create(
    cls,
    strike: Number,
    *,
    price: Number = ZERO,
    implied_vol: float = INITIAL_VOL,
    forward: Number | None = None,
    ref_date: datetime | None = None,
    maturity: datetime | None = None,
    day_counter: DayCounter | None = None,
    option_type: OptionType = OptionType.call,
    open_interest: Number = ZERO,
    volume: Number = ZERO,
    inverse: bool = True,
) -> Self:
    """Create an option price

    mainly used for testing
    """
    ref_date = ref_date or utcnow()
    maturity = maturity or ref_date + timedelta(days=365)
    day_counter = day_counter or default_day_counter
    return cls(
        price=to_decimal(price),
        implied_vol=implied_vol,
        meta=OptionMetadata(
            strike=to_decimal(strike),
            forward=to_decimal(forward or strike),
            option_type=option_type,
            maturity=maturity,
            ttm=day_counter.dcf(ref_date, maturity),
            open_interest=to_decimal(open_interest),
            volume=to_decimal(volume),
            inverse=inverse,
        ),
    )

is_in_the_money

is_in_the_money(forward)

Source code in quantflow/options/surface.py

def is_in_the_money(self, forward: Decimal) -> bool:
    return self.meta.is_in_the_money(forward)

calculate_price

calculate_price()

Source code in quantflow/options/surface.py

def calculate_price(self) -> Self:
    price = Decimal(
        sigfig(
            black_price(
                np.asarray(self.log_strike),
                self.implied_vol,
                self.ttm,
                1 if self.option_type.is_call() else -1,
            ).sum(),
            8,
        )
    )
    self.price = price if self.meta.inverse else price * self.forward
    return self

info_dict

info_dict()

Source code in quantflow/options/surface.py

def info_dict(self) -> dict[str, Any]:
    return dict(
        strike=float(self.strike),
        forward=float(self.forward),
        maturity=self.maturity,
        log_strike=self.log_strike,
        moneyness=self.log_strike / np.sqrt(self.ttm),
        ttm=self.ttm,
        implied_vol=self.implied_vol,
        price=float(self.price_in_forward_space),
        price_bp=float(self.price_bp),
        price_quote=float(self.price_in_quote),
        type=str(self.option_type),
        side=str(self.side),
        open_interest=float(self.open_interest),
        volume=float(self.volume),
    )

quantflow.options.surface.OptionPrices pydantic-model

Bases: BaseModel, Generic[S]

Represents the market for a single option contract (identified by its strike, maturity and option type), holding the bid and ask sides as separate OptionPrice objects.

Fields:

• security (S)
• meta (OptionMetadata)
• bid (OptionPrice)
• ask (OptionPrice)

security pydantic-field

security

The underlying security of the option prices

meta pydantic-field

meta

Metadata for the option prices

bid pydantic-field

bid

Bid option price

ask pydantic-field

ask

Ask option price

converged property

converged

Check if the implied volatility calculation has converged for both bid and ask

mid property

mid

Calculate the mid option price by averaging the bid and ask prices

spread property

spread

Calculate the bid-ask spread

iv_bid_ask_spread

iv_bid_ask_spread()

Calculate the bid-ask spread of the implied volatility

Source code in quantflow/options/surface.py

def iv_bid_ask_spread(self) -> float:
    """Calculate the bid-ask spread of the implied volatility"""
    return self.ask.implied_vol - self.bid.implied_vol

iv_mid

iv_mid()

Calculate the mid implied volatility

Source code in quantflow/options/surface.py

def iv_mid(self) -> float:
    """Calculate the mid implied volatility"""
    return (self.bid.implied_vol + self.ask.implied_vol) / 2

is_in_the_money

is_in_the_money(forward)

Check if the option is in the money given the forward price

Source code in quantflow/options/surface.py

def is_in_the_money(self, forward: Decimal) -> bool:
    """Check if the option is in the money given the forward price"""
    return self.meta.is_in_the_money(forward)

disable

disable()

Disable the option by setting its implied volatility convergence to False

Source code in quantflow/options/surface.py

def disable(self) -> None:
    """Disable the option by setting its implied volatility convergence to False"""
    self.bid.converged = False
    self.ask.converged = False

prices

prices(forward, ttm, *, initial_vol=INITIAL_VOL)

Iterator over bid/ask option prices

PARAMETER	DESCRIPTION
forward	Forward price of the underlying asset TYPE: Decimal
ttm	Time to maturity in years TYPE: float
initial_vol	Initial volatility for the root finding algorithm TYPE: float DEFAULT: INITIAL_VOL

Source code in quantflow/options/surface.py

def prices(
    self,
    forward: Annotated[Decimal, Doc("Forward price of the underlying asset")],
    ttm: Annotated[float, Doc("Time to maturity in years")],
    *,
    initial_vol: Annotated[
        float, Doc("Initial volatility for the root finding algorithm")
    ] = INITIAL_VOL,
) -> Iterator[OptionPrice]:
    """Iterator over bid/ask option prices"""
    self.meta.forward = forward
    self.meta.ttm = ttm
    for o in (self.bid, self.ask):
        o.meta.forward = forward
        o.meta.ttm = ttm
        if not o.implied_vol:
            o.implied_vol = initial_vol
        yield o

inputs

inputs()

Convert the option prices to an OptionInput instance

Source code in quantflow/options/surface.py

def inputs(self) -> OptionInput:
    """Convert the option prices to an OptionInput instance"""
    return OptionInput(
        bid=self.bid.price,
        ask=self.ask.price,
        open_interest=self.meta.open_interest,
        volume=self.meta.volume,
        strike=self.meta.strike,
        maturity=self.meta.maturity,
        option_type=self.meta.option_type,
        iv_bid=to_decimal_or_none(
            None
            if np.isnan(self.bid.implied_vol)
            else round(self.bid.implied_vol, 7)
        ),
        iv_ask=to_decimal_or_none(
            None
            if np.isnan(self.ask.implied_vol)
            else round(self.ask.implied_vol, 7)
        ),
    )

quantflow.options.surface.OptionSelection

Bases: Enum

Option selection method

This enum is used to select which one between calls and puts are used for calculating implied volatility and other operations

best class-attribute instance-attribute

best = auto()

Select the best bid/ask options.

These are the options which are Out of the Money, where their intrinsic value is zero

call class-attribute instance-attribute

call = auto()

Select the call options only

put class-attribute instance-attribute

put = auto()

Select the put options only

all class-attribute instance-attribute

all = auto()

Select all options regardless of their moneyness

quantflow.options.surface.surface_from_inputs

surface_from_inputs(inputs)

Helper function to build a volatility surface from a VolSurfaceInputs instance

PARAMETER	DESCRIPTION
inputs	Volatility surface input data TYPE: VolSurfaceInputs

Source code in quantflow/options/surface.py

def surface_from_inputs(
    inputs: Annotated[VolSurfaceInputs, Doc("Volatility surface input data")],
) -> VolSurface[DefaultVolSecurity]:
    """Helper function to build a volatility surface from a
    [VolSurfaceInputs][quantflow.options.inputs.VolSurfaceInputs] instance
    """
    loader = VolSurfaceLoader()
    for input in inputs.inputs:
        loader.add(input)
    return loader.surface(ref_date=inputs.ref_date)

Vol Surface Inputs

quantflow.options.inputs.OptionType

Bases: StrEnum

Type of option

call class-attribute instance-attribute

call = auto()

put class-attribute instance-attribute

put = auto()

is_call

is_call()

Source code in quantflow/options/inputs.py

def is_call(self) -> bool:
    return self is OptionType.call

is_put

is_put()

Source code in quantflow/options/inputs.py

def is_put(self) -> bool:
    return self is OptionType.put

call_put

call_put()

Return 1 for call options and -1 for put options

Source code in quantflow/options/inputs.py

def call_put(self) -> int:
    """Return 1 for call options and -1 for put options"""
    return 1 if self is OptionType.call else -1

quantflow.options.inputs.VolSurfaceInputs pydantic-model

Bases: BaseModel

Class representing the inputs for a volatility surface

Fields:

• asset (str)
• ref_date (datetime)
• inputs (list[ForwardInput | SpotInput | OptionInput])

asset pydantic-field

asset

Underlying asset of the volatility surface

ref_date pydantic-field

ref_date

Reference date for the volatility surface

inputs pydantic-field

inputs

List of inputs for the volatility surface

quantflow.options.inputs.VolSurfaceInput pydantic-model

Bases: BaseModel

Base class for volatility surface inputs

Fields:

• bid (DecimalNumber)
• ask (DecimalNumber)
• open_interest (DecimalNumber)
• volume (DecimalNumber)

bid pydantic-field

bid

Bid price of the security

ask pydantic-field

ask

Ask price of the security

open_interest pydantic-field

open_interest = ZERO

Open interest of the security

volume pydantic-field

volume = ZERO

Volume of the security

quantflow.options.inputs.SpotInput pydantic-model

Bases: VolSurfaceInput

Input data for a spot contract in the volatility surface

Fields:

• bid (DecimalNumber)
• ask (DecimalNumber)
• open_interest (DecimalNumber)
• volume (DecimalNumber)
• security_type (VolSecurityType)

security_type pydantic-field

security_type = spot

Type of security for the volatility surface

bid pydantic-field

bid

Bid price of the security

ask pydantic-field

ask

Ask price of the security

open_interest pydantic-field

open_interest = ZERO

Open interest of the security

volume pydantic-field

volume = ZERO

Volume of the security

quantflow.options.inputs.ForwardInput pydantic-model

Bases: VolSurfaceInput

Input data for a forward contract in the volatility surface

Fields:

• bid (DecimalNumber)
• ask (DecimalNumber)
• open_interest (DecimalNumber)
• volume (DecimalNumber)
• maturity (datetime)
• security_type (VolSecurityType)

maturity pydantic-field

maturity

Expiry date of the forward contract

security_type pydantic-field

security_type = forward

Type of security for the volatility surface

bid pydantic-field

bid

Bid price of the security

ask pydantic-field

ask

Ask price of the security

open_interest pydantic-field

open_interest = ZERO

Open interest of the security

volume pydantic-field

volume = ZERO

Volume of the security

quantflow.options.inputs.OptionInput pydantic-model

Bases: VolSurfaceInput

Input data for an option in the volatility surface

Fields:

• bid (DecimalNumber)
• ask (DecimalNumber)
• open_interest (DecimalNumber)
• volume (DecimalNumber)
• strike (DecimalNumber)
• maturity (datetime)
• option_type (OptionType)
• security_type (VolSecurityType)
• iv_bid (DecimalNumber | None)
• iv_ask (DecimalNumber | None)
• inverse (bool)

strike pydantic-field

strike

Strike price of the option

maturity pydantic-field

maturity

Expiry date of the option

option_type pydantic-field

option_type

Type of the option - call or put

security_type pydantic-field

security_type = option

Type of security for the volatility surface

iv_bid pydantic-field

iv_bid = None

Implied volatility based on the bid price as decimal number (e.g. 0.2 for 20%)

iv_ask pydantic-field

iv_ask = None

Implied volatility based on the ask price as decimal number (e.g. 0.2 for 20%)

inverse pydantic-field

inverse = True

Whether the security is inverse (i.e. quoted in terms of the underlying) or not (i.e. quoted in terms of the quote currency)

bid pydantic-field

bid

Bid price of the security

ask pydantic-field

ask

Ask price of the security

open_interest pydantic-field

open_interest = ZERO

Open interest of the security

volume pydantic-field

volume = ZERO

Volume of the security