12 Month Seasonal Reversals

1from AlgorithmImports import *
2import numpy as np
3from typing import List, Dict, Tuple
4#endregion
5class TwelveMonthSeasonalReversals(QCAlgorithm):
6def Initialize(self) -> None:
7self.SetStartDate(2000, 1, 1)
8self.SetCash(100_000)
9self.exchange_codes: List[str] = ['NYS', 'NAS', 'ASE']
10market: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
11
12# monthly prices
13self.data: Dict[Symbol, SymbolData] = {} 
14self.period: int = 5 * 12 + 1
15self.weight: Dict[Symbol, float] = {}
16self.min_share_price: int = 5
17self.quantile: int = 3
18self.leverage: int = 5
19self.fundamental_count: int = 1000
20self.fundamental_sorting_key = lambda x: x.DollarVolume
21self.selection_flag: bool = False
22self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
23self.UniverseSettings.Resolution = Resolution.Daily
24self.AddUniverse(self.FundamentalSelectionFunction)
25self.Schedule.On(self.DateRules.MonthStart(market), self.TimeRules.AfterMarketOpen(market), self.Selection)
26self.settings.daily_precise_end_time = False
27def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
28for security in changes.AddedSecurities:
29    security.SetFeeModel(CustomFeeModel())
30    security.SetLeverage(self.leverage)
31def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
32if not self.selection_flag:
33    return Universe.Unchanged
34# update the rolling window every month
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)
40selected: List[Fundamental] = [
41    x for x in fundamental if x.HasFundamentalData and x.Price > self.min_share_price and x.Market == 'usa' \
42    and x.MarketCap > 0 and x.SecurityReference.ExchangeId in self.exchange_codes
43]
44if len(selected) > self.fundamental_count:
45    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
46# warmup price rolling windows
47for stock in selected:
48    symbol: Symbol = stock.Symbol
49    if symbol in self.data:
50        continue
51    
52    self.data[symbol] = SymbolData(self.period)
53    history: DataFrame = self.History(symbol, self.period*30, Resolution.Daily)
54    if history.empty:
55        self.Log(f"Not enough data for {symbol} yet.")
56        continue
57    closes: Series = history.loc[symbol].close
58    
59    closes_len: int = len(closes.keys())
60    # find monthly closes
61    for index, time_close in enumerate(closes.items()):
62        # index out of bounds check.
63        if index + 1 < closes_len:
64            date_month: int = time_close[0].date().month
65            next_date_month: int = closes.keys()[index + 1].month
66        
67            # Found last day of month.
68            if date_month != next_date_month:
69                self.data[symbol].update(time_close[1])
70
71average_return: Dict[FindFundamental, float] = {}
72for stock in selected:
73    if not self.data[symbol].is_ready():
74        continue
75    monthly_returns: np.ndarray = self.data[stock.Symbol].monthly_returns()
76        
77    # other-calendar-month returns
78    relevant_monthly_returns:List[float] = [ret for i, ret in enumerate(monthly_returns) if i % 12 != 0 or i == 0]
79    average_return[stock] = np.average(relevant_monthly_returns)
80
81if len(average_return) < self.quantile:
82    return Universe.Unchanged
83# avg return sorting
84sorted_by_return: List[Tuple] = sorted(average_return.items(), key = lambda x: x[1], reverse = False)
85quantile: int = int(len(sorted_by_return) / self.quantile)
86short: List[Fundamental] = [x[0] for x in sorted_by_return[:quantile]]
87long: List[Fundamental] = [x[0] for x in sorted_by_return[-quantile:]]
88
89for i, portfolio in enumerate([long, short]):
90    mc_sum:float = sum(list(map(lambda stock: stock.MarketCap, portfolio)))
91    for stock in portfolio:
92        self.weight[stock.Symbol] = ((-1)**i) * stock.MarketCap / mc_sum
93return list(self.weight.keys())
94def OnData(self, data: Slice) -> None:
95if not self.selection_flag:
96    return
97self.selection_flag = False
98
99# trade execution
100portfolio: List[PortfolioTarget] = [PortfolioTarget(symbol, w) for symbol, w in self.weight.items() if symbol in data and data[symbol]]
101self.SetHoldings(portfolio, True)
102self.weight.clear()
103
104def Selection(self) -> None:
105self.selection_flag = True
106class SymbolData():
107def __init__(self, period: int) -> None:
108self._monthly_prices: RollingWindow = RollingWindow[float](period)
109    
110def update(self, price: float) -> None:
111self._monthly_prices.Add(price)
112    
113def is_ready(self) -> bool:
114return self._monthly_prices.IsReady
115
116def monthly_returns(self) -> np.ndarray:
117monthly_closes: np.ndarray = np.array([x for x in self._monthly_prices])
118return (monthly_closes[:-1] / monthly_closes[1:] - 1)
119# Custom fee model.
120class CustomFeeModel(FeeModel):
121def GetOrderFee(self, parameters: OrderFeeParameters) -> OrderFee:
122fee: float = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
123return OrderFee(CashAmount(fee, "USD"))