Trade U.S. equity options by forming delta straddles, going long on highest-return quintile and short on lowest, based on 12-month average returns, holding positions to expiration.

I. STRATEGY IN A NUTSHELL

Trade U.S. equity options using delta-neutral straddles. Select at-the-money calls and puts each month, rank stocks by 11-month lagged returns, and go long on top quintile, short on bottom quintile. Rebalance monthly and hold straddles to expiration.

II. ECONOMIC RATIONALE

Momentum comes from underreaction to past volatility shocks and seasonal patterns in realized volatility. Strong performers over 6–36 months tend to continue performing well, while short-term reversals are limited to one-month lookbacks.

III. SOURCE PAPER

Momentum, Reversal, and Seasonality in Option Returns [Click to Open PDF]

Christopher S. Jones, Mehdi Khorram, Haitao Mo, University of Southern California – Marshall School of Business – Finance and Business Economics Department, Rochester Institute of Technology (RIT), University of Kansas

<Abstract>

Option returns display substantial momentum using formation periods ranging from 6 to 36 months long, with long/short portfolios obtaining annualized Sharpe ratios above 1.5. In the short term, option returns exhibit reversal. Options also show marked seasonality at multiples of three and 12 monthly lags. All of these results are highly significant and stable in the cross section and over time. They remain strong after controlling for other characteristics, and momentum and seasonality survive factor risk-adjustment. Momentum is mainly explained by an underreaction to past volatility and other shocks, while seasonality reflects unpriced seasonal variation in stock return volatility.

IV. BACKTEST PERFORMANCE

Annualised Return114.83%
Volatility41.56%
Beta0.112
Sharpe Ratio2.76
Sortino Ratio-0.78
Maximum DrawdownN/A
Win Rate41%

V. FULL PYTHON CODE

from AlgorithmImports import *
from typing import List, Dict
from dataclasses import dataclass
#endregion
class ReversalOnStraddles(QCAlgorithm):
    def Initialize(self) -> None:
        self.SetStartDate(2015, 1, 1)
        self.SetCash(100_000)
        
        self.leverage: int = 20
        self.quantile: int = 5
        self.min_share_price: int = 5
        self.min_expiry: int = 20
        self.max_expiry: int = 30
        
        self.min_daily_period: int = 14          # need n straddle prices
        self.monthly_period: int = 12            # monthly straddle performance values
        
        self.last_fundamental: List[Symbol] = []
        
        self.straddle_price_sum: Dict[Symbol, float] = {}        # call and put price sum
        self.subscribed_contracts: Dict[Symbol, Contracts] = {}      # subscribed option universe
        self.monthly_straddle_returns: Dict[Symbol, RollingWindow] = {}  # monthly straddle return values
        
        # initial data feed
        self.AddEquity('SPY', Resolution.Daily)
        
        self.recent_year: int = -1
        self.fundamental_count: int = 100
        self.fundamental_sorting_key = lambda x: x.DollarVolume
        self.selection_flag: bool = False
        self.rebalance_flag: bool = False
        self.UniverseSettings.Leverage = self.leverage
        self.UniverseSettings.Resolution = Resolution.Daily
        self.AddUniverse(self.FundamentalSelectionFunction)
        self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
        self.settings.daily_precise_end_time = False
        self.SetSecurityInitializer(lambda x: x.SetDataNormalizationMode(DataNormalizationMode.Raw))
        self.UniverseSettings.DataNormalizationMode = DataNormalizationMode.Raw
        
    def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
        for security in changes.AddedSecurities:
            security.SetFeeModel(CustomFeeModel())
    def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
        # rebalance monthly
        if not self.selection_flag:
            return Universe.Unchanged
        
        self.rebalance_flag = True
        if self.Time.month % 12 != 0 or self.recent_year == self.Time.year:
            return self.last_fundamental
        
        self.recent_year = self.Time.year
        
        # filter top n U.S. stocks by dollar volume
        selected: List[Fundamental] = [x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and x.Price > self.min_share_price]
        
        if len(selected) > self.fundamental_count:
            selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]        
        # make sure monthly returns are consecutive
        for stock in selected:
            symbol: Symbol = stock.Symbol
            if symbol not in self.last_fundamental:
                if symbol in self.monthly_straddle_returns:
                    del self.monthly_straddle_returns[symbol]
        # initialize new fundamental 
        self.last_fundamental = [x.Symbol for x in selected]
        # return newly selected symbols
        return self.last_fundamental
        
    def OnData(self, data: Slice) -> None:
        # execute once a day
        # if not (self.Time.hour == 9 and self.Time.minute == 31):
        #     return
        
        for symbol in self.last_fundamental:
            # check if any of the subscribed contracts expired
            if symbol in self.subscribed_contracts and self.subscribed_contracts[symbol].expiry_date - timedelta(days=1) <= self.Time.date():
                # remove expired contracts
                for contract in self.subscribed_contracts[symbol].contracts:
                    self.Liquidate(contract)
                    
                # liquidate hedge
                if self.Portfolio[symbol].Quantity != 0:
                    self.MarketOrder(symbol, -self.Portfolio[symbol].Quantity)
                # remove Contracts object for current symbol
                del self.subscribed_contracts[symbol]
            
            # check if stock has subscribed contracts
            elif symbol in self.subscribed_contracts:
                atm_call, atm_put = self.subscribed_contracts[symbol].contracts
                
                if atm_call in data and atm_put in data and data[atm_call] and data[atm_put]:
                    # store straddle price
                    atm_call_price: float = data[atm_call].Value
                    atm_put_price: float = data[atm_put].Value
                    
                    # store straddle sum price
                    straddle_price_sum: float = atm_call_price + atm_put_price
                    if symbol not in self.straddle_price_sum:
                        self.straddle_price_sum[symbol] = []
                    self.straddle_price_sum[symbol].append(straddle_price_sum)
                    
        # perform next selection, when there are no active contracts
        if len(self.subscribed_contracts) == 0 and not self.selection_flag:
            # liquidate leftovers
            if self.Portfolio.Invested:
                self.Liquidate()
            
            self.selection_flag = True
            return
        
        # subscribe to new contracts after selection
        if len(self.subscribed_contracts) == 0 and self.selection_flag:
            self.selection_flag = False
            
            for symbol in self.last_fundamental:
                if self.Securities[symbol].IsDelisted:
                    continue
                # get all contracts for current stock symbol
                contracts: List[Symbol] = self.OptionChainProvider.GetOptionContractList(symbol, self.Time)
                # get current price for etf
                underlying_price = self.Securities[symbol].Price
                
                # get strikes from commodity future contracts
                strikes: List[float] = [i.ID.StrikePrice for i in contracts]
                
                # can't filter contracts, if there isn't any strike price
                if len(strikes) <= 0 or underlying_price == 0:
                    continue
                
                # filter calls and puts contracts with one month expiry
                calls, puts = self.FilterContracts(strikes, contracts, underlying_price)
                
                # make sure, there is at least one call and put contract
                if len(calls) > 0 and len(puts) > 0:
                    # sort by expiry
                    call: Symbol = sorted(calls, key = lambda x: x.ID.Date, reverse=True)[0]
                    put: Symbol = sorted(puts, key = lambda x: x.ID.Date, reverse=True)[0]
                    
                    subscriptions = self.SubscriptionManager.SubscriptionDataConfigService.GetSubscriptionDataConfigs(call.Underlying)
                    # check if stock's call and put contract was successfully subscribed
                    if subscriptions:
                        # add call contract
                        self.AddOptionContract(call, Resolution.Daily)
                        # add put contract
                        self.AddOptionContract(put, Resolution.Daily)
                        
                        # retrieve expiry date for contracts
                        expiry_date: datetime.date = call.ID.Date.date() if call.ID.Date.date() < put.ID.Date.date() else put.ID.Date.date()
                        # store contracts with expiry date under stock's symbol
                        self.subscribed_contracts[symbol] = Contracts(expiry_date, underlying_price, [call, put])
            
            return  # one day skip for rebalance
            
        # trade subscribed options
        if len(self.subscribed_contracts) != 0 and self.rebalance_flag:
            momentum: Dict[Symbol, float] = {}
            
            for symbol in self.subscribed_contracts:
                # make sure stock's symbol was lastly selected in fundamental and have straddle prices ready
                if symbol not in self.last_fundamental or symbol not in self.straddle_price_sum or not len(self.straddle_price_sum[symbol]) > self.min_daily_period:
                    continue
                
                # calculate straddle performance
                straddle_performance: float = self.straddle_price_sum[symbol][-1] / self.straddle_price_sum[symbol][0] - 1
                
                # calculate average straddle performance
                if symbol not in self.monthly_straddle_returns:
                    self.monthly_straddle_returns[symbol] = RollingWindow[float](self.monthly_period)
                self.monthly_straddle_returns[symbol].Add(straddle_performance)
                
                # calculate straddle momentum
                if self.monthly_straddle_returns[symbol].IsReady:
                    momentum[symbol] = np.mean([x for x in self.monthly_straddle_returns[symbol]][1:])  # skip last month
                
                # reset straddle prices for next month
                self.straddle_price_sum[symbol] = []
            # make sure there are enough stock's for quintile selection
            if len(momentum) < self.quantile:
                self.rebalance_flag = False
                return
            
            # perform quintile selection
            quantile: int = int(len(momentum) / self.quantile)
            sorted_by_momentum: List[Symbol] = [x[0] for x in sorted(momentum.items(), key=lambda item: item[1]) if x[0] in data and data[x[0]]]
            
            # long the quintile with the highest return
            long: List[Symbol] = sorted_by_momentum[-quantile:]
            
            # short the quintile with the lowest return
            short: List[Symbol] = sorted_by_momentum[:quantile]
            
            # trade long
            self.TradeOptions(long, True)
            # trade short
            self.TradeOptions(short, False)
        
        self.rebalance_flag = False
        
    def FilterContracts(self, strikes: List[float], contracts: List[Symbol], underlying_price: float) -> Symbol:
        ''' filter call and put contracts from contracts parameter '''
        ''' return call and put contracts '''
        
        # Straddle
        call_strike: float = min(strikes, key=lambda x: abs(x-underlying_price))
        put_strike: float = call_strike
        
        calls: List[Symbol] = [] # storing call contracts
        puts: List[Symbol] = [] # storing put contracts
        
        for contract in contracts:
            # check if contract has one month expiry
            if self.min_expiry < (contract.ID.Date - self.Time).days < self.max_expiry:
                # check if contract is call
                if contract.ID.OptionRight == OptionRight.Call and contract.ID.StrikePrice == call_strike:
                    calls.append(contract)
                # check if contract is put
                elif contract.ID.OptionRight == OptionRight.Put and contract.ID.StrikePrice == put_strike:
                    puts.append(contract)
        
        # return filtered calls and puts with one month expiry
        return calls, puts
        
    def TradeOptions(self, symbols: List[Symbol], long_flag: bool) -> None:
        ''' on long signal buy call and put option contract '''
        ''' on short signal sell call and put option contract '''
        length: int = len(symbols)
        
        # trade etf's call and put contracts
        for symbol in symbols:
            # get call and put contract
            call, put = self.subscribed_contracts[symbol].contracts
            
            # get underlying price
            underlying_price: float = self.subscribed_contracts[symbol].underlying_price
            
            # calculate option and hedge quantity
            options_q: int = int((self.Portfolio.TotalPortfolioValue / length) / (underlying_price * 100))
            hedge_q: int = options_q*50
            
            if long_flag:
                self.Buy(call, options_q)
                self.Buy(put, options_q)
                
                # initial delta hedge
                self.Sell(symbol, hedge_q)
            else:
                self.Sell(call, options_q)
                self.Sell(put, options_q)
                
                # initial delta hedge
                self.Buy(symbol, hedge_q)
@dataclass
class Contracts():
    expiry_date: datetime.date
    underlying_price: float
    contracts: List[Symbol] 
        
# Custom fee model
class CustomFeeModel(FeeModel):
    def GetOrderFee(self, parameters: OrderFeeParameters) -> OrderFee:
        fee: float = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
        return OrderFee(CashAmount(fee, "USD"))

VI. Backtest Performance

Leave a Reply

Discover more from Quant Buffet

Subscribe now to keep reading and get access to the full archive.

Continue reading