Industry-adjusted Reversal

1from AlgorithmImports import *
2import data_tools
3from typing import List, Dict, Set
4from dateutil.relativedelta import relativedelta
5from numpy import isnan
6# endregion
7
8class IndustryAdjustedReversal(QCAlgorithm):
9
10def Initialize(self):
11self.SetStartDate(2000, 1, 1)
12self.SetCash(100000)
13
14self.exchange_codes:List[str] = ['NYS', 'AMEX', 'NAS']    
15
16self.fundamental_count:int = 1000
17self.fundamental_sorting_key = lambda x: x.DollarVolume
18
19self.leverage:int = 5
20self.quantile:int = 10
21self.monthly_period:int = 2
22
23self.data:Dict[Symbol, SymbolData] = {}
24self.weight:Dict[Symbol, float] = {}
25
26self.current_month:int = -1
27self.rebalance_flag:bool = False
28self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
29self.UniverseSettings.Resolution = Resolution.Daily
30self.AddUniverse(self.FundamentalSelectionFunction)
31
32def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
33for security in changes.AddedSecurities:
34    security.SetFeeModel(data_tools.CustomFeeModel())
35    security.SetLeverage(self.leverage)
36
37def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
38if self.Time.month == self.current_month:
39    return Universe.Unchanged
40self.current_month = self.Time.month
41
42 # store monthly prices
43for stock in fundamental:
44    symbol:Symbol = stock.Symbol
45
46    if symbol in self.data:
47        self.data[symbol].update_monthly_price(stock.AdjustedPrice)
48
49selected:List[Fundamental] = [x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' \
50                and x.MarketCap != 0 and not np.isnan(x.AssetClassification.MorningstarIndustryGroupCode) and x.AssetClassification.MorningstarIndustryGroupCode != 0 \
51                and x.SecurityReference.ExchangeId in self.exchange_codes]
52
53if len(selected) > self.fundamental_count:
54    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
55
56# store stocks by industry code
57industries:Set[MorningstarIndustryGroupCode] = set([x.AssetClassification.MorningstarIndustryGroupCode for x in selected])
58grouped_industries:Dict[MorningstarIndustryGroupCode, List[Symbol]] = { industry : [stock.Symbol for stock in selected if stock.AssetClassification.MorningstarIndustryGroupCode == industry] for industry in industries }
59
60# sort stocks by industry numbers and price warmup
61ISR:Dict[Symbol, float] = {}
62for stock in selected:
63    symbol:Symbol = stock.Symbol
64
65    if symbol not in self.data:               
66        self.data[symbol] = data_tools.SymbolData(self.monthly_period)
67
68        history:DataFrame = self.History(symbol, start=self.Time.date() - relativedelta(months=1), end=self.Time.date()).unstack(level=0)
69        if history.empty:
70            self.Log(f"Not enough data for {symbol} yet.")
71            continue
72        history = history.close.groupby(pd.Grouper(freq='MS')).first()
73        for time, close in history.iterrows():
74            self.data[symbol].update_monthly_price(float(close.values))
75
76    if self.data[symbol].is_ready():
77        industry_return:float = np.mean([self.data[x].get_monthly_return() for x in grouped_industries[stock.AssetClassification.MorningstarIndustryGroupCode] if x in self.data and self.data[x].is_ready()])
78        stock_return:float = self.data[symbol].get_monthly_return()
79        ISR[stock] = stock_return - industry_return
80
81if len(ISR) >= self.quantile:
82    sorted_symbols:List[Symbol] = sorted(ISR, key=ISR.get)
83    quantile:int = len(ISR) // self.quantile
84    long:List[Symbol] = sorted_symbols[:quantile]
85    short:List[Symbol] = sorted_symbols[-quantile:]
86
87    for i, portfolio in enumerate([long, short]):
88        mc_sum:float = sum(list(map(lambda stock: stock.MarketCap, portfolio)))
89        for stock in portfolio:
90            self.weight[stock.Symbol] = ((-1)**i) / len(portfolio)
91    
92    self.rebalance_flag = True
93
94return list(self.weight.keys())
95
96def OnData(self, data: Slice) -> None:
97if not self.rebalance_flag:
98    return
99self.rebalance_flag = False
100
101# trade execution
102portfolio:List[PortfolioTarget] = [PortfolioTarget(symbol, w) for symbol, w in self.weight.items() if symbol in data and data[symbol]]
103self.SetHoldings(portfolio, True)
104self.weight.clear()