High-Volatility Momentum-Reversal Strategy in Large-Cap Stocks

1import numpy as np
2from AlgoLib import *
3
4class MomentumReversalCombinedWithVolatilityEffectinStocks(XXX):
5
6def Initialize(self):
7self.SetStartDate(2010, 1, 1)
8self.SetCash(100000)
9
10market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
11
12# EW Tranching.
13self.holding_period:int = 6
14self.managed_queue:List[RebalanceQueueItem] = []
15
16# Daily price data.
17self.data:Dict[Symbol, SymbolData] = {}
18self.period:int = 6 * 21
19self.leverage:int = 5
20self.min_share_price:float = 5.
21self.exchange_codes:List[str] = ['NYS', 'NAS', 'ASE']
22self.quantile:int = 5
23
24self.fundamental_count:int = 500
25self.fundamental_sorting_key = lambda x: x.DollarVolume
26
27self.selection_flag = False
28self.UniverseSettings.Resolution = Resolution.Daily
29self.AddUniverse(self.FundamentalSelectionFunction)
30self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
31self.Schedule.On(self.DateRules.MonthStart(market), self.TimeRules.AfterMarketOpen(market), self.Selection)
32
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]) -> None:
39# Update the rolling window every day.
40for stock in fundamental:
41    symbol = stock.Symbol
42
43    # Store monthly price.
44    if symbol in self.data:
45        self.data[symbol].update(stock.AdjustedPrice)
46
47if not self.selection_flag:
48    return Universe.Unchanged
49
50selected:List[Fundamental] = [x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and x.AdjustedPrice > self.min_share_price and \
51    x.SecurityReference.ExchangeId in self.exchange_codes and x.MarketCap != 0]
52if len(selected) > self.fundamental_count:
53    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
54
55sorted_by_market_cap:List[Fundamental] = sorted(selected, key = lambda x: x.MarketCap, reverse=True)
56half:int = int(len(sorted_by_market_cap) / 2)
57top_by_market_cap:List[Symbol] = [x.Symbol for x in sorted_by_market_cap][:half]
58
59perf_volatility:Dict[Symbol, Tuple[float, float]] = {}
60
61# Warmup price rolling windows.
62for stock in selected:
63    symbol = stock.Symbol
64    
65    if symbol not in self.data:
66        self.data[symbol] = SymbolData(symbol, self.period)
67        history = self.History(symbol, self.period, Resolution.Daily)
68        if history.empty:
69            self.Log(f"Not enough data for {symbol} yet.")
70            continue
71        closes = history.loc[symbol].close
72        for time, close in closes.items():
73            self.data[symbol].update(close)
74    
75    # Performance and volatility tuple.
76    if self.data[symbol].is_ready():
77        performance = self.data[symbol].performance()
78        annualized_volatility = self.data[symbol].volatility()
79        perf_volatility[symbol] = (performance, annualized_volatility)                
80
81long:List[Symbol] = []
82short:List[Symbol] = []
83if len(perf_volatility) >= self.quantile:
84    sorted_by_perf:List[Tuple] = sorted(perf_volatility.items(), key = lambda x: x[1][0], reverse = True)
85    quantile:int = int(len(sorted_by_perf) / self.quantile)
86    top_by_perf:List[Symbol] = [x[0] for x in sorted_by_perf[:quantile]]
87    low_by_perf:List[Symbol] = [x[0] for x in sorted_by_perf[-quantile:]]
88    
89    sorted_by_vol:List[Tuple] = sorted(perf_volatility.items(), key = lambda x: x[1][1], reverse = True)
90    quantile = int(len(sorted_by_vol) / self.quantile)
91    top_by_vol:List[Symbol] = [x[0] for x in sorted_by_vol[:quantile]]
92    low_by_vol:List[Symbol] = [x[0] for x in sorted_by_vol[-quantile:]]
93    
94    long = [x for x in top_by_perf if x in top_by_vol]
95    short = [x for x in low_by_perf if x in top_by_vol]
96
97if len(long) != 0:
98    long_w:float = self.Portfolio.TotalPortfolioValue / self.holding_period / len(long)
99    # symbol/quantity collection
100    long_symbol_q:List = [(x, np.ceil(long_w / self.data[x].get_last_price())) for x in long]
101else:
102    long_symbol_q:List = []
103
104if len(short) != 0:
105    short_w:float = self.Portfolio.TotalPortfolioValue / self.holding_period / len(short)
106    # symbol/quantity collection
107    short_symbol_q:List = [(x, -np.ceil(short_w / self.data[x].get_last_price())) for x in short]
108else:
109    short_symbol_q:List = []
110        
111self.managed_queue.append(RebalanceQueueItem(long_symbol_q + short_symbol_q))
112
113return long + short
114
115def OnData(self, data: Slice) -> None:
116if not self.selection_flag:
117    return
118self.selection_flag = False
119   
120remove_item = None
121
122# Rebalance portfolio
123for item in self.managed_queue:
124    if item.holding_period == self.holding_period:
125        for symbol, quantity in item.symbol_q:
126                self.MarketOrder(symbol, -quantity)
127        
128        remove_item = item
129    
130    # Trade execution    
131    if item.holding_period == 0:
132        open_symbol_q = []
133        
134        for symbol, quantity in item.symbol_q:
135            if symbol in data and data[symbol] and self.Securities[symbol].IsTradable:
136                self.MarketOrder(symbol, quantity)
137                open_symbol_q.append((symbol, quantity))
138                    
139        # Only opened orders will be closed        
140        item.symbol_q = open_symbol_q
141        
142    item.holding_period += 1
143    
144# We need to remove closed part of portfolio after loop. Otherwise it will miss one item in self.managed_queue.
145if remove_item:
146    self.managed_queue.remove(remove_item)
147
148def Selection(self) -> None:
149self.selection_flag = True
150
151class RebalanceQueueItem():
152def __init__(self, symbol_q:List):
153# symbol/quantity collections
154self.symbol_q:List = symbol_q  
155self.holding_period:int = 0
156
157class SymbolData():
158def __init__(self, symbol: Symbol, period: int):
159self._symbol:Symbol = symbol
160self._price:RollingWindow = RollingWindow[float](period)
161self._last_price:float = 0
162
163def update(self, price: float) -> None:
164self._price.Add(price)
165self._last_price:float = price
166
167def get_last_price(self) -> float:
168return self._last_price
169
170def is_ready(self) -> bool:
171return self._price.IsReady
172
173def volatility(self) -> float:
174closes:np.ndarray = np.array(list(self._price)[5:]) # Skip last week.
175daily_returns:np.ndarray = closes[:-1] / closes[1:] - 1
176return np.std(daily_returns) * np.sqrt(252 / (len(closes)))
177
178def performance(self) -> float:
179closes:List[float] = list(self._price)[5:] # Skip last week.
180return (closes[0] / closes[-1] - 1)
181
182# Custom fee model.
183class CustomFeeModel(FeeModel):
184def GetOrderFee(self, parameters):
185fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
186return OrderFee(CashAmount(fee, "USD"))