Overall Statistics |
Total Trades
30286
Average Win
0.12%
Average Loss
-0.11%
Compounding Annual Return
7.281%
Drawdown
65.300%
Expectancy
0.073
Net Profit
317.665%
Sharpe Ratio
0.386
Probabilistic Sharpe Ratio
0.039%
Loss Rate
49%
Win Rate
51%
Profit-Loss Ratio
1.12
Alpha
0.071
Beta
-0.085
Annual Standard Deviation
0.168
Annual Variance
0.028
Information Ratio
-0.028
Tracking Error
0.24
Treynor Ratio
-0.762
Total Fees
$1651.21
Estimated Strategy Capacity
$1000000000.00
Lowest Capacity Asset
GGB RIWIV7K5Z9LX
|
# https://quantpedia.com/strategies/momentum-and-reversal-combined-with-volatility-effect-in-stocks/ # # The investment universe consists of NYSE, AMEX, and NASDAQ stocks with prices higher than $5 per share. At the beginning of each month, # the sample is divided into equal halves, at the size median, and only larger stocks are used. Then each month, realized returns and realized # (annualized) volatilities are calculated for each stock for the past six months. One week (seven calendar days) prior to the beginning of # each month is skipped to avoid biases due to microstructures. Stocks are then sorted into quintiles based on their realized past returns # and past volatility. The investor goes long on stocks from the highest performing quintile from the highest volatility group and short on # stocks from the lowest-performing quintile from the highest volatility group. Stocks are equally weighted and held for six months # (therefore, 1/6 of the portfolio is rebalanced every month). # # QC implementation changes: # - Instead of all listed stock, we select 1000 most liquid stocks from QC filtered stock universe (~8000 stocks) due to time complexity issues tied to whole universe filtering. import numpy as np class MomentumReversalCombinedWithVolatilityEffectinStocks(QCAlgorithm): def Initialize(self): self.SetStartDate(2002, 1, 1) self.SetCash(100000) self.symbol = self.AddEquity('SPY', Resolution.Daily).Symbol # EW Tranching. self.holding_period = 6 self.managed_queue = [] # Daily price data. self.data = {} self.period = 6 * 21 self.coarse_count = 1000 self.selection_flag = False self.UniverseSettings.Resolution = Resolution.Daily self.AddUniverse(self.CoarseSelectionFunction, self.FineSelectionFunction) self.Schedule.On(self.DateRules.MonthStart(self.symbol), self.TimeRules.AfterMarketOpen(self.symbol), self.Selection) def OnSecuritiesChanged(self, changes): for security in changes.AddedSecurities: security.SetFeeModel(CustomFeeModel(self)) security.SetLeverage(10) def CoarseSelectionFunction(self, coarse): # Update the rolling window every day. for stock in coarse: symbol = stock.Symbol # Store monthly price. if symbol in self.data: self.data[symbol].update(stock.AdjustedPrice) self.data[symbol].LastPrice = stock.AdjustedPrice if not self.selection_flag: return Universe.Unchanged # selected = [x for x in coarse if x.HasFundamentalData and x.Market == 'usa' and x.Price > 5] selected = sorted([x for x in coarse if x.HasFundamentalData and x.Market == 'usa' and x.Price > 5], \ key = lambda x: x.DollarVolume, reverse = True)[:self.coarse_count] # Warmup price rolling windows. for stock in selected: symbol = stock.Symbol if symbol in self.data: continue self.data[symbol] = SymbolData(symbol, self.period) history = self.History(symbol, self.period, Resolution.Daily) if history.empty: self.Log(f"Not enough data for {symbol} yet.") continue closes = history.loc[symbol].close for time, close in closes.iteritems(): self.data[symbol].update(close) self.data[symbol].LastPrice = close return [x.Symbol for x in selected if self.data[x.Symbol].is_ready()] def FineSelectionFunction(self, fine): fine = [x for x in fine if x.MarketCap != 0 and \ ((x.SecurityReference.ExchangeId == "NYS") or (x.SecurityReference.ExchangeId == "NAS") or (x.SecurityReference.ExchangeId == "ASE"))] # if len(fine) > self.coarse_count: # sorted_by_market_cap = sorted(fine, key = lambda x: x.MarketCap, reverse=True) # top_by_market_cap = sorted_by_market_cap[:self.coarse_count] # else: # top_by_market_cap = fine sorted_by_market_cap = sorted(fine, key = lambda x: x.MarketCap, reverse=True) half = int(len(sorted_by_market_cap) / 2) top_by_market_cap = [x.Symbol for x in sorted_by_market_cap][:half] # Performance and volatility tuple. perf_volatility = {} for symbol in top_by_market_cap: performance = self.data[symbol].performance() annualized_volatility = self.data[symbol].volatility() perf_volatility[symbol] = (performance, annualized_volatility) long = [] short = [] if len(perf_volatility) >= 5: sorted_by_perf = sorted(perf_volatility.items(), key = lambda x: x[1][0], reverse = True) quintile = int(len(sorted_by_perf) / 5) top_by_perf = [x[0] for x in sorted_by_perf[:quintile]] low_by_perf = [x[0] for x in sorted_by_perf[-quintile:]] sorted_by_vol = sorted(perf_volatility.items(), key = lambda x: x[1][1], reverse = True) quintile = int(len(sorted_by_vol) / 5) top_by_vol = [x[0] for x in sorted_by_vol[:quintile]] low_by_vol = [x[0] for x in sorted_by_vol[-quintile:]] long = [x for x in top_by_perf if x in top_by_vol] short = [x for x in low_by_perf if x in top_by_vol] if len(long) != 0: long_w = self.Portfolio.TotalPortfolioValue / self.holding_period / len(long) # symbol/quantity collection long_symbol_q = [(x, np.ceil(long_w / self.data[x].LastPrice)) for x in long] else: long_symbol_q = [] if len(short) != 0: short_w = self.Portfolio.TotalPortfolioValue / self.holding_period / len(short) # symbol/quantity collection short_symbol_q = [(x, -np.ceil(short_w / self.data[x].LastPrice)) for x in short] else: short_symbol_q = [] self.managed_queue.append(RebalanceQueueItem(long_symbol_q + short_symbol_q)) return long + short def OnData(self, data): if not self.selection_flag: return self.selection_flag = False remove_item = None # Rebalance portfolio for item in self.managed_queue: if item.holding_period == self.holding_period: for symbol, quantity in item.symbol_q: if self.Securities[symbol].Price != 0 and self.Securities[symbol].IsTradable: self.MarketOrder(symbol, -quantity) remove_item = item # Trade execution if item.holding_period == 0: open_symbol_q = [] for symbol, quantity in item.symbol_q: if self.Securities[symbol].Price != 0 and self.Securities[symbol].IsTradable: self.MarketOrder(symbol, quantity) open_symbol_q.append((symbol, quantity)) # Only opened orders will be closed item.symbol_q = open_symbol_q item.holding_period += 1 # We need to remove closed part of portfolio after loop. Otherwise it will miss one item in self.managed_queue. if remove_item: self.managed_queue.remove(remove_item) def Selection(self): self.selection_flag = True class RebalanceQueueItem(): def __init__(self, symbol_q): # symbol/quantity collections self.symbol_q = symbol_q self.holding_period = 0 class SymbolData(): def __init__(self, symbol, period): self.Symbol = symbol self.Price = RollingWindow[float](period) self.LastPrice = 0 def update(self, value): self.Price.Add(value) def is_ready(self): return self.Price.IsReady def update(self, close): self.Price.Add(close) def volatility(self): closes = np.array([x for x in self.Price][5:]) # Skip last week. daily_returns = closes[:-1] / closes[1:] - 1 return np.std(daily_returns) * np.sqrt(252 / (len(closes))) def performance(self): closes = [x for x in self.Price][5:] # Skip last week. return (closes[0] / closes[-1] - 1) # Custom fee model. class CustomFeeModel(FeeModel): def GetOrderFee(self, parameters): fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005 return OrderFee(CashAmount(fee, "USD"))