Forecasted Unemployment Beta Predicts the Cross-Section of Stock Returns

1from AlgorithmImports import *
2from pandas.core.frame import DataFrame
3from typing import List, Dict
4import statsmodels.api as sm
5from dateutil.relativedelta import relativedelta
6# endregion
7
8class ForecastedUnemploymentBetaPredictstheCrossSectionofStockReturns(QCAlgorithm):
9
10def Initialize(self):
11 self.SetStartDate(2005, 1, 1) 
12 self.SetCash(100000)
13 
14 self.market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
15 self.ue:Symbol = self.AddData(UnemploymentRate, 'UE', Resolution.Daily).Symbol
16
17 self.period:int = 60
18 self.ue_period:int = 72
19 self.ue_len_threshold:int = 24
20 self.unemployment_beta:Dict[Symbol, float] = {}
21
22 self.weight:Dict[Symbol, float] = {}
23 self.quantile:int = 10 
24
25 self.leverage:int = 10
26 self.coarse_count:int = 500
27 self.selection_flag:bool = False
28 self.rebalance_flag:bool = False
29 self.UniverseSettings.Resolution = Resolution.Daily
30 self.AddUniverse(self.CoarseSelectionFunction, self.FineSelectionFunction)
31 self.Schedule.On(self.DateRules.MonthStart(self.market), self.TimeRules.AfterMarketOpen(self.market), self.Selection)
32
33def OnSecuritiesChanged(self, changes:SecurityChanges) -> None:
34 for security in changes.AddedSecurities:
35     security.SetFeeModel(CustomFeeModel())
36     security.SetLeverage(self.leverage)
37
38def CoarseSelectionFunction(self, coarse:List[CoarseFundamental]) -> List[Symbol]:
39 # monthly selection
40 if not self.selection_flag:
41     return Universe.Unchanged
42 self.selection_flag = False
43
44 selected:List[Symbol] = [x.Symbol
45     for x in sorted([x for x in coarse if x.HasFundamentalData and x.Market == 'usa' and x.AdjustedPrice > 5], 
46     key = lambda x: x.DollarVolume, reverse = True)][:self.coarse_count]
47 # selected:List[Symbol] = [x.Symbol for x in coarse if x.HasFundamentalData and x.Market == 'usa' and x.AdjustedPrice > 5]
48
49 return selected
50
51def FineSelectionFunction(self, fine:List[FineFundamental]) -> List[Symbol]:
52 fine:Dict[Symbol, FineFundamental] = {x.Symbol : x for x in fine if x.MarketCap != 0 and \
53                     (x.SecurityReference.ExchangeId == 'NYS') or (x.SecurityReference.ExchangeId == 'NAS') or (x.SecurityReference.ExchangeId == 'ASE')}
54
55 # fine:List[FineFundamental] = [x for x in fine if x.MarketCap != 0 and \
56 #                     (x.SecurityReference.ExchangeId == 'NYS') or (x.SecurityReference.ExchangeId == 'NAS') or (x.SecurityReference.ExchangeId == 'ASE')]
57 # if len(fine) > self.coarse_count:
58 #     fine:Dict[Symbol, FineFundamental] = {x.Symbol : x for x in sorted(fine, key=lambda x:x.MarketCap, reverse=True)[:self.coarse_count]}
59 # else:
60 #     fine = {x.Symbol : x for x in fine}
61
62 # call history on assets and unemployment rates
63 history:DataFrame = self.History(list(fine.keys()) + [self.market], start=self.Time.date() - relativedelta(months=self.period), end=self.Time.date())['close'].unstack(level=0)
64 history = history.groupby(pd.Grouper(freq='M')).last()
65 history = history.iloc[:-1]
66
67 ue_last_update_date:Dict[Symbol, datetime.date] = UnemploymentRate.get_last_update_date()
68 
69 # check if uemployment data are still arriving
70 if self.Securities[self.ue].GetLastData() and self.ue in ue_last_update_date and self.Time.date() <= ue_last_update_date[self.ue]:
71
72     # call history on unemployment rates
73     history_ue:DataFrame = self.History([self.ue], start=self.Time.date() - relativedelta(months=self.ue_period), end=self.Time.date())
74     beta_by_symbol:Dict[FineFundamental, float] = {}
75 
76     if len(history) >= self.period and len(history_ue) >= self.ue_len_threshold:
77         history = history.iloc[-self.period:]
78         asset_returns:DataFrame = history.pct_change().iloc[1:]
79         history_ue = history_ue.reset_index()
80
81         history_ue.set_index('time', inplace=True)
82         
83         rolling_mean_ue:DataFrame = history_ue.rolling(window=4).mean()
84         rolling_mean_ue = rolling_mean_ue.resample('M').last().ffill()
85         x_df:DataFrame = pd.concat((asset_returns[self.market], rolling_mean_ue['value']), axis=1).ffill().dropna()
86         stock_returns:DataFrame = asset_returns.loc[:, asset_returns.columns != self.market]
87         stock_returns = stock_returns.loc[stock_returns.index.isin(x_df.index)]
88
89         # run regression
90         x:np.ndarray = x_df.values
91         y:np.ndarray = stock_returns.values
92         model = self.multiple_linear_regression(x, y)
93         beta_values:np.ndarray = model.params[2]
94
95         for i, asset in enumerate(list(stock_returns.columns)):
96             asset_s:Symbol = self.Symbol(asset)
97
98             if asset_s not in beta_by_symbol:
99                 if beta_values[i] != 0 and beta_values[i] is not None:
100                     beta_by_symbol[fine[asset_s]] = beta_values[i]
101
102     # sort by beta and divide to upper decile and lower decile
103     if len(beta_by_symbol) >= self.quantile:
104         sorted_by_beta:List[FineFundamental] = sorted(beta_by_symbol, key=beta_by_symbol.get)
105         quantile:int = int(len(sorted_by_beta) / self.quantile)
106         long:List[FineFundamental] = sorted_by_beta[:quantile]
107         short:List[FineFundamental] = sorted_by_beta[-quantile:]
108
109         # calculate weights based on values
110         sum_long:float = sum([x.MarketCap for x in long])
111         for stock in long:
112             self.weight[stock.Symbol] = stock.MarketCap / sum_long
113
114         sum_short:float = sum([x.MarketCap for x in short])
115         for stock in short:
116             self.weight[stock.Symbol] = -stock.MarketCap / sum_short
117
118 return list(self.weight.keys())
119
120def OnData(self, data: Slice):
121 # monthly rebalance
122 if not self.rebalance_flag:
123     return
124 self.rebalance_flag = False
125
126 invested:List[Symbol] = [x.Key for x in self.Portfolio if x.Value.Invested]
127 for price_symbol in invested:
128     if price_symbol not in self.weight:
129         self.Liquidate(price_symbol)
130 
131 for price_symbol, weight in self.weight.items():
132     if price_symbol in data and data[price_symbol]:
133         self.SetHoldings(price_symbol, weight)
134
135 self.weight.clear()
136
137def Selection(self):
138 self.selection_flag = True
139 self.rebalance_flag = True
140
141def multiple_linear_regression(self, x:np.ndarray, y:np.ndarray):
142 # x:np.ndarray = np.array(x).T
143 x = sm.add_constant(x, prepend=True)
144 result = sm.OLS(endog=y, exog=x).fit()
145 return result
146
147# Source: https://www.philadelphiafed.org/surveys-and-data/real-time-data-research/survey-of-professional-forecasters
148class UnemploymentRate(PythonData):
149def GetSource(self, config, date, isLiveMode):
150 return SubscriptionDataSource('data.quantpedia.com/backtesting_data/economic/next_q_unemployment_forecast.csv', SubscriptionTransportMedium.RemoteFile, FileFormat.Csv)
151
152_last_update_date:Dict[Symbol, datetime.date] = {}
153
154@staticmethod
155def get_last_update_date() -> Dict[Symbol, datetime.date]:
156return UnemploymentRate._last_update_date
157
158def Reader(self, config, line, date, isLiveMode):
159 data = UnemploymentRate()
160 data.Symbol = config.Symbol
161
162 if not line[0].isdigit(): return None
163 split = line.split(';')
164 
165 # Parse the CSV file's columns into the custom data class
166 data.Time = datetime.strptime(split[0], "%Y-%m-%d") + timedelta(days=1)
167 data.Value = float(split[1])
168
169 if config.Symbol not in UnemploymentRate._last_update_date:
170     UnemploymentRate._last_update_date[config.Symbol] = datetime(1,1,1).date()
171 if data.Time.date() > UnemploymentRate._last_update_date[config.Symbol]:
172     UnemploymentRate._last_update_date[config.Symbol] = data.Time.date()
173 
174 return data
175
176# Custom fee model.
177class CustomFeeModel(FeeModel):
178def GetOrderFee(self, parameters):
179 fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
180 return OrderFee(CashAmount(fee, "USD"))