Sector Rotation via Credit Relative Value

1import numpy as np
2from AlgorithmImports import *
3from scipy import stats
4from typing import List, Dict
5import data_tools
6class SectorRotationViaCreditRelativeValue(QCAlgorithm):
7
8def Initialize(self) -> None:
9self.SetStartDate(2000, 1, 1)
10self.SetCash(100000)
11
12self.symbols: List[str] = [
13    "XLK",  # Technology Select Sector SPDR Fund
14    "XLE",  # Energy Select Sector SPDR Fund
15    "XLV",  # Health Care Select Sector SPDR Fund
16    "XLF",  # Financial Select Sector SPDR Fund
17    "XLI",  # Industrials Select Sector SPDR Fund
18    "XLB",  # Materials Select Sector SPDR Fund
19    "XLY",  # Consumer Discretionary Select Sector SPDR Fund
20    "XLP",  # Consumer Staples Select Sector SPDR Fund
21    "XLU"   # Utilities Select Sector SPDR Fund
22]
23
24self.regression_period: int = 26 * 5 # Need 26 weeks data
25self.leverage: int = 5
26self.segment: int = 6
27self.regression_data: Dict[str, data_tools.SymbolData] = {}
28
29for symbol in self.symbols:
30    self.AddEquity(symbol, Resolution.Daily)
31    self.regression_data[symbol] = data_tools.SymbolData(self.regression_period)
32
33self.rf_asset: Symbol = self.AddEquity('BIL', Resolution.Daily).Symbol
34self.symbol: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
35
36self.HYB: Symbol = self.AddData(data_tools.QuantpediaDailyData, 'BAMLH0A2HYBEY', Resolution.Daily).Symbol
37self.regression_data[self.HYB.Value] = data_tools.SymbolData(self.regression_period)
38
39self.selection_flag: bool = False
40self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
41self.UniverseSettings.Resolution = Resolution.Daily
42self.Schedule.On(self.DateRules.WeekStart(self.symbol), self.TimeRules.AfterMarketOpen(self.symbol), self.Selection)
43self.settings.daily_precise_end_time = False
44def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
45for security in changes.AddedSecurities:
46    security.SetFeeModel(data_tools.CustomFeeModel())
47    security.SetLeverage(self.leverage)
48
49def OnData(self, data: Slice) -> None:
50custom_data_last_update_date: Dict[Symbol, datetime.date] = data_tools.LastDateHandler.get_last_update_date()
51ETF_market: Dict[str, float] = {}
52
53# check if data is still coming
54if self.Securities[self.HYB].GetLastData() and self.Time.date() > custom_data_last_update_date[self.HYB]:
55    self.Liquidate()
56    return
57# Each day storing data about symbols in self.symbols and HYB index
58for symbol in self.symbols: 
59    if symbol in data and data[symbol]:
60        price: float = data[symbol].Value
61        if price != 0:
62            ETF_market[symbol] = price
63            self.regression_data[symbol].update(price)
64
65if self.HYB in data and data[self.HYB]:
66    ETF_market[self.HYB.Value] = data[self.HYB].Value
67    self.regression_data[self.HYB.Value].update(data[self.HYB].Value)
68# Rebalance weekly 
69if not self.selection_flag:
70    return
71self.selection_flag = False
72
73ETF_disconnect: Dict[str, float] = {}
74# If HYB data aren't ready, we can't calculate any regression
75if self.regression_data[self.HYB.Value].is_ready():
76    X: list[float] = [x for x in self.regression_data[self.HYB.Value].RegressionData][::-1]
77    
78    for symbol in self.symbols:
79        if self.regression_data[symbol].is_ready():
80            if symbol in ETF_market and self.HYB.Value in ETF_market:
81                Y: float = [x for x in self.regression_data[symbol].RegressionData][::-1]
82                slope, intercept, r_value, p_value, std_err = stats.linregress(X, Y)
83                ETF_fair: float = slope * ETF_market[self.HYB.Value] + intercept
84                ETF_disconnect[symbol] = (ETF_fair - ETF_market[symbol]) / ETF_market[symbol]
85
86long: List[str] = []
87negative_disconnect: List[str] = []
88rf_weight: float = .0
89if len(ETF_disconnect) != 0:
90    # Sorted descending
91    sorted_by_disconnect: Dict[str, float] = {k: v for k, v in sorted(ETF_disconnect.items(), key=lambda item: item[1], reverse=True)}
92    for symbol, disc in sorted_by_disconnect.items():
93        if disc > 0:
94            long.append(symbol)
95        else:
96            negative_disconnect.append(symbol)
97    
98    long = long[:self.segment]
99        
100    total_count: int = len(long) + len(negative_disconnect)
101    long_weight: float = len(long) / total_count
102    rf_weight = len(negative_disconnect) / total_count
103        
104# Trade execution.
105invested: List[str] = [x.Key.Value for x in self.Portfolio if x.Value.Invested]
106for symbol in invested:
107    if symbol not in long:
108        self.Liquidate(symbol)
109    
110for symbol in long:
111    if symbol in data and data[symbol]:
112        self.SetHoldings(symbol, long_weight / len(long))
113    
114if rf_weight != 0:
115    if self.rf_asset in data and data[self.rf_asset]:
116        self.SetHoldings(self.rf_asset.Value, rf_weight)
117
118def Selection(self) -> None:
119self.selection_flag = True