Backtest

Overall Statistics
Total Orders 30887 Average Win 0.12% Average Loss -0.18% Compounding Annual Return -28.429% Drawdown 99.400% Expectancy -0.250 Start Equity 100000000 End Equity 617738.13 Net Profit -99.382% Sharpe Ratio -1.438 Sortino Ratio -0.863 Probabilistic Sharpe Ratio 0% Loss Rate 55% Win Rate 45% Profit-Loss Ratio 0.65 Alpha -0.211 Beta -0.035 Annual Standard Deviation 0.149 Annual Variance 0.022 Information Ratio -1.419 Tracking Error 0.209 Treynor Ratio 6.104 Total Fees $4911524.31 Estimated Strategy Capacity $0 Lowest Capacity Asset WATR YQYHC5R5ZM1Y\|WATR R735QTJ8XC9X Portfolio Turnover 4.61%

#region imports
from AlgorithmImports import *
#endregion

class SymbolData:
    def __init__(
        self, 
        algo: QCAlgorithm, 
        symbol: Symbol, 
        high_period: int
    ) -> None:

        self._algo: QCAlgorithm = algo
        self._symbol: Symbol = symbol
        self._high_period: int = high_period
        self._max: Optional[Maximum] = None

        self._option: Optional[Option] = None

    @property
    def symbol(self) -> Symbol:
        return self._symbol

    @property
    def option(self) -> Optional[Option]:
        return self._option

    def reset_option(self) -> None:
        if self._option is not None:
            self._algo.remove_security(self._option.symbol)

        self._option = None

    def init_max(self) -> None:
        self._max = self._algo.max(self._symbol, self._high_period, Resolution.DAILY, Field.HIGH)
        self._algo.indicator_history(self._max, self._symbol, self._high_period, Resolution.DAILY)

    def get_open_interest(self) -> float:
        return self._option.open_interest
    
    def get_high(self) -> float:
        return self._max.current.value

    def is_ready(self) -> bool:
        return self._max is not None and self._max.is_ready

    def _subscribe_option(self, option_symbol: symbol) -> None:
        option: Option = self._algo.add_option_contract(option_symbol, Resolution.DAILY)
        option.price_model = OptionPriceModels.black_scholes()
        option.set_option_assignment_model(NullOptionAssignmentModel())
        self._option = option

    def find_atm_option(
        self, 
        min_expiry: int, 
        max_expiry: int, 
        option_right: OptionRight
    ) -> None:

        option_chain: DataFrame = self._algo.option_chain(self._symbol, flatten=True).data_frame
        if option_chain.empty:
            self._algo.log(f'No option chain found for {self._symbol}')
            return

        underlying_price: float = option_chain.underlyinglastprice[0]
        calls: DataFrame = option_chain[option_chain.right == option_right]
        if 'openinterest' not in calls.columns:
            self._algo.log(f'No open interest data found for {self._symbol} options')
            return

        calls = calls[(calls.expiry - self._algo.time).dt.days >= min_expiry][(calls.expiry - self._algo.time).dt.days <= max_expiry]
        if len(calls) == 0:
            self._algo.log(f'No target expiry ({min_expiry}, {max_expiry}) found for {self._symbol} options')
            return
        
        target_strike: float = min(calls.strike, key=lambda x: abs(abs(x) - underlying_price))
        calls = calls[calls.strike == target_strike]
        
        if len(calls) == 0:
            self._algo.log(f'No call options found for {self._symbol} with strike: {target_strike}')
            return

        option_symbol: Symbol = calls.index[0]
        
        self._subscribe_option(option_symbol)

    def get_amount_as_fraction_of_portfolio(self, fraction: float) -> float:
        multiplier: int = self._algo.securities[self._option.symbol].contract_multiplier
        target_notional: float = self._algo.portfolio.total_portfolio_value * fraction
        notional_of_contract: float = multiplier * self._algo.securities[self._option.symbol.underlying].price
        amount: float = target_notional / notional_of_contract

        return amount

    def get_amount_as_fraction_of_cash(self, fraction: float) -> float:
        holding_value: float = self._algo.portfolio.cash
        multiplier: int = self._algo.securities[self._option.symbol].contract_multiplier
        target_notional: float = holding_value * fraction
        notional_of_contract: float = multiplier * self._algo.securities[self._option.symbol.underlying].price
        amount: float = target_notional / notional_of_contract

        return amount

# custom fee model
class CustomFeeModel(FeeModel):
    def GetOrderFee(self, parameters):
        fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
        return OrderFee(CashAmount(fee, "USD"))

# region imports
from AlgorithmImports import *
from abc import abstractmethod, ABC
# endregion

class DeltaHedgeModule(ABC):
    def __init__(
        self, 
        algo: QCAlgorithm, 
    ) -> None:

        self._algo: QCAlgorithm = algo

    @abstractmethod
    def delta_hedge(self, symbol: Symbol) -> None:
        ...

    @abstractmethod
    def _get_delta_hedge_quantity(self, symbol: Symbol) -> float:
        ...

# region imports
from AlgorithmImports import *
from delta_hedge_module import DeltaHedgeModule
# endregion

class ETFOptionsDeltaHedgeModule(DeltaHedgeModule):

    def __init__(
        self, 
        algo: QCAlgorithm
    ) -> None:

        super(ETFOptionsDeltaHedgeModule, self).__init__(algo)

    def delta_hedge(self, symbol: Symbol) -> None:
        delta_hedge_quantity: float = self._get_delta_hedge_quantity(symbol)
        if delta_hedge_quantity != 0:
            self._algo.market_order(symbol, round(delta_hedge_quantity))

    def _get_delta_hedge_quantity(self, symbol: Symbol) -> float:
        chain = self._algo.option_chain(symbol)

        if not chain:
            self._algo.log(f'DeltaHedgeModule._get_delta_hedge_quantity: No option chain found for {symbol}.')
            return 0.
        
        contracts_in_portfolio: List[OptionContract] = [
            contract.value for contract in chain.contracts
            if self._algo.securities.contains_key(contract.value.symbol)
            and self._algo.securities[contract.value.symbol].type == SecurityType.OPTION
            and self._algo.portfolio[contract.value.symbol].invested
        ]

        delta: float = sum([
            self._algo.portfolio[c.symbol].quantity * c.greeks.delta * self._algo.securities[c].contract_multiplier
            for c in contracts_in_portfolio
        ])
        delta_hedge_quantity: float = -(delta)
        amount_to_hedge: float = delta_hedge_quantity - self._algo.portfolio[symbol].quantity

        return amount_to_hedge

# https://quantpedia.com/strategies/option-trading-and-returns-versus-the-52-week-high/
#
# The investment universe consists of all stocks from the CRSP database and their underlying options contracts. First, apply the following screening procedure to the options universe 
# and retain only those which meet the following criteria: trading volume is non-zero, maturity is between 30 and 60 days (i.e., only options maturing in the month following the next),
# moneyness (defined as the exercise price over the stock price) is in the range of (0.8, 1.2), the bid quote and bid-ask spread are positive, and the percentage bid-ask spread (i.e., 
# bid-ask spread divided by the midpoint) is less than 100%. Second, on the last trading day of each month, calculate the price-to-high (PTH) ratio for each stock, defined as the stock 
# price divided by its 52-week high. Third, sort the stocks into quintiles based on their PTH ratio, where the highest quintile consists of the highest PTH stocks, and the lowest quintile 
# consists of the lowest PTH stocks. Fourth, for each stock in the highest quintile, buy one call option and delta-hedge in the next month with daily rebalancing. Conversely, for each 
# stock in the lowest quintile, sell one call option and delta-hedge in the next month with daily rebalancing. Invest/borrow the net balance at the risk-free rate. Positions are 
# value-weighted based on the dollar value of open interest and rebalanced monthly.
#
# QC Implementation changes:
#   - The investment universe consists of 500 most liquid US stocks with price >= 5$.

# region imports
from AlgorithmImports import *
from data_tools import CustomFeeModel, SymbolData
from delta_hedge.etf_options_delta_hedge_module import ETFOptionsDeltaHedgeModule
# endregion

class OptionTradingandReturnsversusthe52WeekHigh(QCAlgorithm):

    def initialize(self) -> None:
        self.set_start_date(2010, 1, 1)
        self.set_cash(100_000_000)

        self._quantile: int = 10
        self._high_period: int = 52 * 5
        self._min_expiry: int = 30
        self._max_expiry: int = 60
        self.leverage: int = 5
        
        self._symbol_data: Dict[Symbol, SymbolData] = {}
        self._selected_stock_symbols: List[Symbol] = []
        self._high_PTH: List[Symbol] = []
        self._low_PTH: List[Symbol] = []

        # delta hedge module
        self._delta_hedge_module: ETFOptionsDeltaHedgeModule = ETFOptionsDeltaHedgeModule(self)

        self._fundamental_count: int = 500
        self._fundamental_sorting_key = lambda x: x.dollar_volume

        # universe settings
        self._min_share_price: float = 5.
        self._exchanges: List[str] = ['NYS', 'NAS', 'ASE']
        self.add_universe(self._fundamental_selection_function)
        self.universe_settings.resolution = Resolution.DAILY
        self.universe_settings.leverage = self.leverage
        self.universe_settings.data_normalization_mode = DataNormalizationMode.RAW
        self.settings.minimum_order_margin_portfolio_percentage = 0.
        self.settings.daily_precise_end_time = False
        self._traded_percentage: float = .5

        self._market: Symbol = self.add_equity('SPY', Resolution.DAILY).symbol
       
        # get last known price after subscribing option contract
        seeder: FuncSecuritySeeder = FuncSecuritySeeder(self.get_last_known_prices)
        self.set_security_initializer(lambda security: seeder.seed_security(security))

        # schedule
        self._selection_flag: bool = False
        self._execution_flag: bool = False
        self.schedule.on(
            self.date_rules.month_end(self._market), 
            self.time_rules.after_market_close(self._market, 0), 
            self._selection
        )
        self.schedule.on(
            self.date_rules.week_end(self._market), 
            self.time_rules.after_market_close(self._market, 0), 
            self._delta_hedge
        )

    def on_securities_changed(self, changes: SecurityChanges) -> None:
        # initialize symbol data with indicator
        for security in changes.added_securities:
            if security.type == SecurityType.EQUITY:
                if security.symbol != self._market:
                    security.set_fee_model(CustomFeeModel())
                    security.set_leverage(2)

                    if security.symbol in self._symbol_data:
                        self._symbol_data[security.symbol].init_max()

        # remove symbol from collection
        # price indicator will need to be initialized once the symbol is subscribed again
        for security in changes.removed_securities:
            symbol: Symbol = security.symbol
            if symbol in self._symbol_data:
                del self._symbol_data[symbol]

    def _fundamental_selection_function(self, fundamental: List[Fundamental]) -> List[Symbol]:
        if not self._selection_flag:
            return Universe.UNCHANGED

        selected: List[Fundamental] = [
            x for x in fundamental if x.has_fundamental_data
            and x.market == 'usa'
            and x.market_cap != 0
            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]
            ]

        for stock in selected:
            symbol: Symbol = stock.symbol

            if symbol in self._symbol_data:
                continue

            self._symbol_data[symbol] = SymbolData(self, symbol, self._high_period)

        self._selected_stock_symbols = list(map(lambda x: x.symbol, selected))
        return self._selected_stock_symbols
    
    def _delta_hedge(self) -> None:
        # delta hedging
        for symbol in self._symbol_data:
            self._delta_hedge_module.delta_hedge(symbol)

    def on_data(self, slice: Slice) -> None:
        # rebalance monthly
        if self._selection_flag:
            self._selection_flag = False

            # calculate PTH
            PTH: Dict[Symbol, float] = {
                symbol : slice[symbol].price / self._symbol_data[symbol].get_high() for symbol in self._selected_stock_symbols if \
                symbol in self._symbol_data and self._symbol_data[symbol].is_ready() and \
                slice.contains_key(symbol) and slice[symbol]
            }

            if len(PTH) < self._quantile:
                self.liquidate()
                return

            # sorting
            quantile: int = len(PTH) // self._quantile
            sorted_by_PTH: List[Symbol] = [
                x[0] for x in sorted(PTH.items(), key=lambda item: item[1])
            ]

            self._high_PTH = sorted_by_PTH[-quantile:]
            self._low_PTH = sorted_by_PTH[:quantile]

            # find ATM option
            for symbol in self._high_PTH + self._low_PTH:
                self._symbol_data[symbol].find_atm_option(
                    self._min_expiry, 
                    self._max_expiry, 
                    OptionRight.CALL
                )
            
            # NOTE add 1 day lag between option subscription and weighting/trade execution
            self._execution_flag = True
        else:
            if not self._execution_flag:
                return
            self._execution_flag = False
        
            # get open interest
            total_oi_high_pth: float = sum([
                self._symbol_data[symbol].get_open_interest()
                for symbol in self._high_PTH
                if self._symbol_data[symbol].option is not None
            ])

            total_oi_low_pth: float = sum([
                self._symbol_data[symbol].get_open_interest()
                for symbol in self._low_PTH
                if self._symbol_data[symbol].option is not None
            ])

            total_oi: List[float] = [total_oi_high_pth, total_oi_low_pth]
            if any(oi == 0 for oi in total_oi):
                self.liquidate()
                self.log(f'missing open interest sum value')
                return

            # trade execution
            for i, portfolio in enumerate([self._high_PTH, self._low_PTH]):
                for symbol in portfolio:
                    option: Option = self._symbol_data[symbol].option
                    if option is not None:
                        self.securities[option.symbol].is_tradable = True

                        weight: float = self._symbol_data[symbol].get_open_interest() / total_oi[i]
                        q: float =  self._symbol_data[symbol].get_amount_as_fraction_of_portfolio(((-1) ** i) * self._traded_percentage * weight)
                        if abs(q) > 1:
                            self.market_order(option.symbol, q)

    def _selection(self) -> None:
        self._selection_flag = True
        self.liquidate()
        
        # clear stored options
        for symbol, symbol_data in self._symbol_data.items():
            symbol_data.reset_option()