book
Checkout our new book! Hands on AI Trading with Python, QuantConnect, and AWS Learn More arrow

Universe Selection

Options Universes

Introduction

An Option Universe Selection model selects contracts for a set of Options.

Options Universe Selection

The OptionUniverseSelectionModel selects all the available contracts for the Equity Options, Index Options, and Future Options you specify. To use this model, provide a refresh_interval and a selector function. The refresh_interval defines how frequently LEAN calls the selector function. The selector function receives a datetime object that represents the current Coordinated Universal Time (UTC) and returns a list of Symbol objects. The Symbol objects you return from the selector function are the Options of the universe.

Select Language:
from Selection.OptionUniverseSelectionModel import OptionUniverseSelectionModel 

# Run universe selection asynchronously to speed up your algorithm. 
# In this case, you can't rely on the method or algorithm state between filter calls.
self.universe_settings.asynchronous = True
# Add a universe of SPY Options.
self.set_universe_selection(
    OptionUniverseSelectionModel(
        # Refresh the universe daily.
        timedelta(1), lambda _: [Symbol.create("SPY", SecurityType.OPTION, Market.USA)]
    )
)

The following table describes the arguments the model accepts:

ArgumentData TypeDescriptionDefault Value
refresh_intervaltimedeltaTime interval between universe refreshes
option_chain_symbol_selectorCallable[[datetime], List[Symbol]]A function that selects the Option symbols
universe_settingsUniverseSettingsThe universe settings. If you don't provide an argument, the model uses the algorithm.universe_settings by default.None

The following example shows how to define the Option chain Symbol selector as an isolated method:

Select Language:
from Selection.OptionUniverseSelectionModel import OptionUniverseSelectionModel 

# In the initialize method, add the OptionUniverseSelectionModel with a custom selection function.
def initialize(self) -> None:
    self.add_universe_selection(
        OptionUniverseSelectionModel(timedelta(days=1), self.select_option_chain_symbols)
    )

# Define the selection function.
def select_option_chain_symbols(self, utc_time: datetime) -> List[Symbol]:
    # Equity Options example:
    #tickers = ["SPY", "QQQ", "TLT"]
    #return [Symbol.create(ticker, SecurityType.OPTION, Market.USA) for ticker in tickers]

    # Index Options example:
    #tickers = ["VIX", "SPX"]
    #return [Symbol.create(ticker, SecurityType.INDEX_OPTION, Market.USA) for ticker in tickers]

    # Future Options example:
    future_symbol = Symbol.create(Futures.Indices.SP_500_E_MINI, SecurityType.FUTURE, Market.CME)
    return [Symbol.create_canonical_option(contract.symbol) for contract in self.futures_chain(future_symbol)]

This model uses the default Option filter, which selects all of the available Option contracts at the current time step. To use a different filter for the contracts, subclass the OptionUniverseSelectionModel and define a method. The method accepts and returns an OptionFilterUniverse object to select the Option contracts. The following table describes the methods of the OptionFilterUniverse class:

The following table describes the filter methods of the OptionFilterUniverse class:

strikes(min_strike: int, max_strike: int)

Selects contracts that are within m_strike strikes below the underlying price and max_strike strikes above the underlying price.

calls_only()

Selects call contracts.

puts_only()

Selects put contracts.

standards_only()

Selects standard contracts.

include_weeklys()

Selects non-standard weeklys contracts.

weeklys_only()

Selects weekly contracts.

front_month()

Selects the front month contract.

back_months()

Selects the non-front month contracts.

back_month()

Selects the back month contracts.

expiration(min_expiryDays: int, max_expiryDays: int)

Selects contracts that expire within a range of dates relative to the current day.

contracts(contracts: List[Symbol])

Selects a list of contracts.

contracts(contract_selector: Callable[[List[Symbol]], List[Symbol]])

Selects contracts that a selector function selects.

The preceding methods return an OptionFilterUniverse, so you can chain the methods together.

The contract filter runs at the first time step of each day.

To move the Option chain Symbol selector outside of the algorithm class, create a universe selection model that inherits the OptionUniverseSelectionModel class.

Select Language:
# In the initialize method, define the universe settings and add data.
self.universe_settings.asynchronous = True
self.add_universe_settings(EarliestExpiringAtTheMoneyCallOptionUniverseSelectionModel(self))

# Outside of the algorithm class, define the universe selection model.
class EarliestExpiringAtTheMoneyCallOptionUniverseSelectionModel(OptionUniverseSelectionModel):
    def __init__(self, algorithm):
        self.algo = algorithm
        super().__init__(timedelta(1), self.select_option_chain_symbols)
    
    def select_option_chain_symbols(self, utc_time: datetime) -> List[Symbol]:
        # Equity Options example:
        #tickers = ["SPY", "QQQ", "TLT"]
        #return [Symbol.create(ticker, SecurityType.OPTION, Market.USA) for ticker in tickers]

        # Index Options example:
        #tickers = ["VIX", "SPX"]
        #return [Symbol.create(ticker, SecurityType.INDEX_OPTION, Market.USA) for ticker in tickers]

        # Future Options example:
        future_symbol = Symbol.create(Futures.Indices.SP_500_E_MINI, SecurityType.FUTURE, Market.CME)
        return [Symbol.create_canonical_option(contract.symbol) for contract in self.algo.futures_chain(future_symbol)]
        
    # Create a filter to select contracts that have the strike price within 1 strike level and expire within 7 days.
    def Filter(self, option_filter_universe: OptionFilterUniverse) -> OptionFilterUniverse:
        return option_filter_universe.strikes(-1, -1).expiration(0, 7).calls_only()

Some of the preceding filter methods only set an internal enumeration in the OptionFilterUniverse that it uses later on in the filter process. This subset of filter methods don't immediately reduce the number of contract Symbol objects in the OptionFilterUniverse.

To override the default pricing model of the Options, set a pricing model in a security initializer.

To override the initial guess of implied volatility, set and warm up the underlying volatility model.

To view the implementation of this model, see the LEAN GitHub repository.

Option Chained Universe Selection

An Option chained universe subscribes to Option contracts on the constituents of a US Equity universe.

Select Language:
# Configure the universe to use price data unadjusted for splits and dividends ("raw") into the algorithm. 
# Options require raw Equity prices.
self.universe_settings.data_normalization_mode = DataNormalizationMode.RAW
self.universe_settings.asynchronous = True
self.add_universe_selection(
    OptionChainedUniverseSelectionModel(
        # Add a universe of the 10 most liquid US Equities.
        self.add_universe(self.universe.dollar_volume.top(10)),
        # Select call Option contracts on the underlying Equities that have the strike price within 2 strike levels.
        lambda option_filter_universe: option_filter_universe.strikes(-2, +2).front_month().calls_only()
    )
)

The following table describes the arguments the model accepts:

ArgumentData TypeDescriptionDefault Value
universeUniverseThe universe to chain onto the Option Universe Selection model
option_filterCallable[[OptionFilterUniverse], OptionFilterUniverse]The Option filter universe to use
universe_settingsUniverseSettingsThe universe settings. If you don't provide an argument, the model uses the algorithm.universe_settings by default.None

The option_filter function receives and returns an OptionFilterUniverse to select the Option contracts. The following table describes the methods of the OptionFilterUniverse class:

The following table describes the filter methods of the OptionFilterUniverse class:

strikes(min_strike: int, max_strike: int)

Selects contracts that are within m_strike strikes below the underlying price and max_strike strikes above the underlying price.

calls_only()

Selects call contracts.

puts_only()

Selects put contracts.

standards_only()

Selects standard contracts.

include_weeklys()

Selects non-standard weeklys contracts.

weeklys_only()

Selects weekly contracts.

front_month()

Selects the front month contract.

back_months()

Selects the non-front month contracts.

back_month()

Selects the back month contracts.

expiration(min_expiryDays: int, max_expiryDays: int)

Selects contracts that expire within a range of dates relative to the current day.

contracts(contracts: List[Symbol])

Selects a list of contracts.

contracts(contract_selector: Callable[[List[Symbol]], List[Symbol]])

Selects contracts that a selector function selects.

The preceding methods return an OptionFilterUniverse, so you can chain the methods together.

The following example shows how to define the Option filter as an isolated method:

Select Language:
# In the initialize method, define the universe settings and add the universe selection model.
def initialize(self) -> None:
    self.universe_settings.data_normalization_mode = DataNormalizationMode.RAW
    self.universe_settings.asynchronous = True
    self.add_universe_selection(
        OptionChainedUniverseSelectionModel(
            self.add_universe(self.universe.dollar_volume.top(10)), self.option_filter_function
        )
    )

# Define the contract filter function to select front month call contracts with a strike price within 2 strike levels.
def option_filter_function(self, option_filter_universe: OptionFilterUniverse) -> OptionFilterUniverse:
    return option_filter_universe.strikes(-2, +2).front_month().calls_only()

Some of the preceding filter methods only set an internal enumeration in the OptionFilterUniverse that it uses later on in the filter process. This subset of filter methods don't immediately reduce the number of contract Symbol objects in the OptionFilterUniverse.

To view the implementation of this model, see the LEAN GitHub repository.

Examples

The following examples demonstrate some common practices for implementing the framework Option Universe Selection Model.

Example 1: Horizontal Jelly Roll

The following algorithm selects SPX index options to construct a Jelly Roll strategy. It filters for ATM calls and puts with 30 days and 90 days till expiration. Using the SMA indicator to predict the interest rate cycle, it longs Jelly Roll if the cycle is considered uprising, otherwise selling the Jelly Roll.

Select Language:
from Selection.OptionUniverseSelectionModel import OptionUniverseSelectionModel

class FrameworkOptionUniverseSelectionAlgorithm(QCAlgorithm):
    def initialize(self) -> None:
        self.set_start_date(2023, 1, 1)
        self.set_end_date(2023, 8, 1)

        # Add a universe that selects the needed option contracts.
        self.add_universe_selection(AtmOptionHorizontalSpreadUniverseSelectionModel())
        # Add Alpha model to trade Jelly Roll, using interest rate data.
        self.add_alpha(JellyRollAlphaModel(self))
        # Invest in the same number of contracts per leg in the Jelly Roll.
        self.set_portfolio_construction(SingleSharePortfolioConstructionModel())
        
class AtmOptionHorizontalSpreadUniverseSelectionModel(OptionUniverseSelectionModel):
    # 30d update with the SelectOptionChainSymbols function since the filter returns at least 30d expiry options.
    def __init__(self) -> None:
        super().__init__(timedelta(30), self.selection_option_chain_symbols)

    def selection_option_chain_symbols(self, utc_time: datetime) -> List[Symbol]:
        # We will focus only on SPX options since they have a relatively stable dividend yield, which we assume will remain the same over time.
        # Also, assignment handling is not required since it is cash-settled.
        return [Symbol.create("SPX", SecurityType.INDEX_OPTION, Market.USA)]

    def filter(self, filter: OptionFilterUniverse) -> OptionFilterUniverse:
        # Jelly Roll is one of the best strategies for trading interest rates using options.
        # It is market-neutral but sensitive to interest rate and dividend yield changes.
        # We target to trade the market speculation between 30d and 90d options interest rate.
        return filter.jelly_roll(0, 30, 90)

class JellyRollAlphaModel(AlphaModel):
    _symbol = Symbol.create("SPX", SecurityType.INDEX_OPTION, Market.USA)
    # Use a 365d SMA indicator of daily interest rate to estimate if the interest rate cycle is upward or downward.
    _sma = SimpleMovingAverage(365)

    def __init__(self, algorithm: QCAlgorithm) -> None:
        self._algorithm = algorithm

        # Warm up the SMA indicator.
        current = algorithm.time
        provider = algorithm.risk_free_interest_rate_model
        dt = current - timedelta(365)
        while dt <= current:
            rate = provider.get_interest_rate(dt)
            self._sma.update(dt, rate)
            self._was_rising = rate > self._sma.current.value
            dt += timedelta(1)
        
        # Set a schedule to update the interest rate trend indicator every day.
        algorithm.schedule.on(
            algorithm.date_rules.every_day(),
            algorithm.time_rules.at(0, 1),
            self.update_interest_rate
        )

    def update_interest_rate(self) -> None:
        # Update interest rate to the SMA indicator to estimate its trend.
        rate = self._algorithm.risk_free_interest_rate_model.get_interest_rate(self._algorithm.time)
        self._sma.update(self._algorithm.time, rate)
        self._was_rising = rate > self._sma.current.value

    def update(self, algorithm: QCAlgorithm, slice: Slice) -> List[Insight]:
        insights = []

        # Hold one position group at a time.
        chain = slice.option_chains.get(self._symbol)
        if algorithm.portfolio.invested or not chain:
            return insights

        # Obtain the Jelly Roll constituents from the option chain.
        calls = sorted([x for x in chain if x.right == OptionRight.CALL], key=lambda x: x.expiry)
        puts = sorted([x for x in chain if x.right == OptionRight.PUT], key=lambda x: x.expiry)
        near_call = calls[0]
        far_call = calls[-1]
        near_put = puts[0]
        far_put = puts[-1]

        # Emit insight of the Jelly Roll constituents, with directions depending on the interest rate trend given by SMA.
        rate = algorithm.risk_free_interest_rate_model.get_interest_rate(algorithm.time)
        # During the rising interest rate cycle, order a long Jelly Roll.
        if rate > self._sma.current.value:
            insights.extend([
                Insight.price(near_call.symbol, timedelta(30), InsightDirection.DOWN),
                Insight.price(far_call.symbol, timedelta(30), InsightDirection.UP),
                Insight.price(near_put.symbol, timedelta(30), InsightDirection.UP),
                Insight.price(far_put.symbol, timedelta(30), InsightDirection.DOWN)
            ])
        # During the downward interest rate cycle, order short Jelly Roll.
        elif rate < self._sma.current.value:
            insights.extend([
                Insight.price(near_call.symbol, timedelta(30), InsightDirection.UP),
                Insight.price(far_call.symbol, timedelta(30), InsightDirection.DOWN),
                Insight.price(near_put.symbol, timedelta(30), InsightDirection.DOWN),
                Insight.price(far_put.symbol, timedelta(30), InsightDirection.UP)
            ])
        # If the interest rate cycle is steady for a long, we expect a flip in the cycle coming up.
        elif self._was_rising:
            insights.extend([
                Insight.price(near_call.symbol, timedelta(30), InsightDirection.UP),
                Insight.price(far_call.symbol, timedelta(30), InsightDirection.DOWN),
                Insight.price(near_put.symbol, timedelta(30), InsightDirection.DOWN),
                Insight.price(far_put.symbol, timedelta(30), InsightDirection.UP)
            ])
        else:
            insights.extend([
                Insight.price(near_call.symbol, timedelta(30), InsightDirection.DOWN),
                Insight.price(far_call.symbol, timedelta(30), InsightDirection.UP),
                Insight.price(near_put.symbol, timedelta(30), InsightDirection.UP),
                Insight.price(far_put.symbol, timedelta(30), InsightDirection.DOWN)
            ])
        
        return insights

class SingleSharePortfolioConstructionModel(PortfolioConstructionModel):
    def create_targets(self, algorithm: QCAlgorithm, insights: List[Insight]) -> List[PortfolioTarget]:
        targets = []
        for insight in insights:
            if algorithm.securities[insight.symbol].is_tradable:
                # Use integer target to create a portfolio target to trade a single contract
                targets.append(PortfolioTarget(insight.symbol, insight.direction))
        return targets

The following example chains a fundamental universe and an Equity Options universe. It first selects 10 stocks with the lowest PE ratio and then selects their front-month call Option contracts. It buys one front-month call Option contract every day.

To override the default pricing model of the Options, set a pricing model in a security initializer.

To override the initial guess of implied volatility, set and warm up the underlying volatility model.

Select Language:
# Example code to chain a fundamental universe and an Equity Options universe by selecting top 10 stocks with lowest PE, indicating potentially undervalued stocks and then selecting their from-month call Option contracts to target contracts with high liquidity.
from AlgorithmImports import *

    class ChainedUniverseAlgorithm(QCAlgorithm):
    def initialize(self):
        self.set_start_date(2023, 2, 2)
        self.set_cash(100000)
        self.universe_settings.asynchronous = True
        self.universe_settings.data_normalization_mode = DataNormalizationMode.RAW
        self.set_security_initializer(CustomSecurityInitializer(self))

        universe = self.add_universe(self.fundamental_function)
        self.add_universe_options(universe, self.option_filter_function)
        self.day = 0

    def fundamental_function(self, fundamental: List[Fundamental]) -> List[Symbol]:
        filtered = (f for f in fundamental if not np.isnan(f.valuation_ratios.pe_ratio))
        sorted_by_pe_ratio = sorted(filtered, key=lambda f: f.valuation_ratios.pe_ratio)
        return [f.symbol for f in sorted_by_pe_ratio[:10]]

    def option_filter_function(self, option_filter_universe: OptionFilterUniverse) -> OptionFilterUniverse:
        return option_filter_universe.strikes(-2, +2).front_month().calls_only()

    def on_data(self, data: Slice) -> None:
        if self.is_warming_up or self.day == self.time.day:
            return
        
        for symbol, chain in data.option_chains.items():
            if self.portfolio[chain.underlying.symbol].invested:
                self.liquidate(chain.underlying.symbol)

            spot = chain.underlying.price
            contract = sorted(chain, key=lambda x: abs(spot-x.strike))[0]
            tag = f"IV: {contract.implied_volatility:.3f} Δ: {contract.greeks.delta:.3f}"
            self.market_order(contract.symbol, 1, True, tag)
            self.day = self.time.day

class CustomSecurityInitializer(BrokerageModelSecurityInitializer):
    def __init__(self, algorithm: QCAlgorithm) -> None:
        super().__init__(algorithm.brokerage_model, FuncSecuritySeeder(algorithm.get_last_known_prices))
        self.algorithm = algorithm

    def initialize(self, security: Security) -> None:
        # First, call the superclass definition
        # This method sets the reality models of each security using the default reality models of the brokerage model
        super().initialize(security)

        # Overwrite the price model        
        if security.type == SecurityType.OPTION: # Option type
            security.price_model = OptionPriceModels.crank_nicolson_fd()

        # Overwrite the volatility model and warm it up
        if security.type == SecurityType.EQUITY:
            security.volatility_model = StandardDeviationOfReturnsVolatilityModel(30)
            trade_bars = self.algorithm.history[TradeBar](security.symbol, 30, Resolution.DAILY)
            for trade_bar in trade_bars:
                security.volatility_model.update(security, trade_bar)

You can also see our Videos. You can also get in touch with us via Discord.

Did you find this page helpful?

Contribute to the documentation: