Mean Variance Factor Timing

1from AlgorithmImports import *
2from scipy.optimize import minimize
3import data_tools
4#endregion
5
6class MeanVarianceFactorTiming(QCAlgorithm):
7
8def Initialize(self):
9self.SetStartDate(2010, 1, 1)
10self.SetCash(100000)
11
12self.period:int = 60 * 21
13
14# warm up fama french values for idiosyncratic volatility
15self.SetWarmup(self.period, Resolution.Daily)
16
17self.data:dict = {}
18
19self.fama_french_symbol:Symbol = self.AddData(data_tools.QuantpediaFamaFrench, 'fama_french_5_factor', Resolution.Daily).Symbol
20self.ff_factor_names:list[str] = ['market', 'size', 'value', 'profitability', 'investment']
21
22# ff performance data
23self.fama_french_data:dict = { ff_factor_name : RollingWindow[float](self.period) for ff_factor_name in self.ff_factor_names }
24
25# ff traded symbols
26for factor_name in self.ff_factor_names:
27    data:Security = self.AddData(data_tools.QuantpediaFamaFrenchEquity, f'fama_french_5_{factor_name}_eq', Resolution.Daily)
28    data.SetLeverage(3)
29    data.SetFeeModel(data_tools.CustomFeeModel())
30
31self.recent_month:int = -1
32
33def OnData(self, data):
34# update fama french values on daily basis
35if self.fama_french_symbol in data and data[self.fama_french_symbol]:
36    for ff_factor_name in self.ff_factor_names:
37        self.fama_french_data[ff_factor_name].Add(data[self.fama_french_symbol].GetProperty(ff_factor_name))
38
39if self.recent_month == self.Time.month:
40    return
41self.recent_month = self.Time.month
42
43# optimization
44if all(x[1].IsReady for x in self.fama_french_data.items()):
45    perf_df:pd.DataFrame = pd.DataFrame(columns=self.ff_factor_names)
46    for ff_factor_name in self.ff_factor_names:
47        perf_df[ff_factor_name] = np.array([x for x in self.fama_french_data[ff_factor_name]][::-1])
48
49    opt, weights = self.optimization_method(perf_df)
50    for ff_factor_symbol, w in weights.items():
51        traded_symbol:str = f'fama_french_5_{ff_factor_symbol}_eq'
52        if abs(w) > 0.001:
53            self.SetHoldings(traded_symbol, w)
54        else:
55            self.Liquidate(traded_symbol)
56
57def optimization_method(self, returns:pd.DataFrame):
58'''Maximize sharpe ratio method'''
59# objective function
60fun = lambda weights: - np.sum(returns.mean() * weights) * 252 / np.sqrt(np.dot(weights.T, np.dot(returns.cov() * 252, weights)))
61
62# Constraint #1: The weights can be negative, which means investors can short a security.
63constraints = [{'type': 'eq', 'fun': lambda w: 1 - np.sum(w)}]
64
65size = returns.columns.size
66x0 = np.array(size * [1. / size])
67# bounds = tuple((self.minimum_weight, self.maximum_weight) for x in range(size))
68bounds = tuple((0, 1) for x in range(size))
69
70opt = minimize(fun,                         # Objective function
71               x0,                          # Initial guess
72               method='SLSQP',              # Optimization method:  Sequential Least SQuares Programming
73               bounds = bounds,             # Bounds for variables 
74               constraints = constraints)   # Constraints definition
75
76return opt, pd.Series(opt['x'], index = returns.columns)