Betting Against Correlation Effect

1import numpy as np
2from AlgorithmImports import *
3from typing import Dict, List
4class BettingAgainstCorrelationEffect(QCAlgorithm):
5def Initialize(self):
6self.SetStartDate(2000, 1, 1)
7self.SetCash(100000)
8self.leverage:int = 5
9self.quantile:int = 5
10self.min_share_price:int = 5
11self.fundamental_sorting_key = lambda x: x.DollarVolume
12self.fundamental_count:int = 500
13self.quintile_internal_count:int = 10   # Pick n low corraleted stocks and n high corraleted stocks in each quintile.
14self.data:Dict[Symbol, RollingWindow] = {} # Storing daily prices
15self.long:List[Symbol] = []
16self.short:List[Symbol] = []
17
18self.symbol:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
19self.period:int = 21
20self.SetWarmUp(self.period)
21
22self.market_prices:RollingWindow = RollingWindow[float](self.period)
23
24self.selection_flag:bool = False
25self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
26self.UniverseSettings.Resolution = Resolution.Daily
27self.AddUniverse(self.FundamentalSelectionFunction)
28self.Schedule.On(self.DateRules.MonthStart(self.symbol), self.TimeRules.AfterMarketOpen(self.symbol), self.Selection)
29
30self.settings.daily_precise_end_time = False
31
32def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
33for security in changes.AddedSecurities:
34    security.SetFeeModel(CustomFeeModel())
35    security.SetLeverage(self.leverage)
36def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
37# Update Prices in RollingWindow
38for stock in fundamental:
39    symbol:Symbol = stock.Symbol
40    
41    if symbol in self.data:
42        self.data[symbol].Add(stock.AdjustedPrice)
43
44if not self.selection_flag:
45    return Universe.Unchanged
46
47selected:List[Fundamental] = [
48    x for x in fundamental if x.HasFundamentalData and x.Price > self.min_share_price and x.Market == 'usa'
49]
50
51if len(selected) > self.fundamental_count:
52    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
53
54# Warmup price rolling windows.
55for stock in selected:
56    symbol:Symbol = stock.Symbol
57    if symbol in self.data:
58        continue
59    self.data[symbol] = RollingWindow[float](self.period)
60    history:DataFrame = self.History(symbol, self.period, Resolution.Daily)
61    if history.empty:
62        self.Log(f"Not enough data for {symbol} yet")
63        continue
64    closes:Series = history.loc[symbol].close
65    for time, close in closes.items():
66        self.data[symbol].Add(close)
67        
68if not self.market_prices.IsReady:
69    return Universe.Unchanged
70
71market_prices:List[float] = [x for x in self.market_prices]
72
73correlation:Dict[Symbol, float] = {}
74volatility:Dict[Symbol, float] = {}
75
76for stock in selected:
77    symbol:Symbol = stock.Symbol
78    if not self.data[symbol].IsReady:
79        continue
80    closes:List[float] = [x for x in self.data[symbol]]
81    volatility[symbol] = self.Volatility(closes)
82    correlation[symbol] = np.correlate(closes, market_prices)
83# Volatility sorting
84sorted_by_vol:List[Symbol] = [x[0] for x in sorted(volatility.items(), key=lambda item: item[1], reverse=True)]
85quantile:int = int(len(sorted_by_vol)/self.quantile)
86
87# Group splitting
88volatility_groups:List[List[Symbol]] = [sorted_by_vol[x:x+quantile] for x in range(0, len(sorted_by_vol),quantile)]
89group_count:int = len(volatility_groups)
90
91# Long = array of arrays = groups by 10 stocks to long
92# Short = array of arrays = groups by 10 stocks to short
93for vol_group in volatility_groups:
94    symbols:List[Symbol] = [x for x in vol_group]
95    sorted_by_correlation:List[Symbol] = sorted(symbols, key = lambda x: correlation[x], reverse = True)
96    
97    # Go long stocks with highest correlations and short stocks with lowest correlation
98    self.long.append([x for x in sorted_by_correlation][:self.quintile_internal_count])
99    self.short.append([x for x in sorted_by_correlation][-self.quintile_internal_count:])
100    
101long_short:List[Symbol] = []
102for group in self.long + self.short:
103    for symbol in group:
104        long_short.append(symbol)
105        
106return long_short
107
108def OnData(self, data: Slice) -> None:
109# store daily market price
110if self.symbol in data and data[self.symbol]:
111    price:float = data[self.symbol].Value
112    self.market_prices.Add(price)
113    
114if not self.selection_flag:
115    return
116self.selection_flag = False
117# Trade execution
118self.Liquidate()
119
120if len(self.long) == 0 or len(self.short) == 0:
121    self.Liquidate()
122    return
123long_length:int = len(self.long)
124short_length:int = len(self.short)
125
126equity_per_long_group:float = float(1 / long_length)
127equity_per_short_group:float = float(1 / short_length)
128
129for group in self.long:
130    count:int = int(len(group))
131    weight_index:float = 1 / sum([x for x in range(count+1)])
132    for symbol in group:
133        if symbol in data and data[symbol]:
134            self.SetHoldings(symbol, equity_per_long_group * (count*weight_index))
135            count -= 1
136for group in self.short:
137    count = int(len(group))
138    weight_index = 1 / sum([x for x in range(count+1)])
139    count = 1
140    for symbol in group:
141        if symbol in data and data[symbol]:
142            self.SetHoldings(symbol, -equity_per_short_group * (count*weight_index))
143            count -= 1
144    
145self.long.clear()
146self.short.clear()
147def Selection(self) -> None:
148# Monthly rebalance
149self.selection_flag = True
150
151def Volatility(self, values) -> float:
152values = np.array(values)
153returns = (values[:-1] - values[1:]) / values[1:]
154return np.std(returns)  
155
156# Custom fee model
157class CustomFeeModel(FeeModel):
158def GetOrderFee(self, parameters):
159fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
160return OrderFee(CashAmount(fee, "USD"))