Trend Factor within Stocks

1import numpy as np
2from AlgorithmImports import *
3import statsmodels.api as sm
4from typing import Dict, List, Tuple
5from numpy import isnan
6class MomentumStockPickingRSIIndicator(QCAlgorithm):
7def Initialize(self) -> None:
8self.SetStartDate(2000, 1, 1)
9self.SetCash(100000)
10symbol:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
11# RSI indicators.
12self.rsi2:Dict[Symbol, RelativeStrengthIndex] = {}
13self.rsi3:Dict[Symbol, RelativeStrengthIndex] = {}
14self.rsi5:Dict[Symbol, RelativeStrengthIndex] = {}
15self.rsi10:Dict[Symbol, RelativeStrengthIndex] = {}
16
17# Normalized monthly rsi history.
18self.n_rsi2_history:Dict[Symbol, RollingWindow] = {}
19self.n_rsi3_history:Dict[Symbol, RollingWindow] = {}
20self.n_rsi5_history:Dict[Symbol, RollingWindow] = {}
21self.n_rsi10_history:Dict[Symbol, RollingWindow] = {}
22
23self.regression_period:int = 12
24self.period:int = 21
25self.exchange_codes:List[str] = ['NYS', 'NAS', 'ASE']
26self.leverage:int = 5
27self.quantile:int = 5
28
29# Daily price data.
30self.data:Dict[Symbol, RollingWindow] = {}
31self.perf_history:Dict[Symbol, RollingWindow] = {}
32self.fundamental_count:int = 500
33self.fundamental_sorting_key = lambda x: x.DollarVolume
34self.selection_flag:bool = False
35self.current_selection:List[Symbol] = []
36self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
37self.UniverseSettings.Resolution = Resolution.Daily
38self.AddUniverse(self.FundamentalSelectionFunction)
39self.Schedule.On(self.DateRules.MonthEnd(symbol), self.TimeRules.AfterMarketOpen(symbol), self.Selection)
40self.settings.daily_precise_end_time = False
41def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
42for security in changes.AddedSecurities:
43    security.SetFeeModel(CustomFeeModel())
44    security.SetLeverage(self.leverage)
45        
46def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
47# Update the rolling window every day.
48for stock in fundamental:
49    symbol:Symbol = stock.Symbol
50    price:float = stock.AdjustedPrice
51    # Store monthly price.
52    if symbol in self.data:  # If symbol is in self.data, then it is in other dictionaries too. So we don't need to check.
53        self.data[symbol].Update(self.Time, price)
54        self.rsi2[symbol].Update(self.Time, price)
55        self.rsi3[symbol].Update(self.Time, price)
56        self.rsi5[symbol].Update(self.Time, price)
57        self.rsi10[symbol].Update(self.Time, price)
58        
59        if self.selection_flag and self.data[symbol].IsReady: # All indicators are ready, when indicator in self.data is.
60            self.perf_history[symbol].Add(self.data[symbol].Current.Value)
61            
62            if self.rsi2[symbol].Current.Value == 0 or self.rsi3[symbol].Current.Value == 0 or  \
63            self.rsi5[symbol].Current.Value == 0 or self.rsi10[symbol].Current.Value == 0:
64                continue
65            
66            self.n_rsi2_history[symbol].Add(price / self.rsi2[symbol].Current.Value)
67            self.n_rsi3_history[symbol].Add(price / self.rsi3[symbol].Current.Value)
68            self.n_rsi5_history[symbol].Add(price / self.rsi5[symbol].Current.Value)
69            self.n_rsi10_history[symbol].Add(price / self.rsi10[symbol].Current.Value)
70if not self.selection_flag:
71    return Universe.Unchanged
72
73selected:List[Fundamental] = [x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and not \
74        isnan(x.MarketCap) and x.MarketCap != 0 and not isnan(x.CompanyReference.IsREIT) and x.CompanyReference.IsREIT == 0 and \
75        x.SecurityReference.ExchangeId in self.exchange_codes]
76if len(selected) > self.fundamental_count:
77    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
78# Warmup price rolling windows.
79for stock in selected:
80    symbol:Symbol = stock.Symbol
81    if symbol in self.data: # If symbol is in self.data, then it is in other dictionaries too. So we don't need to check.
82        continue
83    self.data[symbol] = RateOfChange(self.period)
84    self.perf_history[symbol] = RollingWindow[float](self.regression_period)
85    
86    self.rsi2[symbol] = RelativeStrengthIndex(2, MovingAverageType.Simple)
87    self.n_rsi2_history[symbol] = RollingWindow[float](self.regression_period)
88    
89    self.rsi3[symbol] = RelativeStrengthIndex(3, MovingAverageType.Simple)
90    self.n_rsi3_history[symbol] = RollingWindow[float](self.regression_period)
91    
92    self.rsi5[symbol] = RelativeStrengthIndex(5, MovingAverageType.Simple)
93    self.n_rsi5_history[symbol] = RollingWindow[float](self.regression_period)
94    
95    self.rsi10[symbol] = RelativeStrengthIndex(10, MovingAverageType.Simple)
96    self.n_rsi10_history[symbol] = RollingWindow[float](self.regression_period)
97    
98    history:DataFrame = self.History(symbol, self.period, Resolution.Daily)
99    if history.empty:
100        self.Log(f"Not enough data for {symbol} yet")
101        continue
102    closes:Close = history.loc[symbol].close
103    for time, close in closes.items():
104        self.data[symbol].Update(time, close)
105        self.rsi2[symbol].Update(time, close)
106        self.rsi3[symbol].Update(time, close)
107        self.rsi5[symbol].Update(time, close)
108        self.rsi10[symbol].Update(time, close)
109self.current_selection = [x.Symbol for x in selected if self.data[x.Symbol].IsReady]
110
111return self.current_selection
112
113def OnData(self, data: Slice) -> None:
114if not self.selection_flag:
115    return
116self.selection_flag = False
117
118expected_return:Dict[Symbol, float] = {}
119for symbol in self.current_selection:
120    if self.perf_history[symbol].IsReady and \
121        self.n_rsi2_history[symbol].IsReady and \
122        self.n_rsi3_history[symbol].IsReady and \
123        self.n_rsi5_history[symbol].IsReady and \
124        self.n_rsi10_history[symbol].IsReady:
125                
126        monthly_returns:List[float] = [x for x in self.perf_history[symbol]][:-1]
127            
128        x:List[List[float]] = [
129            [x for x in self.n_rsi2_history[symbol]][1:],
130            [x for x in self.n_rsi3_history[symbol]][1:],
131            [x for x in self.n_rsi5_history[symbol]][1:],
132            [x for x in self.n_rsi10_history[symbol]][1:]
133        ]
134            
135        # Regression.
136        regression_model = MultipleLinearRegression(x, monthly_returns)
137        alpha:float = regression_model.params[0]
138        beta1:float = regression_model.params[1]
139        beta2:float = regression_model.params[2]
140        beta3:float = regression_model.params[3]
141        beta4:float = regression_model.params[4]
142            
143        x1:float = self.n_rsi2_history[symbol][0]
144        x2:float = self.n_rsi3_history[symbol][0]
145        x3:float = self.n_rsi5_history[symbol][0]
146        x4:float = self.n_rsi10_history[symbol][0]
147            
148        y:float = alpha + (beta1 * x1) + (beta2 * x2) + (beta2 * x3) + (beta3 * x4)
149            
150        expected_return[symbol] = y
151
152long:List[Symbol] = []
153short:List[Symbol] = []
154# Sort by expected return.
155if len(expected_return) >= self.quantile:
156    sorted_by_expected_return:List[Tuple[Symbol, float]] = sorted(expected_return.items(), key = lambda x: x[1], reverse = True)
157    quintile:int = int(len(sorted_by_expected_return) / self.quantile)
158    long = [x[0] for x in sorted_by_expected_return[:quintile]]
159    short = [x[0] for x in sorted_by_expected_return[-quintile:]]
160# Trade execution.
161targets:List[PortfolioTarget] = []
162for i, portfolio in enumerate([long, short]):
163    for symbol in portfolio:
164        if symbol in data and data[symbol]:
165            targets.append(PortfolioTarget(symbol, ((-1) ** i) / len(portfolio)))
166
167self.SetHoldings(targets, True)
168    
169def Selection(self) -> None:
170self.selection_flag = True
171
172# Custom fee model.
173class CustomFeeModel(FeeModel):
174def GetOrderFee(self, parameters):
175fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
176return OrderFee(CashAmount(fee, "USD"))
177def MultipleLinearRegression(x, y):
178x:np.ndarray = np.array(x).T
179x = sm.add_constant(x)
180result = sm.OLS(endog=y, exog=x).fit()
181return result