Momentum Seasonality and Investor Preferences

1from AlgorithmImports import *
2from pandas.core.frame import DataFrame
3class MomentumSeasonalityInvestorPreferences(QCAlgorithm):
4
5def Initialize(self):
6self.SetStartDate(2010, 1, 1)
7self.SetCash(100000)
8market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
9
10self.fundamental_count:int = 500
11self.fundamental_sorting_key = lambda x: x.DollarVolume
12
13self.period:int = 12 * 21
14self.quantile:int = 10
15self.leverage:int = 5
16self.exchange_codes:List[str] = ['NYS', 'NAS', 'ASE']
17self.long:List[Symbol] = []
18self.short:List[Symbol] = []
19
20# Daily price data.
21self.data:Dict[Symbol, SymbolData] = {}
22
23self.selection_flag:bool = True
24self.UniverseSettings.Resolution = Resolution.Daily
25self.AddUniverse(self.FundamentalSelectionFunction)
26self.settings.daily_precise_end_time = False
27self.settings.minimum_order_margin_portfolio_percentage = 0.
28self.schedule.on(self.date_rules.month_start(market),
29                self.time_rules.after_market_open(market),
30                self.selection)
31def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
32for security in changes.AddedSecurities:
33    security.SetFeeModel(CustomFeeModel())
34    security.SetLeverage(self.leverage)
35				
36def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
37# Update the rolling window every day.
38for stock in fundamental:
39    symbol:Symbol = stock.Symbol
40    # Store monthly price.
41    if symbol in self.data:
42        self.data[symbol].update(stock.AdjustedPrice)
43if not self.selection_flag:
44    return Universe.Unchanged
45selected:List[Fundamental] = [x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and x.SecurityReference.ExchangeId in self.exchange_codes and x.MarketCap != 0]
46if len(selected) > self.fundamental_count:
47    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
48    
49performance:Dict[Symbol, float] = {}
50# Warmup price rolling windows.
51for stock in selected:
52    symbol:Symbol = stock.Symbol
53    if symbol not in self.data:
54        self.data[symbol] = SymbolData(self.period)
55        history:DataFrame = self.History(symbol, self.period, Resolution.Daily)
56        if history.empty:
57            self.Log(f"Not enough data for {symbol} yet")
58            continue
59        closes:pd.Series = history.loc[symbol].close
60        for time, close in closes.items():
61            self.data[symbol].update(close)
62        
63    if self.data[symbol].is_ready():
64        performance[symbol] = self.data[symbol].performance()
65
66if len(performance) >= self.quantile:
67    sorted_by_performance:List = sorted(performance, key = performance.get, reverse = True)
68    quantile:int = int(len(sorted_by_performance) / self.quantile)
69    self.long = sorted_by_performance[:quantile]
70    self.short = sorted_by_performance[-quantile:]
71
72return self.long + self.short
73def OnData(self, data: Slice) -> None:
74if not self.selection_flag:
75    return
76self.selection_flag = False
77
78# order execution
79targets:List[PortfolioTarget] = []
80for i, portfolio in enumerate([self.long, self.short]):
81    for symbol in portfolio:
82        if symbol in data and data[symbol]:
83            targets.append(PortfolioTarget(symbol, ((-1) ** i) / len(portfolio)))
84		
85self.SetHoldings(targets, True)
86self.long.clear()
87self.short.clear()
88def selection(self) -> None:
89if self.Time.month % 3 == 0:
90    self.Liquidate()
91    self.selection_flag = True
92class SymbolData():
93def __init__(self, period: int):
94self._price:RollingWindow = RollingWindow[float](period)
95
96def update(self, price: float) -> None:
97self._price.Add(price)
98
99def is_ready(self) -> bool:
100return self._price.IsReady
101
102# Yearly performance, one month skipped.
103def performance(self) -> float:
104closes:List[float] = list(self._price)[21:]
105return (closes[0] / closes[-1] - 1)
106
107# Custom fee model.
108class CustomFeeModel(FeeModel):
109def GetOrderFee(self, parameters):
110fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
111return OrderFee(CashAmount(fee, "USD"))