Timing High and Low Volatility Equity Factor Strategy

1import numpy as np
2from AlgorithmImports import *
3from pandas.core.frame import DataFrame
4class TimingHighLowVolatility(QCAlgorithm):
5def Initialize(self):
6self.SetStartDate(2000, 1, 1)
7self.SetCash(100000)
8self.period:int = 21
9self.quantile:int = 10
10self.slope_std_threshold:float = 4.
11self.long:List[Symbol] = []
12self.slopes:RollingWindow = RollingWindow[float](12)
13self.data:Dict[Symbol, RollingWindow] = {}
14
15market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
16self.fundamental_count:int = 500
17self.fundamental_sorting_key = lambda x: x.DollarVolume
18
19self.selection_flag:bool = False
20self.UniverseSettings.Resolution = Resolution.Daily
21self.AddUniverse(self.FundamentalSelectionFunction)
22self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
23self.Schedule.On(self.DateRules.MonthStart(market), self.TimeRules.AfterMarketOpen(market), self.Selection)
24self.settings.daily_precise_end_time = False
25def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
26for security in changes.AddedSecurities:
27    security.SetFeeModel(CustomFeeModel())
28        
29def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
30# Update the rolling window every day.
31for stock in fundamental:
32    symbol:Symbol = stock.Symbol
33    # Store monthly price.
34    if symbol in self.data:
35        self.data[symbol].Add(stock.AdjustedPrice)
36if not self.selection_flag:
37    return Universe.Unchanged
38selected:List[Fundamental] = [x for x in fundamental if x.HasFundamentalData and x.Market == 'usa']
39if len(selected) > self.fundamental_count:
40    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
41
42volatility:Dict[Symbol, float] = {}
43momentum:Dict[Symbol, float] = {}
44# Warmup price rolling windows.
45for stock in selected:
46    symbol:Symbol = stock.Symbol
47    
48    if symbol not in self.data:
49        self.data[symbol] = RollingWindow[float](self.period)
50        history:DataFrame = self.History(symbol, self.period, Resolution.Daily)
51        if history.empty:
52            self.Log(f"Not enough data for {symbol} yet.")
53            continue
54        closes:pd.Series = history.loc[symbol].close
55        for time, close in closes.items():
56            self.data[symbol].Add(close)
57    
58    if self.data[symbol].IsReady:
59        closes:np.ndarray = np.array(list(self.data[symbol]))
60        momentum[symbol] = closes[0] / closes[-1] - 1
61        
62        returns:np.ndarray = closes[:-1] / closes[1:] - 1
63        volatility[symbol] = np.std(returns)
64if len(momentum) >= self.quantile:
65    # Volaility sorting
66    sorted_by_vol:List[Symbol] = sorted(volatility, key = volatility.get, reverse = True)
67    quantile:int = int(len(sorted_by_vol) / self.quantile)
68    high_by_vol:List[Symbol] = sorted_by_vol[:quantile]
69    low_by_vol:List[Symbol] = sorted_by_vol[-quantile:]
70    mid_by_vol:List[Symbol] = [x for x in sorted_by_vol if x not in high_by_vol and x not in low_by_vol]
71
72    # Slope calc
73    high_vol_return:float = sum([momentum[x] for x in high_by_vol if x in momentum])
74    low_vol_return:float = sum([momentum[x] for x in low_by_vol if x in momentum])
75    slope:float = high_vol_return - low_vol_return
76
77    if self.slopes.IsReady: 
78        slopes:List[float] = list(self.slopes)
79        slopes_mean:float = np.mean(slopes)
80        slopes_std:float = np.std(slopes)
81        if slope > slopes_mean + self.slope_std_threshold * slopes_std:
82            self.long = high_by_vol
83        elif slope < slopes_mean - self.slope_std_threshold * slopes_std:
84            self.long = low_by_vol
85        else:
86            self.long = mid_by_vol
87            
88    self.slopes.Add(slope)
89
90return self.long
91
92def OnData(self, data: Slice) -> None:
93if not self.selection_flag:
94    return
95self.selection_flag = False
96
97# order execution
98portfolio:List[PortfolioTarget] = [PortfolioTarget(symbol, 1. / len(self.long)) for symbol in self.long if symbol in data and data[symbol]]
99self.SetHoldings(portfolio, True)
100self.long.clear()
101
102def Selection(self) -> None:
103self.selection_flag = True
104
105# Custom fee model
106class CustomFeeModel(FeeModel):
107def GetOrderFee(self, parameters):
108fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
109return OrderFee(CashAmount(fee, "USD"))