US Sector Rotation with Five-Factor Fama-French Alphas

1from AlgorithmImports import *
2from data_tools import QuantpediaFamaFrench, CustomFeeModel, FamaFrenchData, \
3MultipleLinearRegression, SymbolData
4class USSectorRotationFiveFactorFamaFrenchAlphas(QCAlgorithm):
5def Initialize(self):
6self.SetStartDate(2000, 1, 1)
7self.SetCash(100000)
8self.leverage:int = 10
9self.period:int = 37
10self.total_factor_num:int = 5
11self.max_missing_days:int = 5
12self.min_prices:int = 15
13self.min_daily_perfs:int = 15
14self.recent_month:int = -1
15self.data:dict[Symbol, SymbolData] = {}
16self.long:list[Symbol] = []
17self.fama_french:Symbol = self.AddData(QuantpediaFamaFrench, 'fama_french_5_factor', Resolution.Daily).Symbol
18self.ff_factor_names:list[str] = ['market', 'size', 'value', 'profitability', 'investment']
19self.ff_data:dict[str, FamaFrenchData] = { ff_factor_name: FamaFrenchData(self.period) for ff_factor_name in self.ff_factor_names }
20
21# self.symbols = []
22self.tickers = [
23    "VNQ",  # Vanguard Real Estate Index Fund
24    "XLK",  # Technology Select Sector SPDR Fund
25    "XLE",  # Energy Select Sector SPDR Fund
26    "XLV",  # Health Care Select Sector SPDR Fund
27    "XLF",  # Financial Select Sector SPDR Fund
28    "XLI",  # Industrials Select Sector SPDR Fund
29    "XLB",  # Materials Select Sector SPDR Fund
30    "XLY",  # Consumer Discretionary Select Sector SPDR Fund
31    "XLP",  # Consumer Staples Select Sector SPDR Fund
32    "XLU"   # Utilities Select Sector SPDR Fund
33]
34for symbol in self.tickers:
35    data:Security = self.AddEquity(symbol, Resolution.Daily)
36    data.SetFeeModel(CustomFeeModel())
37    data.SetLeverage(self.leverage)
38    
39    self.data[data.Symbol] = SymbolData(self.period)
40
41self.settings.minimum_order_margin_portfolio_percentage = 0.
42self.settings.daily_precise_end_time = False
43
44def OnData(self, data):
45curr_date:datetime.date = self.Time.date()
46# store FF performance values
47if self.fama_french in data and data[self.fama_french]:
48    for ff_factor_name, factor_data in self.ff_data.items():
49        ff_factor_value:float = float(data[self.fama_french].GetProperty(ff_factor_name))
50        factor_data.update_daily_perfs(ff_factor_value / 100.)
51# store etf prices
52for symbol, symbol_data in self.data.items():
53    if symbol in data and data[symbol]:
54        price:float = data[symbol].Value
55        symbol_data.update_prices(price)
56
57# monthly rebalance
58if self.Time.month == self.recent_month:
59    return
60self.recent_month = self.Time.month
61if self.securities[self.fama_french].get_last_data() and self.time.date() > QuantpediaFamaFrench.get_last_update_date()[self.fama_french]:
62    self.liquidate()
63    return
64# check FF data presence
65for ff_factor_name, ff_factor_data in self.ff_data.items():
66        
67    if ff_factor_data.daily_perfs_ready(self.min_daily_perfs):
68        ff_factor_data.update_monthly_perfs()
69    ff_factor_data.reset_daily_pefs()
70# regression model
71regression_x:list[list[float]] = [factor_data.get_regression_data(self.period - 1) for _, factor_data in self.ff_data.items() \
72    if factor_data.monthly_perfs_ready()]
73for symbol, symbol_data in self.data.items():
74    if symbol_data.prices_ready(self.min_prices):
75        symbol_data.update_monthly_perfs()
76    if symbol_data.monthly_perfs_ready() and len(regression_x) == self.total_factor_num:
77        monthly_perfs:list[float] = symbol_data.get_regression_data(self.period - 1)
78        regression_model = MultipleLinearRegression(regression_x, monthly_perfs)
79        ff5_alpha:float = regression_model.params[0]
80        symbol_data.update_ff5_alpha(ff5_alpha)
81        if symbol_data.ff5_alpha_ready() and symbol_data.are_all_ff5_alpha_positive():
82            self.long.append(symbol)
83    else:
84        # make sure ff5_alpha are consecutive
85        symbol_data.reset_ff5_alpha()
86    symbol_data.reset_daily_pefs()
87
88# trade execution
89long_length:int = len(self.long)
90stocks_invested:list[Symbol] = [x.Key for x in self.Portfolio if x.Value.Invested]
91for symbol in stocks_invested:
92    if symbol not in self.long:
93        self.Liquidate(symbol)
94for symbol in self.long:
95    if symbol in data and data[symbol]:
96        self.SetHoldings(symbol, 1 / long_length)
97
98self.long.clear()