Cross Industry Dispersion Factor

1from AlgorithmImports import *
2from typing import List, Dict
3from collections import deque
4from pandas.core.frame import DataFrame
5import statsmodels.api as sm
6from dateutil.relativedelta import relativedelta
7# endregion
8
9class CrossIndustryDispersionFactor(QCAlgorithm):
10
11def Initialize(self):
12 self.SetStartDate(2000, 1, 1)
13 self.SetCash(100000)
14
15 self.stock_price_data:Dict[Symbol, deque] = {}
16 self.ff_price_data:Dict[Symbol, deque] = {}
17 
18 self.leverage:int = 3
19 self.month_period:int = 2 * 12
20 self.period:int = self.month_period * 30
21 self.quantile:int = 5  
22
23 self.SetWarmup(self.period, Resolution.Daily)
24 
25 self.market:Symbol = self.AddEquity("SPY", Resolution.Daily).Symbol
26 self.ff_industry:Symbol = self.AddData(QuantpediaFamaFrench, "fama_french_49_industry_VW", Resolution.Daily).Symbol
27
28 self.weight:Dict[Symbol, float] = {}
29
30 self.coarse_count:int = 1000
31 self.selection_flag:bool = False
32 self.exchanges:List[str] = ['NYS', 'NAS', 'ASE']
33 self.UniverseSettings.Resolution = Resolution.Daily
34 self.AddUniverse(self.CoarseSelectionFunction, self.FineSelectionFunction)
35 self.Schedule.On(self.DateRules.MonthStart(self.market), self.TimeRules.AfterMarketOpen(self.market), self.Selection)
36
37def CoarseSelectionFunction(self, coarse:List[CoarseFundamental]) -> List[Symbol]:
38 # update price every day
39 for stock in coarse:
40     symbol = stock.Symbol
41     if symbol in self.stock_price_data:
42         self.stock_price_data[symbol].append((self.Time, stock.AdjustedPrice))
43
44 if not self.selection_flag:
45     return Universe.Unchanged
46
47 if self.coarse_count < 3000:
48     selected:list = sorted([x for x in coarse if x.HasFundamentalData and x.AdjustedPrice >= 1],
49             key=lambda x: x.DollarVolume, reverse=True)[:self.coarse_count]
50 else:
51     selected:list = [x for x in coarse if x.HasFundamentalData and x.AdjustedPrice >= 1]
52 
53 # warmup prices
54 for stock in selected:
55     symbol:Symbol = stock.Symbol
56     if symbol in self.stock_price_data:
57         continue
58     
59     self.stock_price_data[symbol] = deque(maxlen=self.period)
60     history = self.History(symbol, self.period, Resolution.Daily)
61     if history.empty:
62         self.Log(f"Not enough data for {symbol} yet.")
63         continue
64     closes = history.loc[symbol].close
65     for time, close in closes.iteritems():
66         self.stock_price_data[symbol].append((time, close))
67     
68 return [x.Symbol for x in selected if len(self.stock_price_data[x.Symbol]) == self.stock_price_data[x.Symbol].maxlen]
69
70def FineSelectionFunction(self, fine:List[FineFundamental]) -> List[Symbol]:
71 # filter fine
72 fine:Dict[Symbol, FineFundamental] = {x.Symbol : x for x in fine if x.MarketCap != 0 and x.SecurityReference.ExchangeId in self.exchanges }
73
74 if len(fine) >= self.coarse_count:
75     fine = {x[0] : x[1] for x in sorted(fine.items(), key = lambda x: x[1].MarketCap, reverse=True)[:self.coarse_count]}
76
77 # create FF df out of deques
78 assets:Dict = {symbol : [i[1] for i in deq] for symbol, deq in self.ff_price_data.items()}
79 ff_df:DataFrame = pd.DataFrame(assets, index=[i[0] for i in list(self.ff_price_data.values())[0]])
80 
81 # daily FF returns to daily equity
82 ff_df[QuantpediaFamaFrench._columns] = (1 + ff_df[QuantpediaFamaFrench._columns]).cumprod()
83 
84 # calculate monthly return
85 ff_df = ff_df.groupby(pd.Grouper(freq='M')).last()
86 ff_df = ff_df.pct_change().iloc[-self.month_period-1:-1]
87 CID:DataFrame = abs(ff_df[QuantpediaFamaFrench._columns].sub(ff_df[self.market], axis=0)).mean(axis=1)
88
89 # create stock df out of deques
90 assets:Dict = {symbol : [i[1] for i in self.stock_price_data[symbol]] for symbol, _  in fine.items()}
91 stock_df:DataFrame = pd.DataFrame(assets, index=[i[0] for i in list(self.stock_price_data.values())[0]])
92 
93 # calculate monthly stock returns
94 stock_df = stock_df.groupby(pd.Grouper(freq='M')).last()
95 stock_df = stock_df.pct_change().iloc[-self.month_period-1:-1]
96
97 # run regression
98 x:np.ndarray = CID.values
99 y:np.ndarray = stock_df.values
100 model = self.multiple_linear_regression(x, y)
101 beta_values:np.ndarray = model.params[1]
102 asset_cols:List[str] = list(stock_df.columns)
103
104 # store beta by symbol
105 beta_by_symbol:Dict[Symbol, float] = {}
106 for i, beta in enumerate(beta_values):
107     beta_by_symbol[fine[asset_cols[i]]] = beta
108
109 if len(beta_by_symbol) >= self.quantile:
110     # sort by beta
111     sorted_beta:List[Tuple] = sorted(beta_by_symbol, key=beta_by_symbol.get, reverse=True)
112     quantile:int = len(sorted_beta) // self.quantile
113     long:List[FineFundamental] = sorted_beta[-quantile:]
114     short:List[FineFundamental] = sorted_beta[:quantile]
115
116     total_market_cap_long:float = sum([x.MarketCap for x in long])
117     total_market_cap_short:float = sum([x.MarketCap for x in short])
118     
119     for stock in long:
120         self.weight[stock.Symbol] = stock.MarketCap / total_market_cap_long
121
122     for stock in short:
123         self.weight[stock.Symbol] = -stock.MarketCap / total_market_cap_short
124
125 return list(self.weight.keys())
126
127def OnSecuritiesChanged(self, changes:SecurityChanges) -> None:
128 for security in changes.AddedSecurities:
129     security.SetFeeModel(CustomFeeModel())
130     security.SetLeverage(self.leverage)
131
132def OnData(self, data: Slice) -> None:
133 # store FF industry data and market data
134 if data.ContainsKey(self.ff_industry) and data[self.ff_industry]:
135     if data.ContainsKey(self.market) and data[self.market]:
136         # init deques
137         for symbol in [self.market] + QuantpediaFamaFrench._columns:
138             if symbol not in self.ff_price_data:
139                 self.ff_price_data[symbol] = deque(maxlen=self.period)
140
141         # FF daily returns
142         for col in QuantpediaFamaFrench._columns:
143             self.ff_price_data[col].append((self.Time, data[self.ff_industry].GetProperty(col) / 100.))
144         
145         # market daily prices
146         self.ff_price_data[self.market].append((self.Time, data[self.market].Value))
147
148 if not self.selection_flag:
149     return
150 self.selection_flag = False
151
152 # liquidate
153 stocks_invested:List[Symbol] = [x.Key for x in self.Portfolio if x.Value.Invested]
154 for symbol in stocks_invested:
155     if symbol not in self.weight:
156         self.Liquidate(symbol)
157
158 # trade execution
159 for symbol, w in self.weight.items():
160     if symbol in data and data[symbol]:
161         self.SetHoldings(symbol, w)
162
163 self.weight.clear()
164
165def Selection(self) -> None:
166 if self.ff_price_data:
167     if all(len(self.ff_price_data[col]) == self.period for col in QuantpediaFamaFrench._columns) and \
168         len(self.ff_price_data[self.market]) == self.period:
169         self.selection_flag = True
170
171def multiple_linear_regression(self, x:np.ndarray, y:np.ndarray):
172 # x = np.array(x).T
173 x = sm.add_constant(x)
174 result = sm.OLS(endog=y, exog=x).fit()
175 return result
176 
177# Quantpedia data
178# NOTE: IMPORTANT: Data order must be ascending (datewise)
179class QuantpediaFamaFrench(PythonData):
180def GetSource(self, config, date, isLiveMode):
181 return SubscriptionDataSource("data.quantpedia.com/backtesting_data/equity/fama_french/fama_french_49_industry_VW.csv", SubscriptionTransportMedium.RemoteFile, FileFormat.Csv)
182
183_last_update_date:datetime.date = datetime(1,1,1).date()
184_columns:List[str] = list()
185
186@staticmethod
187def get_last_update_date() -> Dict[Symbol, datetime.date]:
188return QuantpediaFamaFrench._last_update_date
189
190def Reader(self, config, line, date, isLiveMode):
191 data = QuantpediaFamaFrench()
192 data.Symbol = config.Symbol
193 
194 if not line[0].isdigit():
195     QuantpediaFamaFrench._columns = line.split(',')[1:] # skip 'Date' columns
196     return None
197 
198 split = line.split(',')
199 
200 data.Time = datetime.strptime(split[0], "%Y-%m-%d") + relativedelta(months=1)
201 if data.Time.date() > QuantpediaFamaFrench._last_update_date:
202     QuantpediaFamaFrench._last_update_date = data.Time.date()
203
204 for i, col_name in enumerate(QuantpediaFamaFrench._columns):
205     data[col_name] = float(split[i+1]) 
206
207 return data
208
209# custom fee model
210class CustomFeeModel(FeeModel):
211def GetOrderFee(self, parameters):
212 fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
213 return OrderFee(CashAmount(fee, "USD"))