Industry Herding and Momentum

1from AlgorithmImports import *
2import numpy as np
3from typing import List, Dict, Tuple
4class IndustryHerdingandMomentum(QCAlgorithm):
5
6def Initialize(self):
7self.SetStartDate(2003, 1, 1)
8self.SetCash(100000)
9self.period:int = 7 * 21
10self.leverage:int = 5
11self.min_share_price:int = 5
12self.min_stocks:int = 5
13# Daily price data.
14self.data:Dict[Symbol, SymbolData] = {}
15
16market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
17
18self.long:List[Symbol] = []
19self.short:List[Symbol] = []
20
21# Historical industry CSSD data.
22self.industry_CSSD:Dict[str, float] = {}
23self.cssd_period:int = 3    # Minimum cssd data period.
24
25self.fundamental_count:int = 1000
26self.fundamental_sorting_key = lambda x: x.DollarVolume
27self.selection_flag:bool = True
28self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
29self.UniverseSettings.Resolution = Resolution.Daily
30self.AddUniverse(self.FundamentalSelectionFunction)
31self.Schedule.On(self.DateRules.MonthEnd(market), self.TimeRules.AfterMarketOpen(market), self.Selection)
32self.settings.daily_precise_end_time = False
33def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
34for security in changes.AddedSecurities:
35    security.SetFeeModel(CustomFeeModel())
36    security.SetLeverage(self.leverage)
37        
38def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
39# Update the rolling window every day.
40for stock in fundamental:
41    symbol:Symbol = stock.Symbol
42    # Store monthly price.
43    if symbol in self.data:
44        self.data[symbol].update(stock.AdjustedPrice)
45if not self.selection_flag:
46    return Universe.Unchanged
47selected:List[Fundamental] = [
48    x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and x.Price >= self.min_share_price and x.MarketCap != 0
49]
50if len(selected) > self.fundamental_count:
51    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
52industry_group:Dict[str, List[Symbol]] = {}
53# Warmup price rolling windows.
54for stock in selected:
55    symbol:Symbol = stock.Symbol
56    if symbol not in self.data:
57        self.data[symbol] = SymbolData(self.period)
58        history:DataFrame = self.History(symbol, self.period, Resolution.Daily)
59        if history.empty:
60            self.Log(f"Not enough data for {symbol} yet")
61            continue
62        closes:Series = history.loc[symbol].close
63        for time, close in closes.items():
64            self.data[symbol].update(close)
65    if not self.data[symbol].is_ready():
66        continue
67
68    # Adding stocks in groups
69    industry_group_code:str = stock.AssetClassification.MorningstarIndustryGroupCode
70    
71    if industry_group_code == 0: continue
72    
73    if not industry_group_code in industry_group:
74        industry_group[industry_group_code] = []
75    industry_group[industry_group_code].append(symbol)
76
77# CSSD calc.
78cssd_performance_data:Dict[str, Tuple[float]] = {}
79for group_code, group_symbols in industry_group.items():
80    # Groups with at least 5 stocks, so CSSD is worth to calculate.
81    if len(group_symbols) >= self.min_stocks:
82        # Calculate CSSD.
83        performance_1M:List[float] = []      # Last month's performance to calculate CSSD.
84        performance_6M:List[float] = []      # 6M momentum.
85        
86        for symbol in group_symbols:
87            performances:Tuple[float] = self.data[symbol].performances()
88            performance_1M.append(performances[0]) # Last month performance
89            performance_6M.append(performances[1]) # First six months performance
90                
91        if len(performance_1M) >= self.min_stocks and len(performance_1M) == len(performance_6M):
92            avg_return:float = np.mean(performance_1M)
93            cssd:float = sqrt( ( sum([(x-avg_return)**2 for x in performance_1M]) / (len(performance_1M)-1) ) )
94            
95            if group_code not in self.industry_CSSD:
96                self.industry_CSSD[group_code] = []
97            
98            if len(self.industry_CSSD[group_code]) >= self.cssd_period:
99                normalized_cssd:float = (cssd - np.mean(self.industry_CSSD[group_code])) / np.std(self.industry_CSSD[group_code])
100                avg_momentum:float = np.mean(performance_6M)  # Group average momentum for last 6 months. (skipped last one)
101                
102                cssd_performance_data[group_code] = (normalized_cssd, avg_momentum)
103                
104            self.industry_CSSD[group_code].append(cssd)
105
106if len(cssd_performance_data) != 0:
107    sorted_by_cssd:List[Tuple[str, float]] = sorted(cssd_performance_data.items(), key = lambda x: x[1][0], reverse = True)
108    count:int = int(len(sorted_by_cssd) * 0.3)
109    low_herding:List[Tuple[str, float]] = [x for x in sorted_by_cssd[:count]]
110    
111    sorted_by_momentum:List[Tuple[str, float]] = sorted(low_herding, key = lambda x: x[1][1], reverse = True)
112    count:int = int(len(sorted_by_momentum) * 0.5)
113    long_groups:List[str] = [x[0] for x in sorted_by_momentum[:count]]
114    short_groups:List[str] = [x[0] for x in sorted_by_momentum[-count:]]
115    
116    self.long:List[Symbol] = [symbol for x in long_groups for symbol in industry_group[x]]
117    self.short:List[Symbol] = [symbol for x in short_groups for symbol in industry_group[x]]
118
119return self.long + self.short
120                
121def Selection(self) -> None:
122self.selection_flag = True
123
124def OnData(self, data: Slice) -> None:
125if not self.selection_flag:
126    return
127self.selection_flag = False
128# Trade execution.
129targets:List[PortfolioTarget] = []
130for i, portfolio in enumerate([self.long, self.short]):
131    for symbol in portfolio:
132        if symbol in data and data[symbol]:
133            targets.append(PortfolioTarget(symbol, ((-1) ** i) / len(portfolio)))
134
135self.SetHoldings(targets, True)
136
137self.long.clear()
138self.short.clear()
139
140class SymbolData():
141def __init__(self, period:int):
142self._monthly_closes:RollingWindow = RollingWindow[float](period)
143
144def update(self, close:float) -> None:
145self._monthly_closes.Add(close)
146
147def is_ready(self) -> bool:
148return self._monthly_closes.IsReady
149
150def performances(self) -> Tuple[float]:
151monthly_closes:List[float] = list(self._monthly_closes)
152last_month:List[float] = monthly_closes[:21]
153first_six_months:List[float] = monthly_closes[21:]
154
155last_month_performance:float = last_month[0] / last_month[-1] - 1
156first_months_performance:float = first_six_months[0] / first_six_months[-1] - 1
157
158return (last_month_performance, first_months_performance)
159
160# Custom fee model
161class CustomFeeModel(FeeModel):
162def GetOrderFee(self, parameters):
163fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
164return OrderFee(CashAmount(fee, "USD"))