The Impact of Turnovers on Short-Term Momentum and Reversal

1from AlgorithmImports import *
2from typing import List, Dict, Tuple
3from numpy import isnan
4class TheImpactTurnoversShortTermMomentumReversal(QCAlgorithm):
5def Initialize(self):
6self.SetStartDate(2000, 1, 1)  
7self.SetCash(100000) 
8self.exchange_codes:List[str] = ['NYS', 'NAS', 'ASE']
9market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
10
11self.leverage:int = 10
12self.quantile:int = 5
13self.share_min_price:int = 5
14self.period:int = 21
15self.fundamental_count:int = 500
16self.fundamental_sorting_key = lambda x: x.MarketCap
17
18# Price and volume daily data.
19self.data:Dict[Symbol, SymbolData] = {}
20self.weight:Dict[Symbol, float] = {}
21
22self.selection_flag:bool = False
23self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
24self.UniverseSettings.Resolution = Resolution.Daily
25self.AddUniverse(self.FundamentalSelectionFunction)
26self.Schedule.On(self.DateRules.MonthEnd(market), self.TimeRules.AfterMarketOpen(market), self.Selection)
27self.settings.daily_precise_end_time = False
28def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
29for security in changes.AddedSecurities:
30    security.SetFeeModel(CustomFeeModel())
31    security.SetLeverage(self.leverage)
32    
33def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
34# Update the rolling window every day.
35for stock in fundamental:
36    symbol:Symbol = stock.Symbol
37    # Store monthly price.
38    if symbol in self.data:
39        self.data[symbol].update(stock.AdjustedPrice, stock.Volume)
40if not self.selection_flag:
41    return Universe.Unchanged
42# selected = [x.Symbol for x in fundamental if x.HasFundamentalData and x.Market == 'usa']
43selected:List[Fundamental] = [
44    x for x in fundamental if x.HasFundamentalData and x.Price > self.share_min_price and x.Market == 'usa' and x.MarketCap != 0 \
45    and not isnan(x.EarningReports.BasicAverageShares.ThreeMonths) and x.EarningReports.BasicAverageShares.ThreeMonths > 0 \
46    and x.SecurityReference.ExchangeId in self.exchange_codes
47]
48    
49if len(selected) > self.fundamental_count:
50    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
51# Warmup price rolling windows.
52for stock in selected:
53    symbol:Symbol = stock.Symbol
54    if symbol in self.data:
55        continue
56    self.data[symbol] = SymbolData(self.period)
57    history = self.History(symbol, self.period, Resolution.Daily)
58    if history.empty:
59        self.Log(f"Not enough data for {symbol} yet")
60        continue
61    if 'close' in history and 'volume' in history:
62        closes:DataFrame = history.loc[symbol].close
63        volumes:Series = history.loc[symbol].volume
64        for (_, close), (_, volume) in zip(closes.items(), volumes.items()):
65            self.data[symbol].update(close, volume)
66performance_turnover_market_cap:Dict[Symbol, Tuple[float]] = {}
67
68for stock in selected:
69    symbol:Symbol = stock.Symbol
70    if not self.data[symbol].is_ready():
71        continue
72    # Return calc.
73    perf:float = self.data[symbol].performance()
74    
75    # Turnover calc.
76    monthly_volume:float = self.data[symbol].monthly_volume()
77    shares_outstanding:float = stock.EarningReports.BasicAverageShares.ThreeMonths
78    turnover:float = monthly_volume / shares_outstanding
79    # Market cap calc.
80    market_cap:float = stock.MarketCap
81        
82    performance_turnover_market_cap[symbol] = (perf, turnover, market_cap)
83        
84if len(performance_turnover_market_cap) != 0:
85    # Return sorting.
86    sorted_by_ret:List[Tuple[Symbol, Tuple[float]]] = sorted(performance_turnover_market_cap.items(), key = lambda x: x[1][0], reverse = True)
87    quintile:int = int(len(sorted_by_ret) / self.quantile)
88    
89    high_ret:List[Tuple[Symbol, Tuple[float]]] = [x for x in sorted_by_ret[:quintile]]
90    low_ret:List[Tuple[Symbol, Tuple[float]]] = [x for x in sorted_by_ret[-quintile:]]
91    
92    # Turnover sorting.
93    sorted_by_turnover:List[Tuple[Symbol, Tuple[float]]] = sorted(performance_turnover_market_cap.items(), key = lambda x: x[1][1], reverse = True)
94    high_turnover:List[Tuple[Symbol, Tuple[float]]] = [x for x in sorted_by_turnover[:quintile]]
95    low_turnover:List[Tuple[Symbol, Tuple[float]]] = [x for x in sorted_by_turnover[-quintile:]]
96    
97    # Forming portfolios.
98    long_first_portfolio:List[Tuple[Symbol, Tuple[float]]] = [x for x in low_ret if x in low_turnover]
99    short_first_portfolio:List[Tuple[Symbol, Tuple[float]]] = [x for x in high_ret if x in low_turnover]
100
101    long_second_portfolio:List[Tuple[Symbol, Tuple[float]]] = [x for x in high_ret if x in high_turnover]
102    short_second_portfolio:List[Tuple[Symbol, Tuple[float]]] = [x for x in low_ret if x in high_turnover]
103    
104    # calculate weights
105    for portfolio_lst in [[long_first_portfolio, short_first_portfolio], [long_second_portfolio, short_second_portfolio]]:
106        for i, portfolio in enumerate(portfolio_lst):
107            mc_sum:float = sum(list(map(lambda x: x[1][2], portfolio)))
108            for symbol, perf_turnover_cap in portfolio:
109                self.weight[symbol] = (((-1)**i) * 0.5) * perf_turnover_cap[2] / mc_sum
110
111return [x[0] for x in self.weight.items()]
112
113def OnData(self, data: Slice):
114if not self.selection_flag:
115    return
116self.selection_flag = False
117
118# Trade execution.
119portfolio:List[PortfolioTarget] = [PortfolioTarget(symbol, w) for symbol, w in self.weight.items() if symbol in data and data[symbol]]
120self.SetHoldings(portfolio, True)
121
122self.weight.clear()
123
124def Selection(self):
125self.selection_flag = True
126
127class SymbolData():
128def __init__(self, period:int):
129self.Closes:RollingWindow = RollingWindow[float](period)
130self.Volumes:RollingWindow = RollingWindow[float](period)
131    
132def update(self, close:float, volume:float):
133self.Closes.Add(close)
134self.Volumes.Add(volume)
135    
136def is_ready(self) -> bool:
137return self.Closes.IsReady and self.Volumes.IsReady
138
139def performance(self) -> float:
140closes:List[float] = [x for x in self.Closes]
141return closes[0] / closes[-1] - 1 # Performance
142
143def monthly_volume(self) -> float:
144volumes:List[float] = [x for x in self.Volumes]
145return sum(volumes)
146
147
148class CustomFeeModel(FeeModel):
149def GetOrderFee(self, parameters):
150fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
151return OrderFee(CashAmount(fee, "USD"))