Cross-Sectional Mood Beta 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 CrossSectionalMoodBetaStrategyinEquities(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 = 5
17self.period:int = self.regression_year_window * 12
18self.mood_beta_period:int = 5
19self.historical_mood_beta:Dict[str, RollingWindow] = {}
20        
21# four prespecified (January, March, September, and October) months
22self.high_mood_months:List[int] = [1, 3]
23self.low_mood_months:List[int] = [9, 10]
24
25self.weight:Dict[Symbol, float] = {}
26self.quantile:int = 10
27
28self.leverage:int = 5
29self.coarse_count:int = 500
30self.selection_flag:bool = False
31self.rebalance_flag:bool = False
32self.UniverseSettings.Resolution = Resolution.Daily
33self.AddUniverse(self.CoarseSelectionFunction, self.FineSelectionFunction)
34self.Schedule.On(self.DateRules.MonthStart(self.market), self.TimeRules.AfterMarketOpen(self.market), self.Selection)
35
36def OnSecuritiesChanged(self, changes:SecurityChanges) -> None:
37for security in changes.RemovedSecurities:
38    if security.Symbol in self.historical_mood_beta:
39        del self.historical_mood_beta[security.Symbol]
40    
41for security in changes.AddedSecurities:
42    security.SetFeeModel(CustomFeeModel())
43    security.SetLeverage(self.leverage)
44
45def CoarseSelectionFunction(self, coarse:List[CoarseFundamental]) -> List[Symbol]:
46if not self.selection_flag:
47    return Universe.Unchanged
48self.selection_flag = False
49
50selected = [x.Symbol
51    for x in sorted([x for x in coarse if x.HasFundamentalData and x.Market == 'usa'],
52        key = lambda x: x.DollarVolume, reverse = True)[:self.coarse_count]]
53# selected:List[Symbol] = [x.Symbol for x in coarse if x.HasFundamentalData and x.Market == 'usa']
54
55return selected
56        
57def FineSelectionFunction(self, fine:List[FineFundamental]) -> List[Symbol]:
58fine = [x.Symbol for x in fine if x.MarketCap != 0 and \
59            ((x.SecurityReference.ExchangeId == "NYS") or (x.SecurityReference.ExchangeId == "NAS") or (x.SecurityReference.ExchangeId == "ASE"))]
60
61history:DataFrame = self.History(fine + [self.market], start=self.Time.date() - relativedelta(months=self.period), end=self.Time.date())['close'].unstack(level=0)
62history = history.groupby(pd.Grouper(freq='M')).last()
63
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
70    # find best and worst performing months
71    market_returns:DataFrame = asset_returns[self.market]
72    asset_returns = asset_returns.loc[:, asset_returns.columns != self.market]    # drop market column
73    year_range:List[int] = range(self.Time.year - (self.regression_year_window), self.Time.year-1)
74    
75    # four realized high and low mood months
76    highest_performing_months:np.ndarray = list(np.array([market_returns[market_returns.index.year == year].nlargest(2).index.date for year in year_range]).reshape(-1))
77    lowest_performing_months:np.ndarray = list(np.array([market_returns[market_returns.index.year == year].nsmallest(2).index.date for year in year_range]).reshape(-1))
78
79    # four prespecified (January, March, September, and October)
80    prespecified_months:List[datetime.date] = []
81    for year in [market_returns[market_returns.index.year == year].index for year in year_range]:
82        for date in year:
83            if date.month in [self.high_mood_months + self.low_mood_months]:
84                prespecified_months.append(date.date())
85
86    selected_months:Set[datetime.date] = sorted(set(highest_performing_months + lowest_performing_months + prespecified_months), key=lambda x: x, reverse=False)
87    
88    # run regression
89    x:np.ndarray = market_returns.loc[selected_months].values
90    y:np.ndarray = asset_returns.loc[selected_months].values
91    model = self.multiple_linear_regression(x, y)
92    beta_values:np.ndarray = model.params[1]
93    
94    # store historical beta values
95    beta_by_asset:Dict[str, float] = {}
96    assets:List[str] = list(asset_returns.columns)
97
98    for i, asset in enumerate(assets):
99        asset_s:Symbol = self.Symbol(asset)
100        beta_by_asset[asset_s] = beta_values[i]
101
102    # sort by mean beta
103    if len(beta_by_asset) >= self.quantile:
104        sorted_by_beta:List[Symbol] = sorted(beta_by_asset, key=beta_by_asset.get, reverse=True)
105        quantile:int = int(len(sorted_by_beta) / self.quantile)
106        long:List[Symbol] = sorted_by_beta[:quantile]
107        short:List[Symbol] = sorted_by_beta[-quantile:]
108
109        # EW
110        for asset in long:
111            self.weight[asset] = 1. / float(len(long))
112        for asset in short:
113            self.weight[asset] = -1. / float(len(short))
114
115return list(self.weight.keys())
116
117def OnData(self, data:Slice) -> None:
118# monthly rebalance
119if not self.rebalance_flag:
120    return
121self.rebalance_flag = False
122
123# long the highest decile and short the lowest mood beta decile during the high-mood months 
124# (January and March) and flip the long and short lags during the low-mood months (September and October)
125trade_direction:float = 0.
126if self.Time.month in self.high_mood_months:
127    trade_direction = 1.
128elif self.Time.month in self.low_mood_months:
129    trade_direction = -1.
130
131# rebalance
132for symbol, w in self.weight.items():
133    self.SetHoldings(symbol, trade_direction*w)
134
135def Selection(self) -> None:
136# monthly rebalance
137self.rebalance_flag = True
138
139# yearly selection
140if self.Time.month == 1:
141    self.selection_flag = True
142    self.weight.clear()
143
144def multiple_linear_regression(self, x:np.ndarray, y:np.ndarray):
145x:np.ndarray = np.array(x).T
146x = sm.add_constant(x)
147result = sm.OLS(endog=y, exog=x).fit()
148return result
149
150# custom fee model
151class CustomFeeModel(FeeModel):
152def GetOrderFee(self, parameters):
153fee:float = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
154return OrderFee(CashAmount(fee, "USD"))