The Effect of Market Returns on Factor Returns

1from AlgorithmImports import *
2import data_tools
3from typing import List, Dict
4import numpy as np
5import pandas as pd
6# endregion
7
8class TheEffectofMarketReturnsonFactorReturns(QCAlgorithm):
9
10def Initialize(self):
11self.SetStartDate(2000, 1, 1)
12self.SetCash(100000)
13
14self.market:Symbol = self.AddEquity("SPY", Resolution.Daily).Symbol
15 
16self.fundamental_count:int = 3000
17self.leverage:int = 5
18self.quantile:int = 5
19self.daily_period:int = 252
20self.monthly_period:int = 12
21
22self.tickers_to_ignore:List[str] = ['SGA']
23self.data:Dict[Symbol, SymbolData] = {}
24self.weight:Dict[Symbol, float] = {}
25
26self.selection_flag:bool = False
27self.rebalance_flag:bool = False
28self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
29self.UniverseSettings.Resolution = Resolution.Daily
30self.AddUniverse(self.FundamentalSelectionFunction)
31self.Schedule.On(self.DateRules.MonthStart(self.market), self.TimeRules.AfterMarketOpen(self.market), self.Selection)
32
33self.current_year:int = -1
34
35def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
36for security in changes.AddedSecurities:
37    security.SetFeeModel(data_tools.CustomFeeModel())
38    security.SetLeverage(self.leverage)
39
40def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
41# store daily stock prices
42for stock in fundamental:
43    symbol:Symbol = stock.Symbol
44
45    if symbol in self.data:
46        self.data[symbol].update_daily_return(stock.AdjustedPrice)
47
48# monthly selection
49if not self.selection_flag:
50    return Universe.Unchanged
51self.selection_flag = False
52
53# store monthly market prices
54if self.market in self.data:
55    self.data[self.market].update_monthly_return(self.Securities[self.market].Price, 0)
56
57# store monthly stock prices
58for stock in fundamental:
59    symbol:Symbol = stock.Symbol
60    if stock.Volume != 0:
61        volume:float = stock.Volume
62    else:
63        continue
64
65    if symbol in self.data:
66        self.data[symbol].update_monthly_return(stock.AdjustedPrice, volume)
67
68fundamental:List[Fundamental] = [x for x in fundamental if x.HasFundamentalData and x.MarketCap != 0 and x.Market == 'usa' \
69                        and not np.isnan(x.FinancialStatements.BalanceSheet.TotalAssets.TwelveMonths) and x.FinancialStatements.BalanceSheet.TotalAssets.TwelveMonths != 0 \
70                        and not np.isnan(x.FinancialStatements.IncomeStatement.GrossProfit.TwelveMonths) and x.FinancialStatements.IncomeStatement.GrossProfit.TwelveMonths != 0 \
71                        and not np.isnan(x.FinancialStatements.IncomeStatement.OperatingExpense.TwelveMonths) and x.FinancialStatements.IncomeStatement.OperatingExpense.TwelveMonths != 0 \
72                        and not np.isnan(x.FinancialStatements.IncomeStatement.DepreciationAndAmortization.TwelveMonths) and x.FinancialStatements.IncomeStatement.DepreciationAndAmortization.TwelveMonths != 0 \
73                        and ((x.SecurityReference.ExchangeId == "NYS") or (x.SecurityReference.ExchangeId == "NAS") or (x.SecurityReference.ExchangeId == "ASE")) \
74                        and x.Symbol.Value not in self.tickers_to_ignore]
75
76if len(fundamental) > self.fundamental_count:
77    selection = {x.Symbol:x for x in sorted(fundamental, key=lambda x:x.MarketCap, reverse=True)[:self.fundamental_count]}
78else:
79    selection = {x.Symbol:x for x in fundamental}
80
81# price warmup
82for symbol in list(selection.keys()):
83    if symbol in self.data:
84        continue
85    
86    self.data[symbol] = data_tools.SymbolData(self.daily_period, self.monthly_period)
87    history:DataFrame = self.History(symbol, self.daily_period, Resolution.Daily)
88    if history.empty:
89        self.Log(f"Not enough data for {symbol} yet.")
90        continue
91    data:pd.DataFrame = history.loc[symbol]
92    monthly_data = data.groupby(pd.Grouper(freq='MS')).last()
93    for time, row in monthly_data.iterrows():
94        self.data[symbol].update_monthly_return(row.close, row.volume)
95    for time, row in data.iterrows():
96        self.data[symbol].update_daily_return(row.close)
97
98if self.market not in self.data:
99    self.data[self.market] = data_tools.SymbolData(self.daily_period, self.monthly_period)
100    history:DataFrame = self.History(self.market, self.daily_period, Resolution.Daily)
101    if not history.empty:
102        data:pd.DataFrame = history.loc[self.market].groupby(pd.Grouper(freq='MS')).last()
103        for time, row in data.iterrows():
104            self.data[self.market].update_monthly_return(row.close, 0)
105    else:
106        self.Log(f"Not enough data for {symbol} yet.")
107 
108if self.Time.year != self.current_year:
109    self.current_year = self.Time.year
110    [self.data[sym].update_total_assets(selection[sym].FinancialStatements.BalanceSheet.TotalAssets.TwelveMonths) for sym in selection]
111
112# stock returns
113returns_by_stock:Dict[Symbol, List[Tuple[datetime, float]]] = {sym : sym_data.get_monthly_returns() for sym, sym_data in self.data.items() if sym_data.is_ready() and sym_data.total_asset_is_ready() and sym in selection.keys() and sym != self.market}
114stock_returns:List = list(zip(*[[i for i in x] for x in returns_by_stock.values()]))
115
116market_returns:np.ndarray = np.array(self.data[self.market].get_monthly_returns())
117
118if len(stock_returns) == 0:
119    return Universe.Unchanged
120
121# run stock regression
122x:np.ndarray = market_returns
123y:np.ndarray = np.array(stock_returns)
124model = data_tools.multiple_linear_regression(x, y)
125beta_values:np.ndarray = model.params[1]
126resid = model.resid
127
128data:Dict[str, float] = {}
129
130# get all factors
131data['symbols'] = [sym for sym in list(returns_by_stock.keys())]
132data['betas'] =  [beta_values[n] for n, sym in enumerate(list(returns_by_stock.keys()))]
133data['VOL'] = [self.data[sym].get_volatility() for sym in list(returns_by_stock.keys())]
134data['CORR'] =  [np.corrcoef(market_returns, self.data[sym].get_monthly_returns())[0, 1] for sym in list(returns_by_stock.keys())]
135data['IVOL'] = [np.std(resid.T[n]) for n, sym in enumerate(list(returns_by_stock.keys()))]
136data['ME'] = [selection[sym].MarketCap for sym in list(returns_by_stock.keys())]
137data['BM'] = [selection[sym].ValuationRatios.PBRatio for sym in list(returns_by_stock.keys())] 
138data['OP'] = [(selection[sym].FinancialStatements.IncomeStatement.GrossProfit.TwelveMonths - selection[sym].FinancialStatements.IncomeStatement.OperatingExpense.TwelveMonths - selection[sym].FinancialStatements.IncomeStatement.DepreciationAndAmortization.TwelveMonths) for sym in list(returns_by_stock.keys())]
139data['AG'] = [self.data[sym].get_asset_growth() for sym in list(returns_by_stock.keys())]
140data['MOM'] = [self.data[sym].get_momentum() for sym in list(returns_by_stock.keys())]
141data['STR'] = [self.data[sym].get_str() for sym in list(returns_by_stock.keys())]
142data['MAX'] = [self.data[sym].get_max() for sym in list(returns_by_stock.keys())]
143data['M1'] = [self.data[sym].get_m1() for sym in list(returns_by_stock.keys())]
144data['ILLIQ'] = [self.data[sym].get_illiquidity() for sym in list(returns_by_stock.keys())]
145
146# create dataframe from factors
147df = pd.DataFrame(data)
148
149# rank stocks by percentile
150df = df.dropna(axis=0)
151df[df.columns[1:]] = df[df.columns[1:]].rank(axis=0, pct=True)
152df['average_percentile'] = df[df.columns[1:]].mean(axis=1)
153
154# sort by average percentile and divide to quantiles
155if len(df) >= self.quantile:
156    sorted_symbols:List[Symbol] = df.sort_values(by='average_percentile', ascending=True).symbols.values
157    quantile:int = int(len(sorted_symbols) / self.quantile)
158    long:List[FineFundamental] = list(sorted_symbols)[:quantile]
159    short:List[FineFundamental] = list(sorted_symbols)[-quantile:]
160
161    # calculate weights based on values
162    sum_long:float = sum([selection[x].MarketCap for x in long])
163    for stock in long:
164        self.weight[stock] = selection[stock].MarketCap / sum_long
165
166    sum_short:float = sum([selection[x].MarketCap for x in short])
167    for stock in short:
168        self.weight[stock] = -selection[stock].MarketCap / sum_short
169    
170    self.rebalance_flag = True
171
172return list(selection.keys())
173
174def OnData(self, data: Slice) -> None:
175if not self.rebalance_flag:
176    return
177self.rebalance_flag = False
178
179# trade execution
180if self.data[self.market].get_last_month_return() < 0:
181    invested:List[Symbol] = [x.Key for x in self.Portfolio if x.Value.Invested]
182    for price_symbol in invested:
183        if price_symbol not in self.weight:
184            self.Liquidate(price_symbol)
185    
186    for price_symbol, weight in self.weight.items():
187        if price_symbol in data and data[price_symbol]:
188            self.SetHoldings(price_symbol, weight)
189else:
190    if self.Portfolio.Invested:
191        self.Liquidate()
192
193self.weight.clear()
194
195def Selection(self):
196self.selection_flag = True