Firm-Level Investor Sentiment Factor in US

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 FirmLevelInvestorSentimentFactorInUS(QCAlgorithm):
9
10def Initialize(self) -> None:
11self.SetStartDate(2010, 1, 1)
12self.SetCash(100000)
13
14self.market:Symbol = self.AddEquity("SPY", Resolution.Daily).Symbol
15
16self.periods:List[int] = [3, 60]
17self.long_period:int = 60
18self.short_period:int = 3
19self.rolling_period:int = 2
20self.short_term_stock_beta:Dict[Symbol, RollingWindow] = {}
21self.long_term_stock_beta:Dict[Symbol, RollingWindow] = {}
22self.long_term_industry_beta:Dict[str, RollingWindow] = {}
23
24self.weight:Dict[Symbol, float] = {}
25self.quantile:int = 10
26self.min_share_price:int = 5
27
28self.leverage:int = 3
29self.fundamental_count:int = 500
30self.selection_flag:bool = False
31self.UniverseSettings.Resolution = Resolution.Daily
32self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
33self.AddUniverse(self.FundamentalSelectionFunction)
34self.Schedule.On(self.DateRules.MonthEnd(self.market), self.TimeRules.BeforeMarketClose(self.market), self.Selection)
35
36def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
37for security in changes.RemovedSecurities:
38    if security.Symbol in self.short_term_stock_beta:
39        del self.short_term_stock_beta[security.Symbol]
40    if security.Symbol in self.long_term_stock_beta:
41        del self.long_term_stock_beta[security.Symbol]
42
43for security in changes.AddedSecurities:
44    security.SetFeeModel(CustomFeeModel())
45    security.SetLeverage(self.leverage)
46
47def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
48# monthly selection
49if not self.selection_flag:
50    return Universe.Unchanged
51
52selected:List[Fundamental] = [x for x in sorted([x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and x.AdjustedPrice >= self.min_share_price and \
53    x.AssetClassification.MorningstarIndustryGroupCode != 0 and x.MarketCap != 0 and \
54    (x.SecurityReference.ExchangeId == 'NYS') or (x.SecurityReference.ExchangeId == 'NAS') or (x.SecurityReference.ExchangeId == 'ASE')], 
55    key = lambda x: x.DollarVolume, reverse = True)][:self.fundamental_count]
56
57history:DataFrame = self.History(list(map(lambda x: x.Symbol, selected)) + [self.market], start=self.Time.date() - relativedelta(months=self.long_period), end=self.Time.date())['close'].unstack(level=0)
58history = history.groupby(pd.Grouper(freq='M')).last()
59
60# sort stocks on industry numbers
61industries:Set[MorningstarIndustryGroupCode] = set([x.AssetClassification.MorningstarSectorCode for x in selected])
62grouped_industries:Dict[MorningstarIndustryGroupCode, List[Symbol]] = { industry : [stock.Symbol for stock in selected if stock.AssetClassification.MorningstarSectorCode == industry] for industry in industries }
63
64# get stock returns and clean up the data
65if len(history) >= self.long_period:
66    history = history.iloc[-self.long_period:]
67    history.index = history.index.to_pydatetime()
68
69    asset_returns:DataFrame = history.pct_change().iloc[1:]
70
71    stock_returns = asset_returns.loc[:, asset_returns.columns != self.market].tail(-1)
72    asset_returns['spy_lag'] = asset_returns[self.market].shift(1)
73    asset_returns = asset_returns.iloc[1:]
74
75    industry_df:DataFrame = pd.DataFrame(index=asset_returns.index)
76
77    for industry_code in grouped_industries:
78        ind_symbols:List[Symbol] = grouped_industries[industry_code]
79        if all(symbol in asset_returns.columns for symbol in ind_symbols):
80            industry_df[industry_code] = asset_returns[ind_symbols].mean(axis=1)
81
82    # run industry regression
83    x:np.ndarray = asset_returns[[self.market, 'spy_lag']].values
84    y:np.ndarray = industry_df.values
85    model = self.multiple_linear_regression(x, y)
86    beta_values:np.ndarray = sum(model.params[1:])
87
88    for i, industry in enumerate(industry_df):
89        if industry not in self.long_term_industry_beta:
90            self.long_term_industry_beta[industry] = RollingWindow[float](self.rolling_period)
91        self.long_term_industry_beta[industry].Add(beta_values[i])
92
93    # run stock regression
94    for period in self.periods:
95        x:np.ndarray = asset_returns[[self.market, 'spy_lag']][-period:].values
96        y:np.ndarray = stock_returns[-period:].values
97        model = self.multiple_linear_regression(x, y)
98        beta_values:np.ndarray = sum(model.params[1:])
99
100        for i, asset in enumerate(stock_returns):
101            asset_s:Symbol = self.Symbol(asset)
102
103            # fill rolling windows with data
104            if asset_s not in self.short_term_stock_beta or asset_s not in self.long_term_stock_beta:
105                if beta_values[i] != 0 and beta_values[i] is not None:
106                    if period == self.periods[0]:
107                        self.short_term_stock_beta[asset_s] = RollingWindow[float](self.rolling_period)
108                    else:
109                        self.long_term_stock_beta[asset_s] = RollingWindow[float](self.rolling_period)
110
111            if beta_values[i] != 0 and beta_values[i] is not None:
112                if period == self.periods[0]:
113                    self.short_term_stock_beta[asset_s].Add(beta_values[i])
114                else:
115                    self.long_term_stock_beta[asset_s].Add(beta_values[i])
116
117beta_by_symbol:Dict[Fundamental, float] = {}
118
119# FIS calculation
120for stock in selected:
121    symbol:Symbol = stock.Symbol
122
123    if symbol in self.short_term_stock_beta or symbol in self.long_term_stock_beta:
124        if self.short_term_stock_beta[symbol].IsReady and self.long_term_stock_beta[symbol].IsReady:
125            bc:float = (self.short_term_stock_beta[symbol][0] - self.short_term_stock_beta[symbol][1]) / self.long_term_stock_beta[symbol][1]
126
127            for industry in grouped_industries:
128                if industry in self.long_term_industry_beta:
129                    if symbol in grouped_industries[industry] and self.long_term_industry_beta[industry].IsReady:
130                        dc:float = (self.short_term_stock_beta[symbol][0] - self.long_term_industry_beta[(industry)][1]) / self.long_term_industry_beta[industry][1]
131                        beta_by_symbol[stock] = bc * dc
132
133# sort by beta and divide to upper decile and lower decile
134if len(beta_by_symbol) >= self.quantile:
135    sorted_by_beta:List[Fundamental] = sorted(beta_by_symbol, key=beta_by_symbol.get)
136    quantile:int = int(len(sorted_by_beta) / self.quantile)
137    long:List[Fundamental] = sorted_by_beta[:quantile]
138    short:List[Fundamental] = sorted_by_beta[-quantile:]
139
140    # calculate weights based on values
141    for i, portfolio in enumerate([long, short]):
142        mc_sum:float = sum(list(map(lambda stock: stock.MarketCap, portfolio)))
143        for stock in portfolio:
144            self.weight[stock.Symbol] = ((-1)**i) * stock.MarketCap / mc_sum
145
146return list(map(lambda x: x.Symbol, selected))
147
148def OnData(self, data: Slice) -> None:
149# monthly rebalance
150if not self.selection_flag:
151    return
152self.selection_flag = False
153
154portfolio:List[PortfolioTarget] = [PortfolioTarget(symbol, w) for symbol, w in self.weight.items() if symbol in data and data[symbol]]
155self.SetHoldings(portfolio, True)
156
157self.weight.clear()
158
159def Selection(self) -> None:
160self.selection_flag = True
161
162def multiple_linear_regression(self, x:np.ndarray, y:np.ndarray):
163# x:np.ndarray = np.array(x).T
164x = sm.add_constant(x, prepend=True)
165result = sm.OLS(endog=y, exog=x).fit()
166return result
167
168# Custom fee model.
169class CustomFeeModel(FeeModel):
170def GetOrderFee(self, parameters):
171fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
172return OrderFee(CashAmount(fee, "USD"))