Option Trading and Returns versus the 52-Week High

1from AlgorithmImports import *
2import data_tools
3# endregion
4
5class OptionTradingandReturnsversusthe52WeekHigh(QCAlgorithm):
6
7def Initialize(self):
8self.SetStartDate(2010, 1, 1)
9self.SetCash(100000)
10
11self.leverage:int = 5
12self.quantile:int = 5
13self.portfolio_percentage:float = 1
14self.high_period:int = 52 * 5
15
16self.exchanges:list[str] = ['NYS', 'NAS', 'ASE']
17
18self.data:dict[Symbol, data_tools.SymbolData] = {}
19
20self.selected_symbols:list[Symbol] = []
21self.highest_PTH:List[Symbol] = []
22self.lowest_PTH:List[Symbol] = []
23
24self.market_symbol:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
25
26# options storage
27self.min_expiry:int = 30
28self.max_expiry:int = 60
29self.subscribed_contracts:dict = {}      # subscribed option universe
30
31self.coarse_count:int = 500
32self.selection_flag:bool = False
33self.UniverseSettings.Resolution = Resolution.Daily
34self.AddUniverse(self.CoarseSelectionFunction)
35self.SetSecurityInitializer(lambda x: x.SetDataNormalizationMode(DataNormalizationMode.Raw))
36self.UniverseSettings.DataNormalizationMode = DataNormalizationMode.Raw
37
38self.Schedule.On(self.DateRules.MonthStart(self.market_symbol), self.TimeRules.BeforeMarketClose(self.market_symbol, 0), self.Selection)
39
40def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
41for security in changes.AddedSecurities:
42    security.SetFeeModel(data_tools.CustomFeeModel())
43    security.SetLeverage(self.leverage)
44    
45    symbol:Symbol = security.Symbol
46    if symbol not in self.data:
47        self.data[symbol] = data_tools.SymbolData(self.high_period)
48
49for security in changes.RemovedSecurities:
50    symbol:Symbol = security.Symbol
51    if symbol in self.data:
52        del self.data[symbol]
53
54def CoarseSelectionFunction(self, coarse:List[CoarseFundamental]) -> List[Symbol]:
55if not self.selection_flag:
56    return Universe.Unchanged
57
58selected:list = sorted([x for x in coarse if x.HasFundamentalData and x.Market == 'usa' and x.Price > 5],
59    key=lambda x: x.DollarVolume, reverse=True)[:self.coarse_count]
60
61self.selected_symbols = list(map(lambda stock: stock.Symbol, selected))
62
63return self.selected_symbols
64
65def OnData(self, data:Slice) -> None:
66if self.highest_PTH and self.lowest_PTH:
67    oi_by_symbol:dict[Symobl, float] = {}
68
69    # 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.
70    if self.subscribed_contracts:
71        # store OI values
72        if data.OptionChains.Count != 0:
73            for kvp in data.OptionChains:
74                contract = list(kvp.Value)[0]
75                symbol:Symbol = contract.UnderlyingSymbol
76                open_interest:float = contract.OpenInterest
77
78                if open_interest != 0:
79                    oi_by_symbol[symbol] = open_interest
80
81        # trade options and hedge
82        total_oi:float = sum([oi_by_symbol[x] for x in self.highest_PTH if x in oi_by_symbol])
83        for symbol in self.highest_PTH:
84            if symbol in data and data[symbol]:
85                if symbol in self.subscribed_contracts:
86                    if symbol in oi_by_symbol:
87                        price:float = data[symbol].Value
88                        weight:float = oi_by_symbol[symbol] / total_oi
89                        
90                        equity:float = self.Portfolio.TotalPortfolioValue * weight
91                        options_q:int = int(equity / (price * 100))
92
93                        self.Buy(self.subscribed_contracts[symbol], options_q)
94                        
95        total_oi:float = sum([oi_by_symbol[x] for x in self.lowest_PTH if x in oi_by_symbol])
96        for symbol in self.lowest_PTH:
97            if symbol in data and data[symbol]:
98                if symbol in self.subscribed_contracts:
99                    if symbol in oi_by_symbol:
100                        price:float = data[symbol].Value
101                        weight:float = oi_by_symbol[symbol] / total_oi
102                        
103                        equity:float = self.Portfolio.TotalPortfolioValue * weight
104                        options_q:int = int(equity / (price * 100))
105
106                        self.Sell(self.subscribed_contracts[symbol], options_q)
107                        
108    self.subscribed_contracts.clear()
109    self.highest_PTH.clear()
110    self.lowest_PTH.clear()
111
112# store daily prices
113for symbol in self.selected_symbols:
114    if symbol in self.data:
115        if symbol in data and data[symbol] and data[symbol].High != 0 and data[symbol].Value != 0:
116            price:float = data[symbol].Value
117            high:float = data[symbol].High
118
119            self.data[symbol].update(price, high)
120
121# rebalance monthly
122if not self.selection_flag:
123    return
124self.selection_flag = False
125
126# calculate PTH
127PTH:dict[Symbol, float] = { symbol : self.data[symbol].PTH() for symbol in self.selected_symbols if symbol in self.data and self.data[symbol].PTH_data_ready() }
128
129if len(PTH) < self.quantile:
130    self.Liquidate()
131    return
132
133# sorting
134quantile:int = int(len(PTH) / self.quantile)
135sorted_by_PTH:List[Symbol] = [x[0] for x in sorted(PTH.items(), key=lambda item: item[1])]
136
137self.highest_PTH = sorted_by_PTH[-quantile:]
138self.lowest_PTH = sorted_by_PTH[:quantile]
139
140for symbol in self.highest_PTH + self.lowest_PTH:
141    if symbol in data and data[symbol]:
142        # get all contracts for current stock symbol
143        contracts:List = self.OptionChainProvider.GetOptionContractList(symbol, self.Time)
144        # get current price for stock
145        underlying_price:float = data[symbol].Value
146        
147        # get strikes from commodity future contracts
148        strikes:List[float] = [c.ID.StrikePrice for c in contracts]
149        
150        # can't filter contracts, if there isn't any strike price
151        if len(strikes) <= 0 or underlying_price == 0:
152            continue
153        
154        atm_strike:float = min(strikes, key=lambda x: abs(x-underlying_price))
155        
156        # filter calls contracts with one month expiry
157        atm_calls:List = [ contract for contract in contracts if self.min_expiry < (contract.ID.Date - self.Time).days < self.max_expiry and contract.ID.OptionRight == OptionRight.Call and contract.ID.StrikePrice == atm_strike ]
158
159        # make sure, there is at least one call contract
160        if len(atm_calls) > 0:
161            # sort by expiry
162            atm_call = sorted(atm_calls, key = lambda x: x.ID.Date, reverse=True)[0]
163            
164            atm_call_subscriptions = self.SubscriptionManager.SubscriptionDataConfigService.GetSubscriptionDataConfigs(atm_call.Underlying)
165            
166            # check if stock's call contract was successfully subscribed
167            if atm_call_subscriptions:
168                # add contract
169                option = self.AddOptionContract(atm_call, Resolution.Daily)
170                option.PriceModel = OptionPriceModels.CrankNicolsonFD()
171
172                # store contracts by stock's symbol
173                self.subscribed_contracts[symbol] = atm_call
174
175def Selection(self) -> None:
176self.selection_flag = True
177self.Liquidate()