Value-Growth Timing

1from AlgorithmImports import *
2from numpy import average, std, isnan
3from typing import List, Dict, Tuple
4#endregion    
5class ValueGrowthTiming(QCAlgorithm):
6def Initialize(self):
7self.SetStartDate(2000, 1, 1)
8self.SetCash(100000)
9self.symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
10
11self.fundamental_count:int = 3000
12self.fundamental_sorting_key = lambda x: x.DollarVolume
13
14# Monthly spread data.
15self.period:int = 12
16self.spread:RollinWindow = RollingWindow[float](self.period)
17self.standardized_spread:RollingWindow = RollingWindow[float](self.period)
18self.quantile:int = 10
19self.leverage:int = 10
20
21self.long:List[Symbol] = []
22self.short:List[Symbol] = []
23self.weight:Union[None, int] = None
24self.selection_flag:bool = False
25self.UniverseSettings.Resolution = Resolution.Daily
26self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
27self.UniverseSettings.Resolution = Resolution.Daily
28self.AddUniverse(self.FundamentalSelectionFunction)
29self.Schedule.On(self.DateRules.MonthEnd(self.symbol), self.TimeRules.BeforeMarketClose(self.symbol), self.Selection)
30self.settings.daily_precise_end_time = False
31def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
32for security in changes.AddedSecurities:
33    security.SetFeeModel(CustomFeeModel())
34    security.SetLeverage(self.leverage)
35    
36def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
37if not self.selection_flag:
38    return Universe.Unchanged
39
40selected = [x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' \
41        and not isnan(x.AssetClassification.MorningstarIndustryGroupCode) and x.AssetClassification.MorningstarIndustryGroupCode != 0 \
42        and not isnan(x.ValuationRatios.PBRatio) and x.ValuationRatios.PBRatio != 0 and x.MarketCap != 0]
43if len(selected) > self.fundamental_count:
44    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
45group:Dict[str, Tuple[Symbol, float, float]] = {}
46
47# Stocks returns calc.
48for stock in selected:
49    symbol = stock.Symbol
50    
51    industry_group_code:float = stock.AssetClassification.MorningstarIndustryGroupCode
52    book_market:float = 1 / stock.ValuationRatios.PBRatio
53    
54    # Adding stock in group.
55    if not industry_group_code in group:
56        group[industry_group_code] = []
57    group[industry_group_code].append((symbol, book_market, stock.MarketCap))
58
59# Group's value weighted average calc.
60group_bm_average:Dict[str, float] = {}
61for industry_group_code, symbols in group.items():
62    total_market_cap:float = sum([x[2] for x in symbols])
63    
64    weighted_items:Dict[Symbol, float] = {x[0] : (x[1] * (x[2] / total_market_cap)) for x in symbols}
65    group_bm_average[industry_group_code] = average([x[1] for x in weighted_items.items()])
66
67# Symbol's adjusted bm calc.
68adjusted_bm:Dict[Symbol, float] = {}
69for industry_group_code, symbols in group.items():
70    for symbol in symbols:
71        symbol_bm:float = symbol[1]
72        adjusted_bm[symbol] = symbol_bm - group_bm_average[industry_group_code]
73if len(adjusted_bm) < self.quantile:
74    return Universe.Unchanged
75sorted_by_adjusted_bm:List[Tuple[Symbol, float]] = sorted(adjusted_bm.items(), key = lambda x: x[1], reverse = True)
76quantile:int = int(len(sorted_by_adjusted_bm) / self.quantile)
77value_portfolio:List[Symbol] = sorted_by_adjusted_bm[:quantile]
78growth_portfolio:List[Symbol] = sorted_by_adjusted_bm[-quantile:]
79spread:float = average([x[1] for x in value_portfolio]) - average([x[1] for x in growth_portfolio])
80self.spread.Add(spread)
81if self.spread.IsReady:
82    spread_values:List[float] = [x for x in self.spread]
83    standardized_spread:float = average(spread_values) / std(spread_values)
84    self.standardized_spread.Add(standardized_spread)
85    if self.standardized_spread.IsReady:
86        standardized_spread_values:List[float] = [x for x in self.standardized_spread]
87        
88        self.weight = abs(standardized_spread_values[0] / max(standardized_spread_values[1:]))
89        if self.weight > 1:
90            self.weight = 1
91        self.long = [x[0][0] for x in value_portfolio]
92        self.short = [x[0][0] for x in growth_portfolio]
93    
94return self.long + self.short
95def OnData(self, data: Slice) -> None:
96if not self.selection_flag:
97    return
98self.selection_flag = False
99
100# Trade execution.
101targets:List[PortfolioTarget] = []
102if self.weight:
103    for i, portfolio in enumerate([self.long, self.short]):
104        for symbol in portfolio:
105            if symbol in data and data[symbol]:
106                targets.append(PortfolioTarget(symbol, ((-1) ** i) * (self.weight / len(portfolio))))
107
108self.SetHoldings(targets, True)
109self.long.clear()
110self.short.clear()
111self.weight = None
112def Selection(self) -> None:
113self.selection_flag = True
114
115# Custom fee model.
116class CustomFeeModel(FeeModel):
117def GetOrderFee(self, parameters):
118fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
119return OrderFee(CashAmount(fee, "USD"))