Overall Statistics |
Total Trades
1442
Average Win
1.11%
Average Loss
-1.16%
Compounding Annual Return
10.168%
Drawdown
62.700%
Expectancy
0.264
Net Profit
665.358%
Sharpe Ratio
0.469
Probabilistic Sharpe Ratio
0.254%
Loss Rate
35%
Win Rate
65%
Profit-Loss Ratio
0.96
Alpha
0.125
Beta
-0.095
Annual Standard Deviation
0.252
Annual Variance
0.063
Information Ratio
0.151
Tracking Error
0.317
Treynor Ratio
-1.244
Total Fees
$2950.17
|
# https://quantpedia.com/strategies/momentum-factor-effect-in-reits/ # # The investment universe consists of all US REITs listed on markets. Every month, the investor ranks all available REITs # by their past 11-month return one-month lagged and groups them into equally weighted tercile portfolios. He/she then goes # long on the best performing tercile for three months. One-third of the portfolio is rebalanced this way monthly, and REITs # are equally weighted. This is not the only way to capture the momentum factor in REITs as a consequential portfolio could be # formed as a long/short or from quartiles/quintiles/deciles instead of terciles or based on different formation and holding # periods (additional types of this strategy are stated in the “Other papers” section). # # QC implementation changes: # - Instead of all listed stock, we select 500 most liquid stocks from QC filtered stock universe (~8000 stocks) due to time complexity issues tied to whole universe filtering. from collections import deque class MomentumFactorEffectinREITs(QCAlgorithm): def Initialize(self): self.SetStartDate(2000, 1, 1) self.SetCash(100000) self.symbol = self.AddEquity('SPY', Resolution.Daily).Symbol # EW Trenching. self.holding_period = 3 self.managed_queue = deque(maxlen = self.holding_period + 1) self.data = {} self.period = 12 * 21 self.coarse_count = 500 self.selection_flag = False self.UniverseSettings.Resolution = Resolution.Daily self.AddUniverse(self.CoarseSelectionFunction,self.FineSelectionFunction) self.Schedule.On(self.DateRules.MonthEnd(self.symbol), self.TimeRules.BeforeMarketClose(self.symbol), self.Selection) def OnSecuritiesChanged(self, changes): for security in changes.AddedSecurities: security.SetFeeModel(CustomFeeModel(self)) security.SetLeverage(5) def CoarseSelectionFunction(self, coarse): if not self.selection_flag: return Universe.Unchanged # Update the rolling window every month. for stock in coarse: symbol = stock.Symbol # Store monthly price. if symbol in self.data: self.data[symbol].update(stock.AdjustedPrice) # selected = [x.Symbol for x in coarse if x.HasFundamentalData and x.Market == 'usa'] selected = [x.Symbol for x in sorted([x for x in coarse if x.HasFundamentalData and x.Market == 'usa'], key = lambda x: x.DollarVolume, reverse = True)[:self.coarse_count]] # Warmup price rolling windows. for symbol in selected: if symbol in self.data: continue self.data[symbol] = SymbolData(symbol, 13) history = self.History(symbol, self.period * 30, Resolution.Daily) if history.empty: self.Log(f"Not enough data for {symbol} yet.") continue closes = history.loc[symbol].close closes_len = len(closes.keys()) # Find monthly closes. for index, time_close in enumerate(closes.iteritems()): # index out of bounds check. if index + 1 < closes_len: date_month = time_close[0].date().month next_date_month = closes.keys()[index + 1].month # Found last day of month. if date_month != next_date_month: self.data[symbol].update(time_close[1]) selected = [x for x in selected if self.data[x].is_ready()] return selected def FineSelectionFunction(self, fine): fine = [x.Symbol for x in fine if (x.CompanyReference.IsREIT == 1)] momentum = {x : self.data[x].performance(1) for x in fine} long = [] short = [] if len(momentum) != 0: sorted_by_momentum = sorted(momentum.items(), key = lambda x: x[1], reverse = True) tercile = int(len(sorted_by_momentum) / 3) long = [x[0] for x in sorted_by_momentum[:tercile] if not self.IsInvested(x[0])] short = [] self.managed_queue.append(RebalanceQueueItem(long, short)) return long + short def IsInvested(self, symbol): return self.Securities.ContainsKey(symbol) and self.Portfolio[symbol].Invested def OnData(self, data): if not self.selection_flag: return self.selection_flag = False # Trade execution if len(self.managed_queue) == 0: return # Liquidate first items if queue is full. if len(self.managed_queue) == self.managed_queue.maxlen: item_to_liquidate = self.managed_queue.popleft() for symbol in item_to_liquidate.long_symbols + item_to_liquidate.short_symbols: self.Liquidate(symbol) curr_stock_set = self.managed_queue[-1] if curr_stock_set.count == 0: return weight = 1 / self.holding_period # Open new trades. for symbol in curr_stock_set.long_symbols: self.SetHoldings(symbol, weight / len(curr_stock_set.long_symbols)) for symbol in curr_stock_set.short_symbols: self.SetHoldings(symbol, -weight / len(curr_stock_set.short_symbols)) def Selection(self): self.selection_flag = True class SymbolData(): def __init__(self, symbol, period): self.Symbol = symbol self.Price = RollingWindow[float](period) def update(self, value): self.Price.Add(value) def is_ready(self) -> bool: return self.Price.IsReady # Performance, one month skipped. def performance(self, values_to_skip = 0) -> float: closes = [x for x in self.Price][values_to_skip:] return (closes[0] / closes[-1] - 1) class RebalanceQueueItem(): def __init__(self, long_symbols, short_symbols): self.long_symbols = long_symbols self.short_symbols = short_symbols self.count = len(long_symbols + short_symbols) # Custom fee model class CustomFeeModel(FeeModel): def GetOrderFee(self, parameters): fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005 return OrderFee(CashAmount(fee, "USD"))