Default Risk Filter Applied on Momentum Effect within Stocks

1from AlgorithmImports import *
2import numpy as np
3import statsmodels.api as sm
4from typing import List, Dict
5#endregion
6class DefaultRiskFilterAppliedOnMomentumEffectWithinStocks(QCAlgorithm):
7def Initialize(self) -> None:
8 self.SetStartDate(2010, 1, 1)
9 self.SetCash(100000)
10 
11 self.exchange_codes:List[str] = ['NYS', 'NAS', 'ASE']
12 self.symbol: Symbol  = self.AddEquity("SPY", Resolution.Daily).Symbol
13 
14 self.data: Dict[Symbol, SymbolData] = {}
15 self.managed_queue: List[RebalanceQueueItem] = []
16 
17 self.period: int = 6 * 21    # Storing prices for 6 months
18 self.holding_months: int = 6 # Holding 1/6 of whole portfolio for six months
19 self.month_period: int = 21
20 self.leverage: int = 5
21 self.quantile: int = 10
22 self.min_share_price: int = 1
23 
24 self.regression_period: int = 21 * 3 # Three months of DEFt daily data
25 self.regression_data: RollingWindow = RollingWindow[float](self.regression_period)
26 
27 self.DEFt: Symbol = self.AddData(QuantpediaBAMLH0A3HYC, 'BAMLH0A3HYC', Resolution.Daily).Symbol
28 self.default_shock: RollingWindow = RollingWindow[float](12 * 5)   # Each month stores residual from regression for 5 years.
29 
30 self.fundamental_count: int = 500
31 self.fundamental_sorting_key = lambda x: x.DollarVolume
32 self.stop_trading: bool = False
33 self.selection_flag: bool = False
34 self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
35 self.settings.daily_precise_end_time = False
36 self.UniverseSettings.Resolution = Resolution.Daily
37 self.AddUniverse(self.FundamentalSelectionFunction)
38 self.Schedule.On(self.DateRules.MonthEnd(self.symbol), self.TimeRules.AfterMarketOpen(self.symbol), self.Selection)
39def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
40 for security in changes.AddedSecurities:
41     security.SetFeeModel(CustomFeeModel())
42     security.SetLeverage(self.leverage)
43     
44def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
45 # update the rolling window every day
46 for stock in fundamental:
47     symbol: Symbol = stock.Symbol
48     # store monthly price
49     if symbol in self.data:
50         self.data[symbol].update(stock.AdjustedPrice)
51 if not self.selection_flag:
52     return Universe.Unchanged
53 selected: List[Fundamental] = [
54     x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and x.Price > self.min_share_price and \
55     x.SecurityReference.ExchangeId in self.exchange_codes
56 ]
57 if len(selected) > self.fundamental_count:
58     selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
59 six_months_momentum = {}
60 # warmup price rolling windows
61 for stock in selected:
62     symbol: Symbol = stock.Symbol
63     if symbol not in self.data:
64         self.data[symbol] = SymbolData(self.period)
65         history = self.History(symbol, self.period, Resolution.Daily)
66         if history.empty:
67             self.Log(f"Not enough data for {symbol} yet")
68             continue
69         closes = history.loc[symbol].close
70         for time, close in closes.items():
71             self.data[symbol].update(close)
72     if not self.data[symbol].is_ready():
73         continue
74 
75     six_months_closes: List[float] = [x for x in self.data[symbol]._closes][:self.month_period] # Skip last month according to strategy
76     six_months_return: float = (six_months_closes[0] - six_months_closes[-1]) / six_months_closes[-1] 
77     
78     six_months_momentum[symbol] = six_months_return
79 
80 long: List[Symbol] = []
81 short: List[Symbol] = []
82 
83 # continue only if we have enough stocks for decile selection
84 if len(six_months_momentum) >= self.quantile:
85     quantile: int = int(len(six_months_momentum) / self.quantile)
86     sorted_by_momentum: List[Symbol] = [x[0] for x in sorted(six_months_momentum.items(), key=lambda item: item[1])]
87     
88     long = sorted_by_momentum[-quantile:] # top decile are winners
89     short = sorted_by_momentum[:quantile] # bottom decile are losers
90     
91     long_w: float = self.Portfolio.TotalPortfolioValue / self.holding_months / len(long)
92     short_w: float = self.Portfolio.TotalPortfolioValue / self.holding_months / len(short)
93     
94     # symbol/quantity collection
95     long_symbol_q: List[Tuple[Symbol, int]] = [(x, np.floor(long_w / self.data[x]._last_price)) for x in long]
96     short_symbol_q: List[Tuple[Symbol, int]] = [(x, -np.floor(short_w / self.data[x]._last_price)) for x in short]
97     
98     self.managed_queue.append(RebalanceQueueItem(long_symbol_q + short_symbol_q))
99 
100 return long + short
101def OnData(self, data: Slice) -> None:
102 if self.DEFt in data:
103     if data[self.DEFt]:
104         price: float = data[self.DEFt].Value
105         if price != 0:
106             self.regression_data.Add(price)
107 
108 # make sure DEFt data is still comming in
109 DEFt_last_update_date: Dict[Symbol, datetime.date] = QuantpediaBAMLH0A3HYC.get_last_update_date()
110 if self.Securities[self.DEFt].GetLastData() and self.Time.date() > DEFt_last_update_date[self.DEFt]:
111     self.stop_trading = True
112     if self.Portfolio.Invested:
113         self.Liquidate()
114     return
115 
116 if not self.selection_flag:
117     return
118 self.selection_flag = False
119 
120 current_residual: Union[None, float] = None
121 if self.regression_data.IsReady:
122     regression_data: List[float] = [x for x in self.regression_data]
123     Y: List[float] = regression_data[:self.month_period] # Current month data
124     X: List[float] = [
125         regression_data[self.month_period:self.month_period * 2], # Month before
126         regression_data[-self.month_period:] # Month which was 2 Months before
127     ]
128     
129     regression_model: RegressionResultWrapper = self.MultipleLinearRegression(X, Y)
130     
131     # store last residual
132     current_residual = regression_model.resid[-1]
133     self.default_shock.Add(current_residual)
134 
135 invest_flag: bool = False
136 
137 if self.default_shock.IsReady and current_residual:
138     residuals_median: float = np.median([x for x in self.default_shock])
139     
140     parts_to_remove: List[RebalanceQueueItem] = []
141     
142     # go through each part of portfolio 
143     for portfolio_part in self.managed_queue:
144         # liquidate stocks with 6 months holding
145         if portfolio_part.holding_time == 6:
146             for symbol, quantity in portfolio_part.symbol_q:
147                 if self.Portfolio[symbol].Invested:
148                     # liquidate long and short stocks in this part of portfolio
149                     self.MarketOrder(symbol, -quantity)
150             
151             parts_to_remove.append(portfolio_part)
152         
153         # Increment holding time of 1/6 portfolio    
154         portfolio_part.holding_time += 1
155     
156     for part_to_remove in parts_to_remove:
157         # Remove portfolio part, because we held it for 6 months
158         self.managed_queue.remove(part_to_remove)
159 
160     # Check if we invest or liquidate
161     if current_residual > residuals_median:
162         invest_flag = True
163 
164 # Default shock data aren't ready, so we have to remove portofolio part, which was added.   
165 elif len(self.managed_queue) > 0:
166     self.managed_queue.pop()
167 
168 # Trade execution
169 if invest_flag:
170     # We either hold portfolio and invest into the new part of it
171     # Or invest into whole portfolio, because we liquidated it
172     
173     if self.Portfolio.Invested:
174         # Get portfolio part, which was lastly added
175         portfolio_part = self.managed_queue[-1]
176         # We keep holding old parts of portfolio and we invest into the new one
177         open_symbol_q = []
178         for symbol, quantity in portfolio_part.symbol_q:
179             if symbol in data and data[symbol]:
180                 self.MarketOrder(symbol, quantity)
181                 open_symbol_q.append((symbol, quantity))
182         portfolio_part.symbol_q = open_symbol_q
183     else:
184         # We need to invest into whole portfolio, because we liquidated it
185         for portfolio_part in self.managed_queue:
186             # Go long and short stocks in this part of portofolio
187             open_symbol_q = []
188             for symbol, quantity in portfolio_part.symbol_q:
189                 if symbol in data and data[symbol]:
190                     self.MarketOrder(symbol, quantity)
191                     open_symbol_q.append((symbol, quantity))
192             portfolio_part.symbol_q = open_symbol_q
193                     
194 else: # Signal is to Liquidate whole portfolio
195     self.Liquidate()
196
197def MultipleLinearRegression(self, x, y):
198 x = np.array(x).T
199 x = sm.add_constant(x)
200 result: RegressionResultWrapper = sm.OLS(endog=y, exog=x).fit()
201 return result
202
203def Selection(self) -> None:
204 if not self.stop_trading:
205     self.selection_flag = True
206 
207class RebalanceQueueItem():
208def __init__(self, symbol_q: List[Symbol]) -> None:
209 self.symbol_q: List[Symbol] = symbol_q
210 self.holding_time: int = 0 # Holding in months
211 
212class SymbolData():
213def __init__(self, period: int) -> None:
214 self._last_price: Union[None, float] = None
215 self._closes: RollingWindow = RollingWindow[float](period)
216
217def update(self, close: float) -> None:
218 self._closes.Add(close)
219 self._last_price = close
220 
221def is_ready(self) -> bool:
222 return self._closes.IsReady
223# Custom fee model
224class CustomFeeModel(FeeModel):
225def GetOrderFee(self, parameters: OrderFeeParameters) -> OrderFee:
226 fee: float = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
227 return OrderFee(CashAmount(fee, "USD"))
228 
229# Quantpedia data.
230# NOTE: IMPORTANT: Data order must be ascending (datewise)
231class QuantpediaBAMLH0A3HYC(PythonData):
232_last_update_date:Dict[Symbol, datetime.date] = {}
233@staticmethod
234def get_last_update_date() -> Dict[Symbol, datetime.date]:
235return QuantpediaBAMLH0A3HYC._last_update_date
236def GetSource(self, config: SubscriptionDataConfig, date: datetime, isLiveMode: bool) -> SubscriptionDataSource:
237 return SubscriptionDataSource("data.quantpedia.com/backtesting_data/index/{0}.csv".format(config.Symbol.Value), SubscriptionTransportMedium.RemoteFile, FileFormat.Csv)
238def Reader(self, config: SubscriptionDataConfig, line: str, date: datetime, isLiveMode: bool) -> BaseData:
239 data = QuantpediaBAMLH0A3HYC()
240 data.Symbol = config.Symbol
241 
242 if not line[0].isdigit(): return None
243 split: str = line.split(';')
244 
245 if split[1] != '.':
246     data.Time = datetime.strptime(split[0], "%Y-%m-%d") + timedelta(days=1)
247     value = float(split[1])
248     data.Value = value
249     if config.Symbol not in QuantpediaBAMLH0A3HYC._last_update_date:
250         QuantpediaBAMLH0A3HYC._last_update_date[config.Symbol] = datetime(1,1,1).date()
251     if data.Time.date() > QuantpediaBAMLH0A3HYC._last_update_date[config.Symbol]:
252         QuantpediaBAMLH0A3HYC._last_update_date[config.Symbol] = data.Time.date()
253 return data