Trade U.S. equity options by forming delta straddles, going long on stocks with lowest straddle returns and short on highest, holding zero-delta straddles to maturity monthly.

I. STRATEGY IN A NUTSHELL

The strategy trades U.S. equity options with monthly expirations, forming delta-neutral straddles at-the-money. Stocks are sorted into quintiles based on straddle returns over the past month. The strategy goes long on the lowest-return quintile and short on the highest, holding positions to expiration.

II. ECONOMIC RATIONALE

Momentum in delta-hedged options arises from underreaction to past volatility shocks and seasonal variations in realized volatility. Short-term reversals occur as strong-performing options one month tend to underperform the next, linking straddle payoffs to underlying stock price dynamics.

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 Return49.71%
Volatility36.8%
Beta-0.051
Sharpe Ratio1.35
Sortino Ratio-1.202
Maximum DrawdownN/A
Win Rate37%

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.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
        
        # initial data feed
        self.AddEquity('SPY', Resolution.Daily)
        
        self.fundamental_count: int = 200
        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.daily_precise_end_time = False
        self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
        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
        
        # 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]]
        
        # 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 = data[atm_call].Value
                    atm_put_price = data[atm_put].Value
                    
                    # store straddle sum price
                    straddle_price_sum = 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: float = 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:
            straddle_performance: 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[symbol] = self.straddle_price_sum[symbol][-1] / self.straddle_price_sum[symbol][0] - 1
                
                # reset straddle prices for next month
                self.straddle_price_sum[symbol] = []
            # make sure there are enough stock's for quintile selection
            if len(straddle_performance) < self.quantile:
                self.rebalance_flag = False
                return
            
            # perform quintile selection
            quantile: int = int(len(straddle_performance) / self.quantile)
            sorted_by_straddle_perf: List[Symbol] = [x[0] for x in sorted(straddle_performance.items(), key=lambda item: item[1]) if x[0] in data and data[x[0]]]
            
            # long straddles with lowest performance
            long: List[Symbol] = sorted_by_straddle_perf[:quantile]
            # short straddles with highest performance
            short: List[Symbol] = sorted_by_straddle_perf[-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