Equity · United States · stocks · Semi-annually · United States Free

Enduring Momentum in Stocks

Academic paper

AuthorsHui Zeng; Ben R. Marshall; Nhut H. Nguyen; Nuttawat Visaltanachoti

Institute

NZMassey University
?Massey University - Department of Economics and Finance
?Massey University - School of Economics and Finance
NZAuckland University of Technology

Paper linkhttps://papers.ssrn.com/sol3/papers.cfm?abstract_id=4082543

Strategy in a nutshell

The strategy focuses on U.S. stocks listed on NYSE, AMEX, and NASDAQ, excluding stocks priced below $1. At the end of each month, stocks are ranked into ten deciles based on their past six-month returns. Winners (top decile) and losers (bottom decile) are further evaluated using a Cox Proportional Hazard model applied to 37 firm characteristics to estimate the “enduring” probability—how likely a stock is to remain a winner or loser over the next six months. The strategy goes long the winners and short the losers with the highest enduring probabilities, holding these positions for six months.

Economic rationale

While traditional momentum strategies rely solely on past price trends, they are limited by behavioral biases like herding or confirmation bias. This enhanced momentum approach integrates firm-specific characteristics to estimate enduring momentum probability, identifying winners and losers whose trends are more likely to persist. By combining price momentum with this probabilistic assessment, the strategy captures stronger and more reliable return opportunities, outperforming simple price-based momentum strategies.

Backtest performance

Annualised return26.31%

Volatility38.21%

Beta0.029

Sharpe ratio0.69

Sortino ratio-0.083

Win rate47%

Full Python code

1from AlgorithmImports import *
2from collections import deque
3# endregion
4
5class EnduringMomentuminStocks(QCAlgorithm):
6
7def Initialize(self):
8self.SetStartDate(2000, 1, 1)
9self.SetCash(100000)
10
11self.market_symbol:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
12
13self.leverage:int = 5
14
15self.fundamental_count:int = 500
16self.fundamental_sorting_key = lambda x: x.DollarVolume
17self.min_share_price:float = 5.
18
19self.data:Dict[Symbol, RollingWindow] = {}
20self.momentum_period:int = 21
21self.quantile:int = 10
22self.monthly_period:int = 60             # “enduring” probability period
23self.value_weighted:bool = False         # True - value weighted; False - equally weighted
24self.consecutive_occurance_period:int = 6
25self.traded_symbol_count:int = 10
26self.skip_monthly_period:int = 6
27
28# trenching
29self.managed_queue:List[RebalanceQueueItem] = []
30self.holding_period:int = 6             # months
31
32self.monthly_winners:deque[List[Symbol]] = deque(maxlen=self.monthly_period)
33self.monthly_losers:deque[List[Symbol]] = deque(maxlen=self.monthly_period)
34
35self.required_exchanges:List[str] = ['NYS', 'NAS', 'ASE']
36
37self.selection_flag:bool = False
38self.UniverseSettings.Resolution = Resolution.Daily
39self.AddUniverse(self.FundamentalSelectionFunction)
40self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
41self.Schedule.On(self.DateRules.MonthStart(self.market_symbol), self.TimeRules.BeforeMarketClose(self.market_symbol), self.Selection)
42
43def OnSecuritiesChanged(self, changes:SecurityChanges) -> None:
44for security in changes.AddedSecurities:
45    security.SetFeeModel(CustomFeeModel())
46    security.SetLeverage(self.leverage)
47
48def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
49# update the rolling window every day
50for stock in fundamental:
51    symbol = stock.Symbol
52
53    # store daily price
54    if symbol in self.data:
55        self.data[symbol].Add(stock.AdjustedPrice)
56
57if not self.selection_flag:
58    return Universe.Unchanged
59
60selected:List[Fundamental] = [x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and \
61    x.AdjustedPrice >= self.min_share_price and x.SecurityReference.ExchangeId in self.required_exchanges]
62
63if len(selected) > self.fundamental_count:
64    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
65
66# warmup price rolling windows
67for stock in selected:
68    symbol:Symbol = stock.Symbol
69
70    if symbol in self.data:
71        continue
72    
73    self.data[symbol] = RollingWindow[float](self.momentum_period)
74
75    history = self.History(symbol, self.momentum_period, Resolution.Daily)
76    if history.empty:
77        self.Log(f"Not enough data for {symbol} yet.")
78        continue
79    closes = history.loc[symbol].close
80    for time, close in closes.items():
81        self.data[symbol].Add(close)
82    
83monthly_perf:Dict[Fundamental, float] = {
84    stock: self.data[stock.Symbol][0] / self.data[stock.Symbol][self.momentum_period-1] - 1 for stock in selected if self.data[stock.Symbol].IsReady
85}
86
87losers:List[Symbol] = []
88winners:List[Symbol] = []
89market_cap:Dict[Symbol, float] = {}
90
91# sort by monthly performance
92if len(monthly_perf) >= self.quantile:
93    sorted_by_perf:List = sorted(monthly_perf.items(), key=lambda x: x[1], reverse=True)
94    quantile:int = int(len(monthly_perf) / self.quantile)
95    winners = [x[0].Symbol for x in sorted_by_perf[:quantile]]
96    losers = [x[0].Symbol for x in sorted_by_perf[-quantile:]]
97
98    market_cap = { stock.Symbol : stock.MarketCap for stock in monthly_perf }
99
100# append this month's winners and losers
101self.monthly_winners.append(winners)
102self.monthly_losers.append(losers)
103
104long:List[Symbol] = []
105short:List[Symbol] = []
106
107if len(self.monthly_winners) == self.monthly_period and len(self.monthly_losers) == self.monthly_period:
108    # get relevant winners and losers groups over the time from previous sixty to six months.
109    relevant_monthly_winners_group:List[Symbol] = list(self.monthly_winners)[:(len(self.monthly_winners) - self.skip_monthly_period)]
110    relevant_monthly_losers_group:List[Symbol] = list(self.monthly_losers)[:(len(self.monthly_losers) - self.skip_monthly_period)]
111
112    # calculate the “enduring” probability
113    winners_enduring_prob:Dict[Symbol, float] = {}
114    losers_enduring_prob:Dict[Symbol, float] = {}
115
116    for symbol in winners:
117        n_of_occurance:float = 0.
118        endurance_count:float = 0.
119
120        for i, symbol_list in enumerate(relevant_monthly_winners_group):
121            if symbol in symbol_list:
122                n_of_occurance += 1.
123                
124                # look at forward groups
125                lookahead_period_index:int = i + self.consecutive_occurance_period + 1 if i + self.consecutive_occurance_period + 1 < len(relevant_monthly_winners_group) else -1
126
127                if lookahead_period_index != -1:    # is valid index
128                    if all((symbol in group) for group in relevant_monthly_winners_group[i+1 : lookahead_period_index]):
129                        endurance_count += 1.
130
131        if n_of_occurance != 0.:
132            winners_enduring_prob[symbol] = endurance_count / n_of_occurance
133
134    for symbol in losers:
135        n_of_occurance:float = 0.
136        endurance_count:float = 0.
137
138        for i, symbol_list in enumerate(relevant_monthly_losers_group):
139            if symbol in symbol_list:
140                n_of_occurance += 1.
141                
142                # look at forward groups
143                lookahead_period_index:int = i + self.consecutive_occurance_period + 1 if i + self.consecutive_occurance_period + 1 < len(relevant_monthly_losers_group) else -1
144
145                if lookahead_period_index != -1:    # is valid index
146                    if all((symbol in group) for group in relevant_monthly_losers_group[i+1 : lookahead_period_index]):
147                        endurance_count += 1.
148
149        if n_of_occurance != 0.:
150            losers_enduring_prob[symbol] = endurance_count / n_of_occurance
151    
152    # go long the top ten past winners and go short the top ten past losers with the highest estimated enduring probability and hold this position for six months
153    if len(winners_enduring_prob) >= self.traded_symbol_count: 
154        winners_sorted_by_prob:List = sorted(winners_enduring_prob.items(), key=lambda x: x[1], reverse=True)
155        long = [x[0] for x in winners_sorted_by_prob[:self.traded_symbol_count]]
156        
157    if len(losers_enduring_prob) >= self.traded_symbol_count:
158        losers_sorted_by_prob:List = sorted(losers_enduring_prob.items(), key=lambda x: x[1], reverse=True)
159        short = [x[0] for x in losers_sorted_by_prob[:self.traded_symbol_count]]
160
161if long and short:
162    # calculate quantities for long and short trenche
163    if self.value_weighted:
164        total_market_cap_long:float = sum([market_cap[x] for x in long])
165        total_market_cap_short:float = sum([market_cap[x] for x in short])
166        
167        long_w:float = self.Portfolio.TotalPortfolioValue / self.holding_period
168        short_w:float = self.Portfolio.TotalPortfolioValue / self.holding_period
169        
170        long_symbol_q:List[Tuple[Symbol, float]] = [(x, np.floor(long_w * (market_cap[x] / total_market_cap_long) / self.data[x][0])) for x in long]
171        short_symbol_q:List[Tuple[Symbol, float]] = [(x, -np.floor(short_w * (market_cap[x] / total_market_cap_short) / self.data[x][0])) for x in short]
172        
173        self.managed_queue.append(RebalanceQueueItem(long_symbol_q + short_symbol_q))
174    else:
175        long_w:float = self.Portfolio.TotalPortfolioValue / self.holding_period / len(long)
176        short_w:float = self.Portfolio.TotalPortfolioValue / self.holding_period / len(short)
177        
178        long_symbol_q:List[Tuple[Symbol, float]] = [(x, np.floor(long_w / self.data[x][0])) for x in long]
179        short_symbol_q:List[Tuple[Symbol, float]] = [(x, -np.floor(short_w / self.data[x][0])) for x in short]
180        
181        self.managed_queue.append(RebalanceQueueItem(long_symbol_q + short_symbol_q))
182
183return long + short
184
185def OnData(self, data: Slice) -> None:
186if not self.selection_flag:
187    return
188self.selection_flag = False
189
190# trade execution - rebalance portfolio
191remove_item:Union[RebalanceQueueItem, None] = None
192
193for item in self.managed_queue:
194    # liquidate
195    if item.holding_period == self.holding_period: # all portfolio parts are held for n months
196        for symbol, quantity in item.opened_symbol_q:
197            self.MarketOrder(symbol, -quantity)
198                    
199        remove_item = item
200    
201    # trade execution    
202    if item.holding_period == 0: # all portfolio parts are held for n months
203        opened_symbol_q:List[Tuple[Symbol, float]] = []
204        
205        for symbol, quantity in item.opened_symbol_q:
206            if symbol in data and data[symbol] and self.Securities[symbol].IsTradable:
207                self.MarketOrder(symbol, quantity)
208                opened_symbol_q.append((symbol, quantity))
209                    
210        # only opened orders will be closed        
211        item.opened_symbol_q = opened_symbol_q
212        
213    item.holding_period += 1
214    
215# need to remove closed part of portfolio after loop. Otherwise it will miss one item in self.managed_queue
216if remove_item:
217    self.managed_queue.remove(remove_item)
218
219def Selection(self) -> None:
220self.selection_flag = True
221
222class RebalanceQueueItem():
223def __init__(self, symbol_q:List):
224# symbol/quantity collections
225self.opened_symbol_q:List[Tuple[Symbol, float]] = symbol_q  
226self.holding_period:int = 0
227
228class CustomFeeModel(FeeModel):
229def GetOrderFee(self, parameters):
230fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
231return OrderFee(CashAmount(fee, "USD"))