Cross-Sectional Mood Reversal Strategy in Equities

1from AlgorithmImports import *
2from typing import List, Dict
3from pandas.core.frame import DataFrame
4import statsmodels.api as sm
5from dateutil.relativedelta import relativedelta
6# endregion
7
8class CrossSectionalMoodReversalStrategyinEquities(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.regression_year_window:int = 10
17self.period:int = self.regression_year_window * 12
18self.noncongruent_period:int = 5
19self.historical_pred_high_ret:Dict[Symbol, RollingWindow] = {}
20self.historical_pred_low_ret:Dict[Symbol, RollingWindow] = {}
21
22self.low_mood_months:List[int] = [1, 3]
23self.high_mood_months:List[int] = [9, 10]
24
25self.weight:Dict[Symbol, float] = {}
26self.quantile:int = 10
27
28self.leverage:int = 3
29self.fundamental_count:int = 500
30self.selection_flag:bool = False
31self.rebalance_flag:bool = False
32self.UniverseSettings.Resolution = Resolution.Daily
33self.AddUniverse(self.FundamentalSelectionFunction)
34self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
35self.Schedule.On(self.DateRules.MonthStart(self.market), self.TimeRules.AfterMarketOpen(self.market), self.Selection)
36
37def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
38for security in changes.RemovedSecurities:
39    if security.Symbol in self.historical_pred_low_ret:
40        del self.historical_pred_low_ret[security.Symbol]
41
42    if security.Symbol in self.historical_pred_high_ret:
43        del self.historical_pred_high_ret[security.Symbol]
44    
45for security in changes.AddedSecurities:
46    security.SetFeeModel(CustomFeeModel())
47    security.SetLeverage(self.leverage)
48
49def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
50if not self.selection_flag:
51    return Universe.Unchanged
52self.selection_flag = False
53
54selected:List[Symbol] = [x.Symbol
55    for x in sorted([x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and x.MarketCap != 0 and \
56        ((x.SecurityReference.ExchangeId == "NYS") or (x.SecurityReference.ExchangeId == "NAS") or (x.SecurityReference.ExchangeId == "ASE"))],
57        key = lambda x: x.DollarVolume, reverse = True)[:self.fundamental_count]]
58   
59is_high_month:bool = False
60is_low_month:bool = False
61
62history:DataFrame = self.History(selected, start=self.Time.date() - relativedelta(months=self.period), end=self.Time.date())['close'].unstack(level=0)
63history = history.groupby(pd.Grouper(freq='M')).last()
64if len(history) >= self.period:
65    history = history.iloc[-self.period:]
66    history.index = history.index.to_pydatetime()
67    
68    asset_returns:DataFrame = history.pct_change().iloc[1:]
69    year_range:List[int] = list(range(self.Time.year - self.regression_year_window, self.Time.year + 1))
70
71    if self.Time.month == self.high_mood_months[1] + 1:
72        is_high_month = True
73        x_var_months:List[int] = self.high_mood_months
74        y_var_months:List[int] = self.low_mood_months
75    elif self.Time.month == self.low_mood_months[1] + 1:
76        is_low_month = True
77        x_var_months:List[int] = self.low_mood_months
78        y_var_months:List[int] = self.high_mood_months
79        
80    # regression X variable data
81    x_prespecified_months:List[datetime.date] = []
82    for year in [asset_returns[asset_returns.index.year == year].index for year in year_range]:
83        for date in year:
84            if date.month in x_var_months:
85                x_prespecified_months.append(date.date())
86
87    # regression Y variable data
88    y_prespecified_months:List[datetime.date] = []
89    for year in [asset_returns[asset_returns.index.year == year].index for year in year_range]:
90        for date in year:
91            if date.month in y_var_months:
92                y_prespecified_months.append(date.date())
93
94    # average two relevant months
95    x:np.ndarray = asset_returns.loc[x_prespecified_months].rolling(2).mean().iloc[::2, :].iloc[1:].values.T
96    y:np.ndarray = asset_returns.loc[y_prespecified_months].rolling(2).mean().iloc[::2, :].iloc[1:].values.T
97
98    pred_ret:Dict[Symbol, float] = {}
99    for i, asset in enumerate(list(asset_returns.columns)):
100        asset_s:Symbol = self.Symbol(asset)
101
102        if not (any(np.isnan(value) for value in x[i]) or any(np.isnan(value) for value in x[i])):
103            model = self.multiple_linear_regression(x[i][:-1], y[i][1:])
104            pred_ret_:float = model.predict(x[i][-1])[0]
105            
106            # store mood month predicted non-congruent return
107            if is_low_month:
108                if asset_s not in self.historical_pred_low_ret:
109                    self.historical_pred_low_ret[asset_s] = RollingWindow[float](self.noncongruent_period)
110                self.historical_pred_low_ret[asset_s].Add(pred_ret_)
111                mood_month_storage:Dict[Symbol, RollingWindow] = self.historical_pred_high_ret
112
113            if is_high_month:
114                if asset_s not in self.historical_pred_high_ret:
115                    self.historical_pred_high_ret[asset_s] = RollingWindow[float](self.noncongruent_period)
116                self.historical_pred_high_ret[asset_s].Add(pred_ret_)
117                mood_month_storage:Dict[Symbol, RollingWindow] = self.historical_pred_low_ret
118
119            # sort all selected number of stocks into deciles based on their average historical non-congruent mood month return during years t−2 through t−5
120            if asset_s in mood_month_storage and mood_month_storage[asset_s].IsReady:
121                avg_mood_return:float = np.mean(list(mood_month_storage[asset_s])[2:])
122                pred_ret[asset_s] = avg_mood_return
123
124    # sort by mean predicted non-congruent return
125    if len(pred_ret) >= self.quantile:
126        sorted_by_ret:List[Symbol] = sorted(pred_ret, key=pred_ret.get)
127        quantile:int = int(len(sorted_by_ret) / self.quantile)
128        long:List[Symbol] = sorted_by_ret[-quantile:]
129        short:List[Symbol] = sorted_by_ret[:quantile]
130
131        # EW
132        for i, portfolio in enumerate([long, short]):
133            for symbol in portfolio:
134                self.weight[symbol] = ((-1) ** i) / len(portfolio)
135
136return list(self.historical_pred_low_ret.keys()) if is_low_month else list(self.historical_pred_high_ret.keys())
137
138def OnData(self, data: Slice) -> None:
139if not self.rebalance_flag:
140    return
141self.rebalance_flag = False
142
143if not(self.Time.month in self.low_mood_months + self.high_mood_months):
144    self.Liquidate()
145else:
146    for price_symbol, weight in self.weight.items():
147        if price_symbol in data and data[price_symbol]:
148            self.SetHoldings(price_symbol, weight)
149
150def Selection(self) -> None:
151if self.Time.month in [self.low_mood_months[1] + 1, self.high_mood_months[1] + 1]:
152    self.weight.clear()
153    self.selection_flag = True
154
155self.rebalance_flag = True
156
157def multiple_linear_regression(self, x:np.ndarray, y:np.ndarray):
158# x:np.ndarray = np.array(x).T
159# x = sm.add_constant(x, prepend=True)
160result = sm.OLS(endog=y, exog=x).fit()
161return result
162
163# custom fee model
164class CustomFeeModel(FeeModel):
165def GetOrderFee(self, parameters):
166fee:float = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
167return OrderFee(CashAmount(fee, "USD"))