Tail Risk Hedging with Cheap Options

1from AlgorithmImports import *
2from typing import List, Dict
3# endregion
4
5class TailRiskHedgingwithCheapOptions(QCAlgorithm):
6
7def Initialize(self):
8self.SetStartDate(2010, 1, 1)
9self.SetCash(1_000_000)
10
11seeder = FuncSecuritySeeder(self.GetLastKnownPrices)
12self.SetSecurityInitializer(lambda security: seeder.SeedSecurity(security))
13
14self.leverage: int = 5
15self.quantile: int = 5
16
17self.min_expiry: int = 6 * 30
18self.max_expiry: int = 12 * 30
19self.min_delta: float = 0.1
20
21self.exchanges: List[str] = ['NYS', 'NAS', 'ASE']
22
23self.active_stock_universe: List[Symbol] = []
24self.monthly_subscribed_contracts: List[OptionContract] = []
25
26self.market: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
27
28self.option_port_weight: float = 0.02
29self.market_port_weight: float = 0.98
30
31self.fundamental_count: int = 100
32self.fundamental_sorting_key = lambda x: x.MarketCap
33
34self.selection_flag: bool = False
35self.rebalance_flag: bool = False
36self.UniverseSettings.Resolution = Resolution.Daily
37self.AddUniverse(self.FundamentalSelectionFunction)
38self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
39self.UniverseSettings.DataNormalizationMode = DataNormalizationMode.Raw
40
41self.Schedule.On(self.DateRules.MonthStart(self.market), self.TimeRules.BeforeMarketClose(self.market, 0), self.Selection)
42
43def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
44for security in changes.AddedSecurities:
45    security.SetFeeModel(CustomFeeModel())
46    security.SetLeverage(self.leverage)
47
48def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
49if not self.selection_flag:
50    return Universe.Unchanged
51
52selected: List[Fundamental] = [
53    f for f in fundamental if f.HasFundamentalData and \
54    f.MarketCap != 0 and \
55    f.SecurityReference.ExchangeId in self.exchanges
56]
57
58if len(selected) > self.fundamental_count:
59    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
60
61self.monthly_subscribed_contracts.clear()
62self.active_stock_universe = list(map(lambda stock: stock.Symbol, selected))
63
64return self.active_stock_universe
65
66def OnData(self, slice: Slice) -> None:
67if self.selection_flag:
68    self.selection_flag = False
69
70    for symbol in self.active_stock_universe:
71        # subscribe to contract
72        contracts: List[Symbol] = self.OptionChainProvider.GetOptionContractList(symbol, self.Time)
73        underlying_price: float = self.Securities[symbol].Price
74        
75        strikes: List[float] = [i.ID.StrikePrice for i in contracts]
76        
77        if len(strikes) <= 0:
78            continue
79        
80        otm_strike: float = min(strikes, key=lambda x: abs(x - (underlying_price * (1. + self.min_delta))))
81        otm_puts: List[Symbol] = self.SelectContracts(contracts, OptionRight.Put, otm_strike)
82
83        # make sure there are enough contracts
84        if len(otm_puts) > 0:
85            # sort by expiry and subscribe nearest contract
86            nearest_contracts: Symbol = sorted(otm_puts, key=lambda item: item.ID.Date)[0]
87            
88            option: OptionContract = self.AddOptionContract(nearest_contracts, Resolution.Daily)
89            option.PriceModel = OptionPriceModels.CrankNicolsonFD()
90
91            self.monthly_subscribed_contracts.append(option)
92    
93    self.rebalance_flag = True
94
95if len(self.monthly_subscribed_contracts) != 0 and slice.OptionChains.Count != 0 and self.rebalance_flag:
96    self.rebalance_flag = False
97    
98    option_price: Dict[Symbol, float] = {
99        c.Symbol : c.AskPrice for c in self.monthly_subscribed_contracts
100    }
101
102    if len(option_price) < self.quantile:
103        return
104
105    quantile: int = int(len(option_price) / self.quantile)
106    sorted_by_opt_price: List[OptionContract] = sorted(option_price, key=option_price.get)
107    cheapest: List[OptionContract] = sorted_by_opt_price[:quantile]
108
109    # trade execution
110    for option_c in cheapest:
111        if option_c in slice and slice[option_c]:
112            self.SetHoldings(option_c, self.option_port_weight / len(cheapest))
113
114    self.SetHoldings(self.market, self.market_port_weight)
115
116def SelectContracts(self, contracts: List[Symbol], option_right: int, strike: float) -> List[Symbol]:
117return [i for i in contracts if i.ID.OptionRight == option_right \
118    and i.ID.StrikePrice == strike and self.min_expiry < (i.ID.Date - self.Time).days < self.max_expiry]
119
120def Selection(self) -> None:
121self.selection_flag = True
122
123# Custom fee model
124class CustomFeeModel(FeeModel):
125def GetOrderFee(self, parameters):
126fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
127return OrderFee(CashAmount(fee, "USD"))