Option Trading and Returns versus the 52-Week Low

1from AlgorithmImports import *
2import data_tools
3# endregion
4
5class OptionTradingandReturnsversusthe52WeekLow(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_ptl:List[Symbol] = []
22self.lowest_ptl: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 = {}      # 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
58if self.coarse_count <= 1000:
59    selected:List[CoarseFundamental] = sorted([x for x in coarse if x.HasFundamentalData and x.Market == 'usa' and x.Price > 5],
60        key=lambda x: x.DollarVolume, reverse=True)[:self.coarse_count]
61else:
62    selected:List[CoarseFundamental] = [x for x in coarse if x.HasFundamentalData and x.Market == 'usa']
63
64self.selected_symbols = list(map(lambda stock: stock.Symbol, selected))
65
66return self.selected_symbols
67
68def OnData(self, data:Slice) -> None:
69if self.highest_ptl and self.lowest_ptl:
70    delta_by_symbol:dict[Symobl, float] = {}
71    oi_by_symbol:dict[Symobl, float] = {}
72
73    # 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.
74    if self.subscribed_contracts:
75        # store OI values
76        if data.OptionChains.Count != 0:
77            for kvp in data.OptionChains:
78                contract = list(kvp.Value)[0]
79                symbol:Symbol = contract.UnderlyingSymbol
80                open_interest:float = contract.OpenInterest
81
82                if open_interest != 0:
83                    oi_by_symbol[symbol] = open_interest
84
85        # trade options and hedge
86        total_oi:float = sum([oi_by_symbol[x] for x in self.highest_ptl if x in oi_by_symbol])
87        for symbol in self.highest_ptl:
88            if symbol in data and data[symbol]:
89                if symbol in self.subscribed_contracts:
90                    if symbol in oi_by_symbol:
91                        price:float = data[symbol].Value
92                        weight:float = oi_by_symbol[symbol] / total_oi
93                        
94                        equity:float = self.Portfolio.TotalPortfolioValue * weight
95                        options_q:int = int(equity / (price * 100))
96
97                        self.Buy(self.subscribed_contracts[symbol], options_q)
98
99        total_oi:float = sum([oi_by_symbol[x] for x in self.lowest_ptl if x in oi_by_symbol])
100        for symbol in self.lowest_ptl:
101            if symbol in data and data[symbol]:
102                if symbol in self.subscribed_contracts:
103                    if symbol in oi_by_symbol:
104                        price:float = data[symbol].Value
105                        weight:float = oi_by_symbol[symbol] / total_oi
106                        
107                        equity:float = self.Portfolio.TotalPortfolioValue * weight
108                        options_q:int = int(equity / (price * 100))
109
110                        self.Sell(self.subscribed_contracts[symbol], options_q)
111
112    self.subscribed_contracts.clear()
113    self.highest_ptl.clear()
114    self.lowest_ptl.clear()
115
116# store daily prices
117for symbol in self.selected_symbols:
118    if symbol in self.data:
119        if symbol in data and data[symbol] and data[symbol].High != 0 and data[symbol].Value != 0:
120            price:float = data[symbol].Value
121            low:float = data[symbol].Low
122
123            self.data[symbol].update(price, low)
124
125# rebalance monthly
126if not self.selection_flag:
127    return
128self.selection_flag = False
129
130# calculate ptl
131ptl:dict[Symbol, float] = { symbol : self.data[symbol].PTL() for symbol in self.selected_symbols if symbol in self.data and self.data[symbol].PTL_data_ready() }
132
133if len(ptl) < self.quantile:
134    self.Liquidate()
135    return
136
137# sorting
138quantile:int = int(len(ptl) / self.quantile)
139sorted_by_ptl:List[Symbol] = [x[0] for x in sorted(ptl.items(), key=lambda item: item[1])]
140
141self.highest_ptl = sorted_by_ptl[-quantile:]
142self.lowest_ptl = sorted_by_ptl[:quantile]
143
144for symbol in self.highest_ptl + self.lowest_ptl:
145    if symbol in data and data[symbol]:
146        # get all contracts for current stock symbol
147        contracts:List = self.OptionChainProvider.GetOptionContractList(symbol, self.Time)
148        # get current price for stock
149        underlying_price:float = data[symbol].Value
150        
151        # get strikes from commodity future contracts
152        strikes:List[float] = [c.ID.StrikePrice for c in contracts]
153        
154        # can't filter contracts, if there isn't any strike price
155        if len(strikes) <= 0 or underlying_price == 0:
156            continue
157        
158        atm_strike:float = min(strikes, key=lambda x: abs(x-underlying_price))
159        
160        # filter calls contracts with one month expiry
161        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 ]
162
163        # make sure, there is at least one call contract
164        if len(atm_calls) > 0:
165            # sort by expiry
166            atm_call = sorted(atm_calls, key = lambda x: x.ID.Date, reverse=True)[0]
167            
168            atm_call_subscriptions = self.SubscriptionManager.SubscriptionDataConfigService.GetSubscriptionDataConfigs(atm_call.Underlying)
169            
170            # check if stock's call contract was successfully subscribed
171            if atm_call_subscriptions:
172                # add contract
173                option = self.AddOptionContract(atm_call, Resolution.Daily)
174                option.PriceModel = OptionPriceModels.CrankNicolsonFD()
175
176                # store contracts by stock's symbol
177                self.subscribed_contracts[symbol] = atm_call
178
179def Selection(self) -> None:
180self.selection_flag = True
181self.Liquidate()