Market Timing Filter Applied to a Momentum and Other Factor Strategies

1from AlgorithmImports import *
2import numpy as np
3from numpy import isnan
4class MarketTimingFilterAppliedMomentumOtherFactorStrategies(QCAlgorithm):
5def Initialize(self):
6self.SetStartDate(2000, 1, 1)
7self.SetCash(100000)
8market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
9
10self.SMA_period:int = 24
11self.period:int = 13
12self.quantile:int = 10
13self.leverage:int = 5
14self.fundamental_count:int = 500
15self.fundamental_sorting_key = lambda x: x.DollarVolume
16# Equity holdings value.
17self.mimic_equity_value = self.Portfolio.TotalPortfolioValue
18self.holdings_value:Dict[Symbol, List[float]] = {}
19self.equity_sma = SimpleMovingAverage(self.SMA_period)
20
21# Monthly close data.
22self.data:Dict[Symbol, SymbolData] = {}
23self.weight:Dict[Symbol, float] = {}
24
25self.plot = Chart('Strategy EQ')
26self.plot.AddSeries(Series('EQ', SeriesType.Line, 0))
27
28self.selection_flag:bool = False
29self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
30self.UniverseSettings.Resolution = Resolution.Daily
31self.AddUniverse(self.FundamentalSelectionFunction)
32self.Schedule.On(self.DateRules.MonthEnd(market), self.TimeRules.AfterMarketOpen(market), self.Selection)
33self.settings.daily_precise_end_time = False
34def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
35for security in changes.AddedSecurities:
36    security.SetSlippageModel(CustomSlippageModel())
37    security.SetFeeModel(CustomFeeModel())
38    security.SetLeverage(self.leverage)
39
40def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
41if not self.selection_flag:
42    return Universe.Unchanged
43# Update the rolling window every month.
44for stock in fundamental:
45    symbol:Symbol = stock.Symbol
46    # Store monthly price.
47    if symbol in self.data:
48        self.data[symbol].update(stock.AdjustedPrice)
49    
50selected:List[Funamental] = [
51    x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' \
52    and not isnan(x.EarningReports.BasicAverageShares.ThreeMonths) and x.EarningReports.BasicAverageShares.ThreeMonths > 0 \
53    and not isnan(x.EarningReports.BasicEPS.TwelveMonths) and x.EarningReports.BasicEPS.TwelveMonths > 0 \
54    and not isnan(x.ValuationRatios.PERatio) and x.ValuationRatios.PERatio > 0
55]
56if len(selected) > self.fundamental_count:
57    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
58    
59performance_market_cap:Dict[Symbol, List[float]] = {}
60# Warmup price rolling windows.
61for stock in selected:
62    symbol:Symbol = stock.Symbol
63    if symbol not in self.data: 
64        self.data[symbol] = SymbolData(self.period)
65        history:DataFrame = self.History(symbol, self.period*30, Resolution.Daily)
66        if history.empty:
67            self.Log(f"Not enough data for {symbol} yet.")
68            continue
69        closes:Series = history.loc[symbol].close
70        
71        closes_len:int = len(closes.keys())
72        # Find monthly closes.
73        for index, time_close in enumerate(closes.items()):
74            # index out of bounds check.
75            if index + 1 < closes_len:
76                date_month:int = time_close[0].date().month
77                next_date_month:int = closes.keys()[index + 1].month
78            
79                # Found last day of month.
80                if date_month != next_date_month:
81                    self.data[symbol].update(time_close[1])
82
83    if not self.data[symbol].is_ready():
84        continue
85    
86    # Market cap calc.
87    market_cap:float = float(stock.EarningReports.BasicAverageShares.ThreeMonths * (stock.EarningReports.BasicEPS.TwelveMonths * stock.ValuationRatios.PERatio))
88    
89    performance_market_cap[symbol] = [self.data[symbol].performance(), market_cap]
90        
91if len(performance_market_cap) <= self.quantile:
92    return Universe.Unchanged
93
94# Return sorting.
95sorted_by_ret:List[Tuple[Symbol, List[float]]] = sorted(performance_market_cap.items(), key = lambda x: x[1][0], reverse = True)
96quantile:int = int(len(sorted_by_ret) / self.quantile)
97long:List[Tuple[Symbol, List[float]]] = [x for x in sorted_by_ret[:quantile]]
98
99# Market cap weighting.
100total_market_cap:float = sum([x[1][1] for x in long])
101for symbol, perf_market_cap in long:
102    self.weight[symbol] = perf_market_cap[1] / total_market_cap
103
104return list(self.weight.keys())
105
106def OnData(self, data: Slice) -> None:
107if not self.selection_flag:
108    return
109self.selection_flag = False
110
111# Trade execution
112if len(self.weight) == 0: 
113    self.Liquidate()
114    return
115stocks_invested:List[Symbol] = [x.Key for x in self.Portfolio if x.Value.Invested]
116for symbol in stocks_invested:
117    if symbol not in self.weight:
118        self.Liquidate(symbol)
119
120# Calculate symbol equity. - mimic trading.
121for symbol, holdings in self.holdings_value.items():
122    curr_price:float = self.Securities[symbol].Price
123    
124    holdings_price:float = holdings[0]
125    holdings_q:float = holdings[1]
126    fee:float = holdings_price * abs(holdings_q) * 0.00005
127    slippage:float = curr_price * float(0.0001 * np.log10(2*float(abs(holdings_q))))
128    
129    last_holdings_value:float = holdings_price * holdings_q - fee - slippage
130    new_holdings_value:float = (curr_price * holdings_q)
131    trade_pl:float = (new_holdings_value - last_holdings_value)
132    self.mimic_equity_value += trade_pl
133self.equity_sma.Update(self.Time, self.mimic_equity_value)
134self.Plot("Strategy EQ", "EQ", self.mimic_equity_value)
135# self.Log('Real portfolio value: {0}; Alternative portfolio value: {1}'.format(self.Portfolio.TotalPortfolioValue, self.mimic_equity_value))
136self.holdings_value.clear()
137
138for symbol, w in self.weight.items():
139    if symbol in data and data[symbol]:
140        # Store symbol equity holdings. - mimic trading.
141        curr_price:float = data[symbol].Value
142        if curr_price != 0:
143            q:float = (self.mimic_equity_value * w) / curr_price
144            
145            self.holdings_value[symbol] = [curr_price, q]
146            if self.equity_sma.IsReady:
147                if self.mimic_equity_value > self.equity_sma.Current.Value:
148                    self.SetHoldings(symbol, w)
149            else:
150                continue
151            
152self.weight.clear()
153def Selection(self) -> None:
154self.selection_flag = True
155            
156class SymbolData():
157def __init__(self, period: int) -> None:
158self.Closes:RollingWindow = RollingWindow[float](period)
159
160def update(self, close: float) -> None:
161self.Closes.Add(close)
162
163def is_ready(self) -> bool:
164return self.Closes.IsReady
165
166def performance(self) -> float:
167closes = [x for x in self.Closes][1:]   # skip last month
168return (closes[0] - closes[-1]) / closes[-1]
169            
170# Custom fee model.
171class CustomFeeModel(FeeModel):
172def GetOrderFee(self, parameters):
173fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
174return OrderFee(CashAmount(fee, "USD"))
175
176# Custom slippage model.
177class CustomSlippageModel:
178def GetSlippageApproximation(self, asset, order):
179# custom slippage math
180slippage = asset.Price * float(0.0001 * np.log10(2*float(order.AbsoluteQuantity)))
181return slippage