Long-Term PB Ratio Effect in Stocks Combined with Momentum

1from numpy import average
2from AlgorithmImports import *
3class LongTermPBRatio(QCAlgorithm):
4def Initialize(self):
5self.SetStartDate(2000, 1, 1)
6self.SetCash(100000)
7market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
8self.bm_period:int = 10 * 12
9self.momentum_period:int = 13
10self.quantile:int = 5
11self.leverage:int = 5
12self.exchange_codes:List[str] = ['NYS', 'NAS', 'ASE']
13# Daily close data.
14self.data:Dict[Symbol, SymbolData] = {}
15self.long:List[Symbol] = []
16self.fundamental_count:int = 3000
17self.fundamental_sorting_key = lambda x: x.MarketCap
18self.selection_flag:bool = False
19self.UniverseSettings.Resolution = Resolution.Daily
20self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
21self.AddUniverse(self.FundamentalSelectionFunction)
22self.Schedule.On(self.DateRules.MonthEnd(market), self.TimeRules.BeforeMarketClose(market), self.Selection)
23
24self.settings.daily_precise_end_time = False
25
26def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
27for security in changes.AddedSecurities:
28    security.SetFeeModel(CustomFeeModel())
29    security.SetLeverage(self.leverage)
30				
31def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
32if not self.selection_flag:
33    return Universe.Unchanged
34# Update the rolling window every month.
35for stock in fundamental:
36    symbol:Symbol = stock.Symbol
37    # Store monthly price.
38    if symbol in self.data:
39        self.data[symbol].update_price(stock.AdjustedPrice)
40selected:List[Fundamental] = [x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and x.SecurityReference.ExchangeId in self.exchange_codes and \
41    not np.isnan(x.ValuationRatios.BookValuePerShare) and x.ValuationRatios.BookValuePerShare > 0 and x.CompanyReference.IsREIT == 0]
42if len(selected) > self.fundamental_count:
43	selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
44bm_return_data:Dict[Symbol, Tuple[float, float]] = {}
45# Warmup price rolling windows.
46for stock in selected:
47    symbol:Symbol = stock.Symbol
48    
49    if symbol not in self.data:
50        self.data[symbol] = SymbolData(symbol, self, self.momentum_period, self.bm_period)
51        history = self.History(symbol, self.momentum_period*30, Resolution.Daily)
52        if history.empty:
53            self.Log(f"Not enough data for {symbol} yet.")
54            continue
55        closes = history.loc[symbol].close
56        
57        closes_len = len(closes.keys())
58        # Find monthly closes.
59        for index, time_close in enumerate(closes.items()):
60            # index out of bounds check.
61            if index + 1 < closes_len:
62                date_month = time_close[0].date().month
63                next_date_month = closes.keys()[index + 1].month
64            
65                # Found last day of month.
66                if date_month != next_date_month:
67                    self.data[symbol].update_price(time_close[1])
68    
69    if self.data[symbol].performance_is_ready():
70        self.data[symbol].update_book_value(stock.ValuationRatios.BookValuePerShare)
71        # BM ratio calc.
72        if self.data[symbol].book_values_are_ready():
73            book_values:List[float] = list(self.data[symbol]._book_value)
74            book_values_avg:float = average(book_values)
75            
76            last_price:float = self.data[symbol].get_last_price()
77            bm_ratio:float = book_values_avg / last_price
78    
79            # Daily close data is ready.
80            perf:float = self.data[symbol].performance()
81            
82            # Store bm ratio and performance tuple.
83            bm_return_data[symbol] = (bm_ratio, perf)
84if len(bm_return_data) >= self.quantile * 2:
85    sorted_by_bm:List = sorted(bm_return_data.items(), key=lambda x: x[1][0], reverse=True)
86    quantile:int = int(len(sorted_by_bm) / self.quantile)
87    top_by_bm = [x for x in sorted_by_bm[:quantile]]
88    
89    sorted_by_ret:List = sorted(top_by_bm, key=lambda x: bm_return_data[x[0]][1], reverse=True)
90    half = int(len(sorted_by_ret) / 2)
91    self.long = [x[0] for x in sorted_by_ret[:half]]
92return self.long
93def OnData(self, data: Slice) -> None:
94# Rebalance once a month.
95if not self.selection_flag:
96    return
97self.selection_flag = False
98# Trade execution.
99portfolio:List[PortfolioTarget] = [PortfolioTarget(symbol, 1/len(self.long)) for symbol in self.long if symbol in data and data[symbol]]
100self.SetHoldings(portfolio, True)
101self.long.clear()
102
103def Selection(self) -> None:
104self.selection_flag = True
105
106class SymbolData():
107def __init__(self, symbol, algorithm, momentum_period, book_period):
108self._price:RollingWindow = RollingWindow[float](momentum_period)
109self._book_value:RollingWindow = RollingWindow[float](book_period)
110
111def update_book_value(self, value: float) -> None:
112self._book_value.Add(value)
113
114def book_values_are_ready(self) -> bool:
115return self._book_value.IsReady
116
117def update_price(self, price: float) -> None:
118self._price.Add(price)
119
120def performance_is_ready(self) -> bool:
121return self._price.IsReady
122
123def get_last_price(self) -> float:
124return self._price[0]
125# 12 month momentum, one month skipped.
126def performance(self) -> float:
127return self._price[1] / self._price[self._price.Count - 1] - 1
128# Custom fee model.
129class CustomFeeModel(FeeModel):
130def GetOrderFee(self, parameters):
131fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
132return OrderFee(CashAmount(fee, "USD"))