Intraday VIX Betas Predict Stocks Returns

1from AlgorithmImports import *
2from scipy import stats
3# endregion
4
5class IntradayVIXBetasPredictStocksReturns(QCAlgorithm):
6
7def Initialize(self):
8self.SetStartDate(2010, 1, 1)
9self.SetCash(100000)
10
11self.daily_return_data:dict[Symbol, List[float]] = {}
12
13self.vix = self.AddData(CBOE, 'VIX', Resolution.Daily).Symbol
14self.daily_return_data[self.vix] = []
15
16self.active_universe:List[Symbol] = []
17
18self.market_symbol:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
19
20self.min_return_period:int = 15
21self.quantile:int = 10
22self.leverage:int = 5
23self.min_share_price:float = 5.
24
25self.SetWarmup(self.min_return_period, Resolution.Daily)
26
27self.fundamental_count:int = 3000
28self.fundamental_sorting_key = lambda x: x.MarketCap
29
30self.selection_flag:bool = False
31self.rebalance_flag:bool = False
32self.UniverseSettings.Resolution = Resolution.Daily
33self.AddUniverse(self.FundamentalSelectionFunction)
34self.Settings.MinimumOrderMarginPortfolioPercentage = 0
35self.Schedule.On(self.DateRules.MonthStart(self.market_symbol), self.TimeRules.BeforeMarketClose(self.market_symbol), self.Selection)
36
37def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
38for security in changes.AddedSecurities:
39    security.SetFeeModel(CustomFeeModel())
40    security.SetLeverage(self.leverage)
41
42    self.daily_return_data[security.Symbol] = []
43
44def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
45if not self.selection_flag:
46    return Universe.Unchanged
47self.selection_flag = False
48
49selected:List[CoarseFundamental] = sorted([x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and \
50    x.AdjustedPrice >= self.min_share_price and x.SecurityReference.ExchangeId == "NAS"],
51    key=lambda x: x.DollarVolume, reverse=True)
52
53if len(selected) > self.fundamental_count:
54    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
55
56self.active_universe = list(map(lambda stock: stock.Symbol, selected))
57
58return self.active_universe
59
60def OnData(self, data: Slice) -> None:
61beta:dict[Symbol, float] = {}
62
63# store daily returns
64if self.vix in data and data[self.vix]:
65    vix_return:float = data[self.vix].Close / data[self.vix].Open - 1
66    self.daily_return_data[self.vix].append(vix_return)
67    
68    for symbol in self.active_universe:
69        if symbol in data and data[symbol]:
70            daily_return:float = data[symbol].Close / data[symbol].Open - 1
71            self.daily_return_data[symbol].append(daily_return)
72        
73        # calculate VIX beta if rebalance flag is set
74        if not self.IsWarmingUp and self.rebalance_flag:
75            vix_hist_len:int = int(len(self.daily_return_data[self.vix]))
76            if vix_hist_len >= self.min_return_period:
77                # both return series have equall lenghts
78                if len(self.daily_return_data[symbol]) == vix_hist_len:
79                    # calculate VIX beta
80                    slope, _, _, _, _ = stats.linregress(self.daily_return_data[self.vix], self.daily_return_data[symbol])
81                    beta[symbol] = slope
82        
83            # reset daily prices for a given month
84            self.daily_return_data[symbol].clear()
85
86if self.IsWarmingUp:
87    return
88
89if not self.rebalance_flag:
90    return
91self.rebalance_flag = False
92
93self.daily_return_data[self.vix].clear()
94
95long:List[Symbol] = []
96short:List[Symbol] = []
97
98if len(beta) >= self.quantile:
99    # sort by intraday beta
100    sorted_by_beta:List = sorted(beta.items(), key=lambda x: x[1], reverse=True)
101    quantile:int = int(len(beta) / self.quantile)
102    long = [x[0] for x in sorted_by_beta[-quantile:]]
103    short = [x[0] for x in sorted_by_beta[:quantile]]
104
105# order execution
106targets:List[PortfolioTarget] = []
107for i, portfolio in enumerate([long, short]):
108    for symbol in portfolio:
109        if symbol in data and data[symbol]:
110            targets.append(PortfolioTarget(symbol, ((-1) ** i) / len(portfolio)))
111
112self.SetHoldings(targets, True)
113
114def Selection(self) -> None:
115# quarterly universe selection
116if self.Time.month % 3 == 0:
117    self.selection_flag = True
118
119# monthly rebalance
120self.rebalance_flag = True
121
122class CustomFeeModel(FeeModel):
123def GetOrderFee(self, parameters):
124fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
125return OrderFee(CashAmount(fee, "USD"))