Consistent Momentum Long-Short Portfolio with Six-Month Hold

1from AlgoLib import *
2
3class ConsistentMomentumStrategy(XXX):
4
5def Initialize(self):
6self.SetStartDate(2010, 1, 1)
7self.SetCash(100000)
8
9market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
10
11self.long:List[Symbol] = []
12self.short:List[Symbol] = []
13
14self.data:Dict[Symbol, SymbolData] = {}
15
16self.fundamental_count:int = 1000
17self.fundamental_sorting_key = lambda x: x.DollarVolume
18
19self.period:int = 7 * 21
20self.quantile:int = 10
21self.leverage:int = 5
22self.exchange_codes:List[str] = ['NYS', 'NAS', 'ASE']
23
24self.selection_flag:bool = False
25self.UniverseSettings.Resolution = Resolution.Daily
26self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
27self.AddUniverse(self.FundamentalSelectionFunction)
28self.Schedule.On(self.DateRules.MonthStart(market), self.TimeRules.AfterMarketOpen(market), self.Rebalance)
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]:
36# Update the rolling window every day.
37for stock in fundamental:
38    symbol:Symbol = stock.Symbol
39
40    # Store monthly price.
41    if symbol in self.data:
42        self.data[symbol].update(stock.AdjustedPrice)
43
44if not self.selection_flag:
45    return Universe.Unchanged
46
47selected:List[Fundamental] = [x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and x.SecurityReference.ExchangeId in self.exchange_codes and \
48    x.MarketCap != 0 and x.CompanyReference.IsREIT != 1]
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
52momentum_t71_t60:Dict[Symbol, float] = {}
53
54# Warmup price rolling windows.
55for stock in selected:
56    symbol:Symbol = stock.Symbol
57
58    if symbol not in self.data:
59        self.data[symbol] = SymbolData(self.period)
60        history:DataFrame = self.History(symbol, self.period, Resolution.Daily)
61        if history.empty:
62            self.Log(f"Not enough data for {symbol} yet")
63            continue
64        closes:pd.Series = history.loc[symbol].close
65        for time, close in closes.items():
66            self.data[symbol].update(close)
67        
68    if self.data[symbol].is_ready():
69        momentum_t71_t60[symbol] = (self.data[symbol].performance_t7t1(), self.data[symbol].performance_t6t0())
70
71if len(momentum_t71_t60) >= self.quantile:
72    # Momentum t-7 to t-1 sorting
73    sorted_by_perf_t71:List = sorted(momentum_t71_t60.items(), key = lambda x: x[1][0], reverse = True)
74    quantile:int = int(len(sorted_by_perf_t71) / self.quantile)
75    high_by_perf_t71:List[Symbol] = [x[0] for x in sorted_by_perf_t71[:quantile]]
76    low_by_perf_t71:List[Symbol] = [x[0] for x in sorted_by_perf_t71[-quantile:]]
77
78    # Momentum t-6 to t sorting
79    sorted_by_perf_t60:List = sorted(momentum_t71_t60.items(), key = lambda x: x[1][1], reverse = True)
80    quantile = int(len(sorted_by_perf_t60) / self.quantile)
81    high_by_perf_t60:List[Symbol] = [x[0] for x in sorted_by_perf_t60[:quantile]]
82    low_by_perf_t60:List[Symbol] = [x[0] for x in sorted_by_perf_t60[-quantile:]]
83    
84    self.long = [x for x in high_by_perf_t71 if x in high_by_perf_t60]
85    self.short = [x for x in low_by_perf_t71 if x in low_by_perf_t60]
86
87return self.long + self.short
88
89def OnData(self, data: Slice) -> None:
90if not self.selection_flag:
91    return
92self.selection_flag = False
93
94# order execution
95targets:List[PortfolioTarget] = []
96for i, portfolio in enumerate([self.long, self.short]):
97    for symbol in portfolio:
98        if symbol in data and data[symbol]:
99            targets.append(PortfolioTarget(symbol, ((-1) ** i) / len(portfolio)))
100		
101self.SetHoldings(targets, True)
102
103self.long.clear()
104self.short.clear()
105
106def Rebalance(self) -> None:
107if self.Time.month % 6 == 0:
108    self.selection_flag = True
109
110class SymbolData():
111def __init__(self, period: int):
112self._price:RollingWindow = RollingWindow[float](period)
113
114def update(self, price: float) -> None:
115self._price.Add(price)
116
117def is_ready(self) -> bool:
118return self._price.IsReady
119
120def performance_t7t1(self) -> float:
121closes:List[float] = [x for x in self._price][21:]
122return (closes[0] / closes[-1] - 1)
123
124def performance_t6t0(self) -> float:
125closes:List[float] = [x for x in self._price][:-21]
126return (closes[0] / closes[-1] - 1)
127
128# Custom fee model.
129class CustomFeeModel(FeeModel):
130def GetOrderFee(self, parameters):
131fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
132return OrderFee(CashAmount(fee, "USD"))