Backtest

Overall Statistics
Total Orders 31893 Average Win 0.10% Average Loss -0.10% Compounding Annual Return -28.044% Drawdown 99.300% Expectancy -0.014 Start Equity 100000000 End Equity 670258.09 Net Profit -99.330% Sharpe Ratio -1.205 Sortino Ratio -0.746 Probabilistic Sharpe Ratio 0% Loss Rate 53% Win Rate 47% Profit-Loss Ratio 1.09 Alpha -0.204 Beta -0.026 Annual Standard Deviation 0.171 Annual Variance 0.029 Information Ratio -1.285 Tracking Error 0.224 Treynor Ratio 7.793 Total Fees $4858827.66 Estimated Strategy Capacity $77000.00 Lowest Capacity Asset JNPR YQYHC5QU2XYE\|JNPR RLTVP7P85ZS5 Portfolio Turnover 4.58%

#region imports
from AlgorithmImports import *
#endregion

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

        self._algo: QCAlgorithm = algo
        self._symbol: Symbol = symbol
        self._low_period: int = low_period
        self._min: Optional[Minimum] = 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_min(self) -> None:
        self._min = self._algo.min(self._symbol, self._low_period, Resolution.DAILY)
        self._algo.indicator_history(self._min, self._symbol, self._low_period, Resolution.DAILY, Field.LOW)

    def get_open_interest(self) -> float:
        return self._option.open_interest
    
    def get_low(self) -> float:
        return self._min.current.value

    def is_ready(self) -> bool:
        return self._min is not None and self._min.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-low/
#
# 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-low (PTL) ratio for each stock, defined as the stock price divided by its 52-week low. Next, multiply the ratio by minus one so that an increase in PTL means
# the stock price is moving closer to its 52-week low. Third, sort the stocks into quintiles based on their PTL ratio, where the highest quintile consists of the highest PTL stocks, and the lowest quintile consists of the lowest PTL stocks. Fourth, for each stock in the highest 
# quintile, buy one put option and delta-hedge in the next month with daily rebalancing. Conversely, for each stock in the lowest quintile, sell one put 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 OptionTradingandReturnsversusthe52WeekLow(QCAlgorithm):

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

        self._quantile: int = 10
        self._low_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_PTL: List[Symbol] = []
        self._low_PTL: 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_min()

        # 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._low_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 PTL
            PTL: Dict[Symbol, float] = {
                symbol : slice[symbol].price / self._symbol_data[symbol].get_low() 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(PTL) < self._quantile:
                self.liquidate()
                return

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

            self._high_PTL = sorted_by_PTL[-quantile:]
            self._low_PTL = sorted_by_PTL[:quantile]

            # find ATM option
            for symbol in self._high_PTL + self._low_PTL:
                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_ptl: float = sum([
                self._symbol_data[symbol].get_open_interest()
                for symbol in self._high_PTL
                if self._symbol_data[symbol].option is not None
            ])

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

            total_oi: List[float] = [total_oi_high_ptl, total_oi_low_ptl]
            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_PTL, self._low_PTL]):
                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()