Betting Against Correlation in S&P500 Stocks

1from AlgorithmImports import *
2from pandas.core.frame import DataFrame
3# endregion
4
5class BettingAgainstCorrelationInSP500Stocks(QCAlgorithm):
6
7def Initialize(self):
8self.SetStartDate(2000, 1, 1)
9self.SetCash(100000)
10
11self.volatility_period:int = 12*21
12self.correlation_period:int = 5*12*21
13self.quantile:int = 4
14self.portfolio_percentage:float = 1.
15
16self.prices:Dict[Symbol, RollingWindow] = {}
17self.weight:Dict[Symbol, float] = {}
18
19self.exchanges:List[str] = ['NYS', 'NAS', 'ASE']
20
21self.market_symbol:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
22self.prices[self.market_symbol] = RollingWindow[float](self.correlation_period)
23
24self.max_cap_weight:float = .1
25self.long_leg_corr_treshold:float = 0.
26self.long_leg_corr_substitute:float = .001
27
28self.fundamental_count:int = 1000
29self.fundamental_sorting_key = lambda x: x.DollarVolume
30
31self.selection_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, 0), self.Selection)
36
37def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
38for security in changes.AddedSecurities:
39    security.SetFeeModel(CustomFeeModel())
40    security.SetLeverage(10)
41
42def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
43# update daily prices
44for equity in fundamental:
45    symbol:Symbol = equity.Symbol
46
47    if symbol in self.prices:
48        self.prices[symbol].Add(equity.AdjustedPrice)
49
50if not self.selection_flag:
51    return Universe.Unchanged
52
53selected:List[Fundamental] = [x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and x.SecurityReference.ExchangeId in self.exchanges]
54if len(selected) > self.fundamental_count:
55    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
56
57volatility:Dict[Symbol, list[float, list]] = {}
58stocks_returns:Dict[Symbol, np.ndarray] = {}
59
60# warm up stock prices
61for stock in selected:
62    symbol:Symbol = stock.Symbol
63    
64    if symbol not in self.prices:
65        self.prices[symbol] = RollingWindow[float](self.correlation_period)
66        history:DataFrame = self.History(symbol, self.volatility_period, Resolution.Daily)
67        if history.empty:
68            continue
69
70        closes:pd.Series = history.loc[symbol].close
71
72        for time, close in closes.items():
73            self.prices[symbol].Add(close)
74    
75    # make sure SPY prices are ready
76    if not self.prices[self.market_symbol].IsReady:
77        continue
78    
79    # calculate volatility and store daily returns
80    if self.prices[symbol].IsReady:
81        prices:np.ndarray = np.array([x for x in self.prices[symbol]])
82        returns:np.ndarray = prices[:-1] / prices[1:] - 1
83        vol_value:float = np.std(returns[:self.volatility_period])
84
85        volatility[symbol] = vol_value
86        stocks_returns[symbol] = returns
87
88# make sure enough stocks has volatility value
89if len(volatility) < self.quantile:
90    return Universe.Unchanged
91
92quantile:int = int(len(volatility) / self.quantile)
93sorted_by_vol:List[Symbol] = [x[0] for x in sorted(volatility.items(), key=lambda item: item[1])]
94
95market_prices:np.ndarray = np.array([x for x in self.prices[self.market_symbol]])
96market_returns:np.ndarray = market_prices[:-1] / market_prices[1:] - 1
97
98# create long and short portfolio part
99for i in range(self.quantile):
100    long_leg:List[tuple[Symbol, float]] = []
101    short_leg:List[tuple[Symbol, float]] = []
102
103    total_long_corr:float = 0
104    total_short_corr:float = 0
105
106    correlation:Dict[Symbol, float] = {}
107    curr_quantile_stocks:List[Symbol] = sorted_by_vol[i * quantile : (i + 1) * quantile]
108
109    for symbol in curr_quantile_stocks:
110        stock_returns:np.ndarray = stocks_returns[symbol]
111        correlation_matrix:np.ndarray = np.corrcoef(stock_returns, market_returns)
112        corr_value:float = correlation_matrix[0][1]
113        correlation[symbol] = corr_value
114
115    corr_median:float = np.median(list(correlation.values()))
116
117    for symbol, corr_value in correlation.items():
118        # within each quartile we go long (short) low (high) correlation stocks using the median as a threshold
119        if corr_value >= corr_median:
120            short_leg.append((symbol, corr_value))
121            total_short_corr += abs(corr_value)
122        else:
123            if corr_value < self.long_leg_corr_treshold:
124                corr_value = self.long_leg_corr_substitute
125
126            long_leg.append((symbol, corr_value))
127            total_long_corr += 1 / abs(corr_value)
128
129    # weights calculations
130    for i, portfolio in enumerate([long_leg, short_leg]):
131        for symbol, corr_value in portfolio:
132            w:float = ((1 / corr_value) / total_long_corr) * (1 / self.quantile) * self.portfolio_percentage
133            w = min(self.max_cap_weight, w) # weight cap
134            self.weight[symbol] = ((-1) ** i) * w
135            
136return list(self.weight.keys())
137
138def OnData(self, data: Slice) -> None:
139# rebalance monthly
140if not self.selection_flag:
141    return
142self.selection_flag = False
143
144# trade execution
145portfolio:List[PortfolioTarget] = [PortfolioTarget(symbol, w) for symbol, w in self.weight.items() if symbol in data and data[symbol]]
146self.SetHoldings(portfolio, True)
147
148self.weight.clear()
149        
150def Selection(self) -> None:
151self.selection_flag = True
152
153# Custom fee model
154class CustomFeeModel(FeeModel):
155def GetOrderFee(self, parameters):
156fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
157return OrderFee(CashAmount(fee, "USD"))