Switching Between Momentum and Reversal Strategies Based on Market Volatility

1from AlgorithmImports import *
2from pandas.core.frame import DataFrame
3from typing import List, Dict
4from dateutil.relativedelta import relativedelta
5# endregion
6
7class SwitchingBetweenMomentumandReversalStrategiesBasedonMarketVolatility(QCAlgorithm):
8
9def Initialize(self):
10self.SetStartDate(2005, 1, 1) 
11self.SetCash(100000)
12 
13self.market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
14
15self.period:int = 60
16
17self.weight:Dict[Symbol, float] = {}
18self.quantile:int = 10
19self.min_share_price:int = 5
20
21self.leverage:int = 5
22self.fundamental_count:int = 500
23self.selection_flag:bool = False
24self.UniverseSettings.Resolution = Resolution.Daily
25self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
26self.AddUniverse(self.FundamentalSelectionFunction)
27self.Schedule.On(self.DateRules.MonthStart(self.market), self.TimeRules.AfterMarketOpen(self.market), 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]:
35# monthly selection
36if not self.selection_flag:
37    return Universe.Unchanged
38
39selected:Dict[Symbol, Fundamental] = { x.Symbol : x
40    for x in sorted([x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and x.AdjustedPrice > x.AdjustedPrice >= self.min_share_price and 
41    (x.SecurityReference.ExchangeId == 'NYS') or (x.SecurityReference.ExchangeId == 'NAS') or (x.SecurityReference.ExchangeId == 'ASE') and x.MarketCap != 0], 
42    key = lambda x: x.DollarVolume, reverse = True)[:self.fundamental_count] } 
43
44# call history on assets
45history:DataFrame = self.History(list(selected.keys()) + [self.market], start=self.Time.date() - relativedelta(months=self.period), end=self.Time.date())['close'].unstack(level=0)
46market_volatility:DataFrame = history[self.market].pct_change().rolling(window='21D').std().dropna() * np.sqrt(252)
47market_volatility = market_volatility.resample('M').last()
48
49# check market volatility and set strategy
50low_vol_flag = False
51if market_volatility.iloc[-1] <= market_volatility.quantile(0.75):
52    low_vol_flag = True
53
54stocks:DataFrame = history.loc[:, history.columns != self.market]
55stocks = stocks.resample('M').last()
56stocks_performance = (stocks.iloc[-1] / stocks.iloc[-12]) - 1
57stocks_returns:Dict[Fundamental, float] = { selected[self.Symbol(asset)] : value for asset, value in stocks_performance.items() }
58
59# sort by performance of stocks
60if len(stocks_returns) >= self.quantile:
61    sorted_assets:List[Fundamental] = sorted(stocks_returns, key=stocks_returns.get, reverse=low_vol_flag)
62    quantile:int = int(len(sorted_assets) / self.quantile)
63    long:List[Fundamental] = sorted_assets[:quantile]
64    short:List[Fundamental] = sorted_assets[-quantile:]
65
66    # calculate weights based on values
67    for i, portfolio in enumerate([long, short]):
68        mc_sum:float = sum([x.MarketCap for x in portfolio])
69        for stock in portfolio:
70            self.weight[stock.Symbol] = ((-1) ** i) * (stock.MarketCap / mc_sum)
71
72return list(self.weight.keys())
73
74def OnData(self, data: Slice) -> None:
75# monthly rebalance
76if not self.selection_flag:
77    return
78self.selection_flag = False
79
80portfolio:List[PortfolioTarget] = [PortfolioTarget(symbol, w) for symbol, w in self.weight.items() if symbol in data and data[symbol]]
81self.SetHoldings(portfolio, True)
82
83self.weight.clear()
84
85def Selection(self) -> None:
86self.selection_flag = True
87
88# Custom fee model.
89class CustomFeeModel(FeeModel):
90def GetOrderFee(self, parameters):
91fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
92return OrderFee(CashAmount(fee, "USD"))