Lagged Momentum Rotation Strategy in US REITs

1from AlgoLib import *
2
3class MomentumFactorEffectinREITs(XXX):
4
5def Initialize(self):
6self.SetStartDate(2000, 1, 1)
7self.SetCash(100000) 
8
9self.symbol:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
10
11# EW Trenching.
12self.holding_period:int = 3
13self.managed_queue:List[RebalanceQueueItem] = []
14
15self.data:Dict[Symbol, SymbolData] = {}
16self.period:int = 12 * 21
17self.quantile:int = 3
18self.leverage:int = 5
19
20self.fundamental_count:int = 500
21self.fundamental_sorting_key = lambda x: x.DollarVolume
22
23self.selection_flag = False
24self.UniverseSettings.Resolution = Resolution.Daily
25self.AddUniverse(self.FundamentalSelectionFunction)
26self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
27self.Schedule.On(self.DateRules.MonthEnd(self.symbol), self.TimeRules.BeforeMarketClose(self.symbol), self.Selection)
28
29def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
30for security in changes.AddedSecurities:
31    security.SetFeeModel(CustomFeeModel())
32    security.SetLeverage(self.leverage)
33    
34def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
35if not self.selection_flag:
36    return Universe.Unchanged
37
38# Update the rolling window every month.
39for stock in fundamental:
40    symbol:Symbol = stock.Symbol
41
42    # Store monthly price.
43    if symbol in self.data:
44        self.data[symbol].update(stock.AdjustedPrice)
45
46selected:List[Fundamental] = [x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and x.CompanyReference.IsREIT == 1]
47if len(selected) > self.fundamental_count:
48    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
49
50momentum:Dict[Symbol, float] = {}
51
52# Warmup price rolling windows.
53for stock in selected:
54    symbol:Symbol = stock.Symbol
55
56    if symbol not in self.data:
57        self.data[symbol] = SymbolData(symbol, 13)
58        history = self.History(symbol, self.period * 30, Resolution.Daily)
59        if history.empty:
60            self.Log(f"Not enough data for {symbol} yet.")
61            continue
62        closes = history.loc[symbol].close
63        
64        closes_len = len(closes.keys())
65        # Find monthly closes.
66        for index, time_close in enumerate(closes.items()):
67            # index out of bounds check.
68            if index + 1 < closes_len:
69                date_month = time_close[0].date().month
70                next_date_month = closes.keys()[index + 1].month
71            
72                # Found last day of month.
73                if date_month != next_date_month:
74                    self.data[symbol].update(time_close[1])
75    
76    if self.data[symbol].is_ready():
77        momentum[symbol] = self.data[symbol].performance(1)
78
79long:List[Symbol] = []
80
81if len(momentum) >= self.quantile:
82    sorted_by_momentum:List = sorted(momentum.items(), key = lambda x: x[1], reverse = True)
83    quantile:int = int(len(sorted_by_momentum) / self.quantile)
84    long = [x[0] for x in sorted_by_momentum[:quantile]]
85    
86    weight:float = self.Portfolio.TotalPortfolioValue / self.holding_period / len(long)
87    long_symbol_q:List = [(symbol, np.floor(weight / self.data[symbol].get_recent_price())) for symbol in long]
88    
89    self.managed_queue.append(RebalanceQueueItem(long_symbol_q))
90
91return long
92
93def OnData(self, data: Slice) -> None:
94if not self.selection_flag:
95    return
96self.selection_flag = False
97
98# rebalance portfolio
99remove_item:Union[RebalanceQueueItem, None] = None
100
101for item in self.managed_queue:
102    if item.holding_period == self.holding_period: # all portfolio parts are held for n months
103        for symbol, quantity in item.opened_symbol_q:
104            self.MarketOrder(symbol, -quantity)
105                    
106        remove_item = item
107    
108    # trade execution    
109    if item.holding_period == 0: # all portfolio parts are held for n months
110        opened_symbol_q = []
111        
112        for symbol, quantity in item.opened_symbol_q:
113            if symbol in data and data[symbol]:
114                self.MarketOrder(symbol, quantity)
115                opened_symbol_q.append((symbol, quantity))
116                    
117        # only opened orders will be closed        
118        item.opened_symbol_q = opened_symbol_q
119        
120    item.holding_period += 1
121    
122# need to remove closed part of portfolio after loop. Otherwise it will miss one item in self.managed_queue
123if remove_item:
124    self.managed_queue.remove(remove_item)
125
126def Selection(self) -> None:
127self.selection_flag = True
128
129class SymbolData():
130def __init__(self, symbol: Symbol, period: int):
131self._symbol:Symbol = symbol
132self._prices:RollingWindow = RollingWindow[float](period)
133
134def update(self, value: float) -> None:
135self._prices.Add(value)
136
137def is_ready(self) -> bool:
138return self._prices.IsReady
139
140def get_recent_price(self) -> float:
141return self._prices[0]
142
143# Performance, one month skipped.
144def performance(self, values_to_skip = 0) -> float:
145closes = [x for x in self._prices][values_to_skip:]
146return (closes[0] / closes[-1] - 1)
147
148class RebalanceQueueItem():
149def __init__(self, symbol_q):
150# symbol/quantity collections
151self.opened_symbol_q = symbol_q  
152self.holding_period = 0
153
154# Custom fee model
155class CustomFeeModel(FeeModel):
156def GetOrderFee(self, parameters):
157fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
158return OrderFee(CashAmount(fee, "USD"))