Harvesting Volatility Risk Premia and Crisis Alpha via ETFs

1from AlgorithmImports import *
2from data_tools import SymbolData, EWMParamType, halflife, annual_port_vol, round_float
3import itertools
4import numpy as np
5# endregion
6class HarvestingVolatilityRiskPremiaandCrisisAlphaviaETFs(QCAlgorithm):
7def Initialize(self):
8self.SetStartDate(2010, 1, 1)
9self.SetCash(100000)
10# VRP symbols
11self.long_vol_symbol: Symbol = self.AddEquity('VIXY', Resolution.Daily).Symbol
12self.short_vol_symbol: Symbol = self.AddEquity('SVXY', Resolution.Daily).Symbol
13self.slope_symbols: List[Symbol] = [
14    self.AddData(CBOE, 'VIX', Resolution.Daily).Symbol, 
15    self.AddData(CBOE, 'VIX3M', Resolution.Daily).Symbol
16]
17self.min_model_period: int = 2 * 12 * 21
18self.slope_threshold: float = 0.
19self.VRP_vol_target: float = 0.2
20self.CTA_vol_target: float = 0.15
21self.CTA_final_vol_period: int = 252
22# VRP params
23self.center_of_mass: int = 63 
24# CTA params (Risk Adjusted Momentum, EMA Crossover, EMA Breakout)
25self.n: List[int] = [21, 63, 252]
26self.crossover_s: List[int] = [5, 10, 20]
27self.crossover_n: List[int] = [20, 40, 80]
28self.symbol_data: Dict[Union[Symbol, str], SymbolData] = {symbol : SymbolData(self.min_model_period) for symbol in [self.long_vol_symbol, self.short_vol_symbol]}
29self.symbol_data['slope'] = SymbolData(self.min_model_period)
30# CTA symbols
31self.CTA_commodities: List[str] = ['UGL', 'USO', 'USL', 'UNG', 'SOYB']   # 'JJC' (history since 2018)
32self.gross_weight_commodities: float = 0.1
33self.CTA_bonds: List[str] = ['TMF', 'TYD']
34self.gross_weight_bonds: float = 0.2
35self.CTA_symbols: List[Symbol] = [self.AddEquity(ticker, Resolution.Daily).Symbol for ticker in self.CTA_commodities + self.CTA_bonds]
36
37for symbol in self.CTA_symbols:
38    self.symbol_data[symbol] = SymbolData(self.min_model_period)
39
40[self.Securities[s].SetLeverage(3) for s in self.symbol_data if s != 'slope']
41
42self.SetWarmup(self.min_model_period, Resolution.Daily)
43self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
44def OnData(self, slice: Slice) -> None:
45# store price data
46for symbol in self.symbol_data:
47    if symbol == 'slope': continue
48    if slice.ContainsKey(symbol) and slice[symbol] is not None:
49        self.symbol_data[symbol].update(slice[symbol].Value)
50# store VIX data
51if all(slice.ContainsKey(symbol) and slice[symbol] is not None for symbol in self.slope_symbols):
52    self.symbol_data['slope'].update(slice[self.slope_symbols[1]].Value - slice[self.slope_symbols[0]].Value)
53
54if self.IsWarmingUp: return
55
56if all(self.symbol_data[symbol].is_ready() for symbol in self.symbol_data):
57    # VRP strategy
58    slope_data: SymbolData = self.symbol_data['slope']
59    # calculate two latest values of slope z-score; 
60    # index: 0-latest, 1-one day lag
61    z_score_values: List[float] = [(slope_data.get_latest_value(lag) - slope_data.value_EWMA(EWMParamType.SPAN, self.center_of_mass, lag + 1)) \
62        / slope_data.value_EWSD(EWMParamType.SPAN, self.center_of_mass, lag + 1) \
63        for lag in range(2)]
64    
65    # long-short trading signal
66    VRP_traded_symbol: Union[Symbol, None] = None
67    if slope_data.get_latest_value() > self.slope_threshold and z_score_values[0] > 0:
68        VRP_traded_symbol = self.short_vol_symbol
69    
70    elif slope_data.get_latest_value() > self.slope_threshold and z_score_values[0] < 0:
71        VRP_traded_symbol = self.long_vol_symbol
72    
73    elif (slope_data.get_latest_value(1) > self.slope_threshold and z_score_values[1] < 0) and \
74        (slope_data.get_latest_value() < self.slope_threshold and z_score_values[0] < -1):
75        VRP_traded_symbol = self.long_vol_symbol
76    
77    elif (slope_data.get_latest_value(1) < self.slope_threshold and z_score_values[1] < -1) and \
78        (slope_data.get_latest_value() < self.slope_threshold and z_score_values[0] > -1):
79        VRP_traded_symbol = self.short_vol_symbol
80    
81    if VRP_traded_symbol is not None:
82        VRP_symbol_volatility: float = self.symbol_data[VRP_traded_symbol].return_EWSD(self.center_of_mass)
83        
84        # scaled volatility
85        VRP_weight: float = round_float(self.VRP_vol_target / VRP_symbol_volatility)
86    
87        # CTA strategy
88        # 1. Risk Adjusted Momentum
89        risk_adjusted_momentum: Dict[Symbol, List[float]] = {
90            symbol : [self.symbol_data[symbol].compounded_return(lookback) / self.symbol_data[symbol].scaled_volatility(lookback) for lookback in self.n]
91            for symbol in self.CTA_symbols
92        }
93        # 2. EMA Crossover
94        EMA_crossover: Dict[Symbol, List[float]] = {
95            symbol : [(self.symbol_data[symbol].value_EWMA(EWMParamType.HALFLIFE, halflife(s)) - self.symbol_data[symbol].value_EWMA(EWMParamType.HALFLIFE, halflife(n))) / self.symbol_data[symbol].value_EWSD(EWMParamType.HALFLIFE, halflife(n)) \
96            for s, n in list(zip(self.crossover_s, self.crossover_n))]
97            for symbol in self.CTA_symbols
98        }
99        # 3. EMA Breakout
100        EMA_breakout: Dict[Symbol, List[float]] = {
101            symbol : [(self.symbol_data[symbol].get_latest_value() - self.symbol_data[symbol].value_EWMA(EWMParamType.SPAN, lookback)) / self.symbol_data[symbol].value_EWSD(EWMParamType.SPAN, lookback) for lookback in self.n]
102            for symbol in self.CTA_symbols
103        }
104        # update x_n
105        for symbol in self.CTA_symbols:
106            x_n: np.ndarray = np.array([risk_adjusted_momentum[symbol], EMA_crossover[symbol], EMA_breakout[symbol]])
107            self.symbol_data[symbol].update_x_n(x_n)
108        if all(self.symbol_data[symbol].x_n_is_ready() for symbol in self.CTA_symbols):
109            # calculate CTA weights
110            n: np.ndarray = [
111                self.n,
112                self.crossover_n,
113                self.n
114            ]
115            
116            CTA_weight: float = 1. - VRP_weight
117            
118            CTA_gross_weight: Dict[Symbol, float] = {
119                symbol : self.gross_weight_commodities if symbol.Value in self.CTA_commodities else self.gross_weight_bonds for symbol in self.CTA_symbols
120            }
121            # signal strength
122            CTA_x_n_weight: Dict[Symbol, float] = {
123                symbol : self.symbol_data[symbol].get_weight(n, 252) for symbol in self.CTA_symbols
124            }
125            
126            CTA_volatility_weight: Dict[Symbol, float] = {
127                symbol : (self.CTA_vol_target / self.symbol_data[symbol].return_EWSD(self.center_of_mass)) ** 2 for symbol in self.CTA_symbols
128            }
129            
130            CTA_asset_weight: Dict[Symbol, float] = {
131                symbol : CTA_gross_weight[symbol] * CTA_x_n_weight[symbol] * CTA_volatility_weight[symbol] for symbol in self.CTA_symbols
132            }
133            # instantaneous portfolio volatility
134            returns: np.ndarray = np.array([self.symbol_data[symbol].get_daily_returns()[-self.CTA_final_vol_period:] for symbol in self.CTA_symbols])
135            weights: np.ndarray = abs(np.array(list(CTA_x_n_weight.values())))
136            CTA_portfolio_volatility: float = annual_port_vol(returns, weights)
137            final_CTA_weight: float = round_float(self.CTA_vol_target / CTA_portfolio_volatility)
138            # concat final VRP and CTA portfolio assets
139            targets: List[PortfolioTarget] = [
140                PortfolioTarget(symbol, round_float(CTA_weight * final_CTA_weight * CTA_asset_weight[symbol])) for symbol in self.CTA_symbols
141            ] + [PortfolioTarget(VRP_traded_symbol, VRP_weight)]
142            self.SetHoldings(targets, True)