Market Timing with Aggregate and Idiosyncratic Stock Volatilities

1import numpy as np
2from AlgorithmImports import *
3import statsmodels.api as sm
4import numpy as np
5from typing import Dict, List, Tuple
6from pandas.core.frame import DataFrame
7from pandas.core.series import Series
8class MarketTimingwithAggregateandIdiosyncraticStockVolatilities(QCAlgorithm):
9def Initialize(self):
10self.SetStartDate(2000, 1, 1)
11self.SetCash(100000)
12self.data:Dict[Symbol, SymbolData] = {}
13self.weight:Dict[Symbol, float] = {}
14
15self.period:int = 3*21
16self.fundamental_count:int = 500
17self.leverage:int = 5
18self.quantile:int = 10
19self.exchange_codes:List[str] = ['NYS', 'NAS', 'ASE']
20
21self.last_size_long:List[Symbol] = []
22self.last_size_short:List[Symbol] = []
23self.last_book_long:List[Symbol] = []
24self.last_book_short:List[Symbol] = []
25
26self.market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
27self.data[self.market] = SymbolData(self.period)
28
29self.bonds:Symbol = self.AddEquity('SHY', Resolution.Daily).Symbol
30
31self.regression_data:List[Tuple[float, float, float]] = []
32self.regression_data_min_period:int = 12
33
34self.selection_flag:bool = False
35self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
36self.UniverseSettings.Resolution = Resolution.Daily
37self.AddUniverse(self.FundamentalSelectionFunction)
38
39self.Schedule.On(self.DateRules.MonthEnd(self.market), self.TimeRules.AfterMarketOpen(self.market), self.Selection)
40
41self.settings.daily_precise_end_time = False
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]:
49# Update the rolling window every day.
50for stock in fundamental:
51    symbol:Symbol = stock.Symbol
52    # Store monthly price.
53    if symbol in self.data:
54        self.data[symbol].update(stock.AdjustedPrice)
55if not self.selection_flag:
56    return Universe.Unchanged
57
58selected: List[Fundamental] = sorted([x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and x.MarketCap != 0 \
59                    and not np.isnan(x.ValuationRatios.PBRatio) and x.ValuationRatios.PBRatio != 0
60                    and x.SecurityReference.ExchangeId in self.exchange_codes],
61                    key=lambda x: x.DollarVolume, reverse=True)[:self.fundamental_count]
62    
63selected_dict: Dict[Symbol, Fundamental] = {x.Symbol : x for x in selected}
64market_cap:Dict[Symbol, float] = {}
65book_to_market:Dict[Symbol, float] = {}
66# Warmup price rolling windows.
67for symbol in list(selected_dict.keys()) + [self.market]:
68    if symbol in self.data:
69        continue
70    
71    self.data[symbol] = SymbolData(self.period)
72    history:DataFrame = self.History(symbol, self.period, Resolution.Daily)
73    if history.empty:
74        self.Log(f"Not enough data for {symbol} yet")
75        continue
76    closes:Series = history.loc[symbol].close
77    for time, close in closes.items():
78        self.data[symbol].update(close)
79    if self.data[symbol].is_ready():
80        market_cap[symbol] = selected_dict[symbol].MarketCap
81        book_to_market[symbol] = 1 / selected_dict[symbol].ValuationRatios.PBRatio # book-to-market
82        
83# Return, if after fundamental filtration we don't have enough stocks to sort them into deciles.
84# In next month we will use size and book factors from t-3 month, where t is current month.
85if len(market_cap) < self.quantile:
86    return Universe.Unchanged
87quantile:int = int(len(market_cap) / self.quantile)
88sorted_by_market_cap:List[Symbol] = [x[0] for x in sorted(market_cap.items(), key=lambda item: item[1])]
89sorted_book_to_market:List[Symbol] = [x[0] for x in sorted(book_to_market.items(), key=lambda item: item[1])]
90
91# Size Factor
92current_size_long:List[Symbol] = sorted_by_market_cap[:quantile]
93current_size_short:List[Symbol] = sorted_by_market_cap[-quantile:]
94
95# Book to market Factor
96current_book_long:List[Symbol] = sorted_book_to_market[:quantile]
97current_book_short:List[Symbol] = sorted_book_to_market[-quantile:]
98
99# Check if factors are ready
100if (len(self.last_size_long) > 0 and len(self.last_size_short) > 0 and
101    len(self.last_book_long) > 0 and len(self.last_book_short) > 0):
102    # Check if market prices are ready
103    if self.data[self.market].is_ready(): 
104        # Calculate factors daily returns
105        book_factor_daily_returns:List[float] = self.CalculateFactorDailyReturns(self.last_book_long, self.last_book_short)
106        size_factor_daily_returns:List[float] = self.CalculateFactorDailyReturns(self.last_size_long, self.last_size_short)
107        
108        market_returns:np.ndarray = self.data[self.market].daily_performances()
109        market_variance:np.ndarray = (np.std(market_returns) * np.sqrt(252)) ** 2
110        market_return:float = self.data[self.market].performance()
111        
112        # stock residuals
113        idiosyncratic_variance:List[Tuple[float, float]] = []
114        for stock in selected:
115            symbol: Symbol = stock.Symbol
116            if symbol not in self.data:
117                continue
118            
119            if not self.data[symbol].is_ready():
120                continue
121            Y:np.ndarray = self.data[symbol].daily_performances()
122            X:np.ndarray = [
123                market_returns,
124                size_factor_daily_returns,
125                book_factor_daily_returns
126            ]
127                
128            regression_model = self.MultipleLinearRegression(X, Y)
129            
130            idiosyncratic_variance.append((regression_model.resid[-1], selected_dict[symbol].MarketCap))
131        
132        # summary idiosyncratic variance calculation
133        summary_market_cap:float = sum([x[1] for x in idiosyncratic_variance])
134        summary_idiosyncratic_variance:float = sum([x[0] * (x[1] / summary_market_cap) for x in idiosyncratic_variance])
135        
136        self.regression_data.append((market_return, market_variance, summary_idiosyncratic_variance))
137        
138self.last_size_long = current_size_long
139self.last_size_short = current_size_short
140self.last_book_long = current_book_long
141self.last_book_short = current_book_short
142            
143return Universe.Unchanged
144def OnData(self, data:Slice) -> None:
145if not self.selection_flag:
146    return
147self.selection_flag = False
148
149if self.bonds not in data or not data[self.bonds]:
150    return
151# regression data is ready
152if len(self.regression_data) >= self.regression_data_min_period:
153    Y = [x[0] for x in self.regression_data][:-1]
154    X = [
155        [x[1] for x in self.regression_data][1:],
156        [x[2] for x in self.regression_data][1:]
157    ]
158    
159    # expected market return calculation
160    regression_model = self.MultipleLinearRegression(X, Y)
161    alpha:float = regression_model.params[0]
162    beta1:float = regression_model.params[1]
163    beta2:float = regression_model.params[2]
164    
165    expected_market_return:float = alpha + self.regression_data[0][1]*beta1 + self.regression_data[0][2]*beta2
166    expected_market_variance:float = np.mean([x[1] for x in self.regression_data])
167    risk_aversion:int = 5
168    
169    # equity and tbill weight
170    equity_weight:float = expected_market_return / (risk_aversion * expected_market_variance)
171    tbill_weight:float = 1 - equity_weight
172    
173    # trade execution
174    if self.market in data and data[self.market] and self.bonds in data and data[self.bonds]:
175        self.SetHoldings(self.market, equity_weight)
176        self.SetHoldings(self.bonds, tbill_weight)
177    
178def CalculateFactorDailyReturns(self, factor_long:List[Symbol], factor_short:List[Symbol]) -> List[float]:
179long_daily_returns:List[Symbol] = self.ListOfDailyReturns(factor_long, True)
180short_daily_returns:List[Symbol] = self.ListOfDailyReturns(factor_short, False)
181
182stocks_daily_returns:np.ndarray = np.array(long_daily_returns + short_daily_returns)
183
184factor_daily_returns:List[float] = []
185
186for i in range(len(stocks_daily_returns[0])):
187    factor_daily_returns.append(np.mean(stocks_daily_returns[:, i])) # Takes column of 2d array
188    
189return factor_daily_returns
190
191def ListOfDailyReturns(self, symbols:List[Symbol], long_flag:bool) -> List[Symbol]:
192symbol_list = []
193
194for symbol in symbols:
195    # Those aren't symbols which were return from coarse
196    if symbol in self.data:
197        if self.data[symbol].is_ready():
198            if long_flag:
199                symbol_list.append(self.data[symbol].daily_performances())
200            else:
201                symbol_list.append(self.data[symbol].short_daily_performances())
202return symbol_list
203 
204def MultipleLinearRegression(self, x, y):
205x:np.ndarray = np.array(x).T
206x = sm.add_constant(x)
207result = sm.OLS(endog=y, exog=x).fit()
208return result     
209
210def Selection(self):
211if self.Time.month % 3 == 0:
212    self.selection_flag = True
213class SymbolData():
214def __init__(self, period:int) -> None:
215self.closes:RollingWindow[float] = RollingWindow[float](period)
216self.period:int = period
217
218def update(self, close:float) -> None:
219self.closes.Add(close)
220
221def is_ready(self) -> bool:
222return self.closes.IsReady
223
224def performance(self) -> float:
225return self.closes[0] / self.closes[self.period - 1] - 1
226
227def daily_performances(self) -> np.ndarray:
228closes = np.array([x for x in self.closes])
229return (closes[:-1] - closes[1:]) / closes[1:]
230
231def short_daily_performances(self) -> np.ndarray:
232closes = np.array([x for x in self.closes])
233return ( -(closes[:-1] - closes[1:]) ) / closes[1:] # We change mark of daily performance for short stocks.
234
235# Custom fee model
236class CustomFeeModel(FeeModel):
237def GetOrderFee(self, parameters):
238fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
239return OrderFee(CashAmount(fee, "USD"))