Absolute Momentum Effect in Stocks

1from AlgorithmImports import *
2from scipy import stats
3from pandas.core.frame import DataFrame
4class AbsoluteMomentumEffectStocks(QCAlgorithm):
5def Initialize(self):
6self.SetStartDate(2000, 1, 1)
7self.SetCash(100000)
8market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
9
10self.period:int = 13
11self.quantile:int = 5
12self.leverage:int = 5
13self.exchange_codes:List[str] = ['NYS', 'NAS', 'ASE']
14self.required_yearly_return_period:int = 10 # Minimum of years to calculate distribution from.
15
16self.data:Dict[Symbol, SymbolData] = {} # Monthly price data.
17self.weight:Dict[Symbol, float] = {}
18
19self.fundamental_count:int = 500
20self.fundamental_sorting_key = lambda x: x.DollarVolume
21self.selection_flag:bool = False
22self.UniverseSettings.Resolution = Resolution.Daily
23self.AddUniverse(self.FundamentalSelectionFunction)
24self.settings.daily_precise_end_time = False
25self.settings.minimum_order_margin_portfolio_percentage = 0.
26self.schedule.on(self.date_rules.month_start(market),
27                self.time_rules.after_market_open(market),
28                self.selection)
29
30def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
31for security in changes.AddedSecurities:
32    security.SetFeeModel(CustomFeeModel())
33    security.SetLeverage(self.leverage)
34        
35def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
36if not self.selection_flag:
37    return Universe.Unchanged
38# Update the rolling window every month.
39for stock in fundamental:
40    symbol:Symbol = stock.Symbol
41    # Store monthly price.
42    if symbol in self.data:
43        self.data[symbol].update(stock.AdjustedPrice)
44
45        # Add yearly performance.
46        if self.data[symbol].is_ready():
47            self.data[symbol].add_yearly_return(self.data[symbol].performance())
48selected:List[Fundamental] = [x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and \
49    x.SecurityReference.ExchangeId in self.exchange_codes and x.MarketCap != 0]
50if len(selected) > self.fundamental_count:
51    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
52long:List[Fundamental] = []
53short:List[Fundamental] = []
54# Warmup price rolling windows.
55for stock in selected:
56    symbol:Symbol = stock.Symbol
57    if symbol not in self.data:
58        self.data[symbol] = SymbolData(self.period, self.required_yearly_return_period)
59        history:DataFrame = self.History(symbol, self.period*30, Resolution.Daily)
60        if history.empty:
61            self.Log(f"Not enough data for {symbol} yet.")
62            continue
63        closes:pd.Series = history.loc[symbol].close
64        
65        closes_len:int = len(closes.keys())
66        # Find monthly closes.
67        for index, time_close in enumerate(closes.items()):
68            # index out of bounds check.
69            if index + 1 < closes_len:
70                date_month:int = time_close[0].date().month
71                next_date_month:int = closes.keys()[index + 1].month
72            
73                # Found last day of month.
74                if date_month != next_date_month:
75                    self.data[symbol].update(time_close[1])
76    
77    if self.data[symbol].yearly_returns_ready():
78        # Calculate distribution.
79        yearly_returns:List[float] = [x for x in self.data[symbol]._yearly_returns]
80        prev_yearly_returns:List[float] = yearly_returns[:-1]
81        yearly_ret:float = yearly_returns[-1]
82        
83        percentile:float = stats.percentileofscore(prev_yearly_returns, yearly_ret) / 100
84        if percentile >= 0.9:
85            long.append(stock)
86        elif percentile <= 0.1:
87            short.append(stock)
88
89# Market cap weighting.
90for i, portfolio in enumerate([long, short]):
91    mc_sum:float = sum(map(lambda x: x.MarketCap, portfolio))
92    for stock in portfolio:
93        self.weight[stock.Symbol] = ((-1) ** i) * stock.MarketCap / mc_sum
94
95return list(self.weight.keys())
96def OnData(self, data: Slice) -> None:
97if not self.selection_flag:
98    return
99self.selection_flag = False
100# Trade execution.
101portfolio:List[PortfolioTarget] = [
102    PortfolioTarget(symbol, w) for symbol, w in self.weight.items() if symbol in data and data[symbol]
103]
104self.SetHoldings(portfolio, True)
105self.weight.clear()
106def selection(self) -> None:
107self.selection_flag = True
108
109class SymbolData():
110def __init__(self, period: int, required_yearly_return_period: int):
111self._prices:RollingWindow = RollingWindow[float](period)
112self._yearly_returns:List[float] = []
113self._required_yearly_return_period:int = required_yearly_return_period
114
115def update(self, price: float) -> None:
116self._prices.Add(price)
117
118def add_yearly_return(self, value: float) -> None:
119self._yearly_returns.append(value)
120
121def is_ready(self) -> bool:
122return self._prices.IsReady
123
124def yearly_returns_ready(self) -> bool:
125return len(self._yearly_returns) >= self._required_yearly_return_period
126
127# Yearly performance, one month skipped.
128def performance(self) -> float:
129return (self._prices[1] / self._prices[self._prices.Count - 1] - 1)
130
131# Custom fee model.
132class CustomFeeModel(FeeModel):
133def GetOrderFee(self, parameters):
134fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
135return OrderFee(CashAmount(fee, "USD"))