Timing Betting-Against-Beta (BAB) Anomaly v.2

1from AlgorithmImports import *
2import numpy as np
3from pandas.core.frame import DataFrame
4from typing import List, Dict
5
6class TimingBettingAgainstBetaAnomalyv2(QCAlgorithm):
7
8def Initialize(self):
9self.SetStartDate(2000, 1, 1)
10self.SetCash(100000)
11
12# daily price data
13self.data:Dict[Symbol, RollingWindow] = {}
14self.period:int = 12 * 21
15
16self.factor_perf_d_period:int = 21
17self.factor_perf_min_period:int = 36
18self.factor_performance:List[float] = []
19self.leverage_restriction:float = 2.
20
21self.symbol:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
22self.data[self.symbol] = RollingWindow[float](self.period)
23
24self.vix:Symbol = self.AddData(CBOE, 'VIX', Resolution.Daily).Symbol
25self.data[self.vix] = RollingWindow[float](self.factor_perf_d_period)
26
27self.long:List[Symbol] = []
28self.short:List[Symbol] = []
29self.long_lvg:float = 1.   # leverage for long portfolio calculated from average beta
30self.short_lvg:float = 1.  # leverage for short portfolio calculated from average beta
31self.leverage_cap:float = 2.
32
33self.coarse_count:int = 1000
34self.quantile:int = 10
35self.min_share_price:float = 5.
36
37self.selection_flag:bool = False
38self.UniverseSettings.Resolution = Resolution.Daily
39self.AddUniverse(self.FundamentalSelectionFunction)
40self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
41self.Schedule.On(self.DateRules.MonthStart(self.symbol), self.TimeRules.AfterMarketOpen(self.symbol), self.Selection)
42
43def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
44for security in changes.AddedSecurities:
45    security.SetFeeModel(CustomFeeModel())
46    security.SetLeverage(self.leverage_cap * self.leverage_restriction * 5)
47    
48def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
49# update the rolling window every day
50for stock in fundamental:
51    symbol:Symbol = stock.Symbol
52
53    if symbol in self.data:
54        # store daily price
55        self.data[symbol].Add(stock.AdjustedPrice)
56
57# selection once a month
58if not self.selection_flag:
59    return Universe.Unchanged
60
61# store BAB performance
62if self.long and self.short:
63    long_closes:np.ndarray = np.array([np.array(list(self.data[symbol])[:self.factor_perf_d_period][::-1]) for symbol in self.long])
64    short_closes:np.ndarray = np.array([np.array(list(self.data[symbol])[:self.factor_perf_d_period][::-1]) for symbol in self.short])
65    long_returns:np.ndarray = (np.diff(long_closes) / long_closes[:,:-1])
66    short_returns:np.ndarray = (np.diff(short_closes) / short_closes[:,:-1])
67
68    factor_perf:float = np.cumproduct(1 + (np.mean(long_returns, axis=0) - np.mean(short_returns, axis=0)))[-1] - 1
69
70    self.factor_performance.append(factor_perf)
71
72selected:List[Symbol] = [x.Symbol
73    for x in sorted([x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and x.AdjustedPrice >= self.min_share_price and x.MarketCap != 0],
74        key = lambda x: x.DollarVolume, reverse = True)[:self.coarse_count]]
75
76rebalance:bool = False
77if self.data[self.symbol].IsReady:
78    rebalance = True
79
80beta:Dict[Symbol, float] = {}
81
82for symbol in selected:
83    # warmup price rolling windows
84    if symbol not in self.data:
85        self.data[symbol] = RollingWindow[float](self.period)
86        history:DataFrame = self.History(symbol, self.period, Resolution.Daily)
87        if history.empty:
88            self.Log(f"Not enough data for {symbol} yet")
89            continue
90        closes:pd.Series = history.loc[symbol].close
91        for time, close in closes.items():
92            self.data[symbol].Add(close)
93    
94    if rebalance:
95        if self.data[symbol].IsReady:
96            market_closes:np.ndarray = np.array(list(self.data[self.symbol]))
97            stock_closes:np.ndarray = np.array(list(self.data[symbol]))
98            
99            market_returns:np.ndarray = (market_closes[:-1] - market_closes[1:]) / market_closes[1:]
100            stock_returns:np.ndarray = (stock_closes[:-1] - stock_closes[1:]) / stock_closes[1:]
101            
102            cov:float = np.cov(stock_returns[::-1], market_returns[::-1])[0][1]
103            market_variance:float = np.var(market_returns)
104            beta[symbol] = cov / market_variance
105
106if len(beta) >= self.quantile:
107    # sort by beta
108    sorted_by_beta:List = sorted(beta.items(), key = lambda x:x[1], reverse=True)
109    quantile:int = int(len(sorted_by_beta) / self.quantile)
110    self.long = [x for x in sorted_by_beta[-quantile:]]
111    self.short = [x for x in sorted_by_beta[:quantile]]
112    
113    # create zero-beta portfolio
114    long_mean_beta:float = np.mean([x[1] for x in self.long])
115    short_mean_beta:float = np.mean([x[1] for x in self.short])
116    
117    self.long = [x[0] for x in self.long]
118    self.short = [x[0] for x in self.short]
119    
120    # cap leverage
121    self.long_lvg = min(self.leverage_cap, abs(1. / long_mean_beta))
122    self.short_lvg = min(self.leverage_cap, abs(1. / short_mean_beta))
123
124return self.long + self.short
125
126def OnData(self, data: Slice) -> None:
127# store vix value
128if self.vix in data and data[self.vix]:
129    self.data[self.vix].Add(data[self.vix].Value)
130
131if not self.selection_flag:
132    return
133self.selection_flag = False
134
135if len(self.factor_performance) < self.factor_perf_min_period or not self.data[self.vix].IsReady:
136    return
137
138# calculate variance
139sample_std:float = np.std(self.factor_performance) * np.sqrt(12)
140conditional_variance:float = sum((np.array(list(self.data[self.vix])) / 100) ** 2) / 12
141w:float = min([self.leverage_restriction, sample_std / conditional_variance])
142
143# trade execution
144stocks_invested:List[Symbol] = [x.Key for x in self.Portfolio if x.Value.Invested]
145for symbol in stocks_invested:
146    if symbol not in self.long + self.short:
147        self.Liquidate(symbol)
148
149long_len:int = len(self.long)
150short_len:int = len(self.short)
151
152for symbol in self.long:
153    if symbol in data and data[symbol]:
154        self.SetHoldings(symbol, (1. / long_len) * self.long_lvg * w)
155
156for symbol in self.short:
157    if symbol in data and data[symbol]:
158        self.SetHoldings(symbol, -(1. / short_len) * self.short_lvg * w)
159
160self.long.clear()
161self.short.clear()
162
163self.long_lvg = 1
164self.short_lvg = 1
165
166def Selection(self) -> None:
167self.selection_flag = True
168    
169# Custom fee model.
170class CustomFeeModel(FeeModel):
171def GetOrderFee(self, parameters):
172fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
173return OrderFee(CashAmount(fee, "USD"))