Gamma Factor Premium

1from AlgorithmImports import *
2import data_tools
3import statsmodels.api as sm
4import numpy as np
5from dateutil.relativedelta import relativedelta
6# endregion
7
8class GammaFactorPremium(QCAlgorithm):
9
10def Initialize(self):
11self.SetStartDate(2010, 1, 1)
12self.SetCash(100_000)
13
14self.tickers_to_ignore:List[str] = ['TOPS']
15
16self.market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
17self.world_market:Symbol = self.AddEquity('VT', Resolution.Daily).Symbol
18self.asia_market:Symbol = self.AddEquity('ASHR', Resolution.Daily).Symbol
19self.vix:Symbol = self.AddData(CBOE, "VIX").Symbol
20# self.cpi_us:symbol = self.AddData(data_tools.QuandlValue, 'RATEINF/CPI_USA', Resolution.Daily).Symbol
21self.market_ff:Symbol = self.AddData(data_tools.MarketEQ, 'ff_market', Resolution.Daily).Symbol
22
23self.custom_data_monthly:List[str] = ['GS3M', 'GS1', 'GS2', 'GS10', 'UMCSENT', 'RTWEXBGS', 'BOGMBASE', 'CPIAUCSL']
24self.custom_data_weekly:List[str] = ['WALCL', 'ECBASSETSW']
25self.custom_data_daily:List[str] = ['DTWEXBGS', 'T10Y2Y', 'DCOILWTICO']
26self.currencies:List[str] = ['USDEUR', 'CNHUSD']
27
28# subscribe data
29self.currencies_symbol:List[Symbol] = [self.AddForex(x, Resolution.Daily, Market.Oanda).Symbol for x in self.currencies]
30
31self.monthly_custom_data:List[Symbol] = [self.AddData(data_tools.MonthlyQPData, ticker, Resolution.Daily).Symbol for ticker in self.custom_data_monthly]
32self.weekly_custom_data:List[Symbol] = [self.AddData(data_tools.WeeklyQPData, ticker, Resolution.Daily).Symbol for ticker in self.custom_data_weekly]
33self.daily_custom_data:List[Symbol] = [self.AddData(data_tools.DailyQPData, ticker, Resolution.Daily).Symbol for ticker in self.custom_data_daily]
34
35self.custom_data:List[Symbol] = self.monthly_custom_data + self.weekly_custom_data + self.daily_custom_data
36self.qc_data:List[Symbol] = [self.market, self.asia_market, self.vix] + self.currencies_symbol
37
38self.data:Dict[Symbol, float] = {}
39self.convex_stocks:List[Symbol] = []
40self.concave_stocks:List[Symbol] = []
41self.xma_factor_returns:List[float] = []
42
43self.period:int = 36
44self.quantile:int = 10
45self.leverage:int = 5
46self.threshold:int = 12
47
48self.traded_portolio:None|List[Symbol] = None
49
50self.fundamental_sorting_key = lambda x: x.MarketCap
51self.fundamental_count:int = 3000
52
53self.selection_flag:bool = False
54self.rebalance_flag:bool = False
55self.Settings.MinimumOrderMarginPortfolioPercentage = 0
56self.UniverseSettings.Resolution = Resolution.Daily
57self.AddUniverse(self.FundamentalSelectionFunction)
58self.Schedule.On(self.DateRules.MonthStart(self.market), self.TimeRules.AfterMarketOpen(self.market), self.Selection)
59
60def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
61for security in changes.AddedSecurities:
62    security.SetFeeModel(data_tools.CustomFeeModel())
63    security.SetLeverage(self.leverage)
64
65for security in changes.RemovedSecurities:
66    if security.Symbol in self.data:
67        self.data.pop(security.Symbol)
68    if security.Symbol in self.convex_stocks:
69        self.convex_stocks.remove(security.Symbol)
70    if security.Symbol in self.concave_stocks:
71        self.concave_stocks.remove(security.Symbol)
72
73def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
74# selected on month start
75if not self.selection_flag:
76    return Universe.Unchanged
77self.selection_flag = False
78
79market_ff_last_update_date:datetime.date = data_tools.MarketEQ._last_update_date
80monthly_custom_data_last_update_date:Dict[Symbol, datetime.date] = data_tools.MonthlyQPData._last_update_date
81weekly_custom_data_last_update_date:Dict[Symbol, datetime.date] = data_tools.WeeklyQPData._last_update_date
82daily_custom_data_last_update_date:Dict[Symbol, datetime.date] = data_tools.DailyQPData._last_update_date
83
84# data is still comming in
85if all([self.Securities[x].GetLastData() for x in self.custom_data]) and any([self.Time.date() >= monthly_custom_data_last_update_date[x] for x in monthly_custom_data_last_update_date]) \
86    and any([self.Time.date() >= weekly_custom_data_last_update_date[x] for x in weekly_custom_data_last_update_date]) and any([self.Time.date() >= daily_custom_data_last_update_date[x] for x in daily_custom_data_last_update_date]) \
87    and any(symbol not in data and not data[symbol] for symbol in self.qc_data):
88    self.Liquidate()
89    return Universe.Unchanged
90
91# FF data is still comming in
92if self.Securities[self.market_ff].GetLastData() and self.Time.date() >= market_ff_last_update_date:
93    self.Liquidate()
94    return Universe.Unchanged
95
96# store monthly FF prices
97if self.market_ff in self.data:
98    self.data[self.market_ff].update_daily_return(self.Securities[self.market_ff].Price)
99
100# store daily stock prices
101for stock in fundamental:
102    symbol:Symbol = stock.Symbol
103
104    if symbol in self.data:
105        self.data[symbol].update_daily_return(stock.AdjustedPrice)
106
107selected:List[Fundamental] = [x for x in fundamental if x.HasFundamentalData and x.MarketCap != 0 and x.Symbol.Value not in self.tickers_to_ignore]
108if len(selected) > self.fundamental_count:
109            selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
110
111# price warmup
112for stock in selected:
113    symbol:Symbol = stock.Symbol
114    if symbol in self.data:
115        continue
116    
117    self.data[symbol] = data_tools.SymbolData(self.period)
118    history:DataFrame = self.History(symbol, self.period, Resolution.Daily)
119    if history.empty:
120        self.Log(f"Not enough data for {symbol} yet.")
121        continue
122    closes:pd.Series = history.loc[symbol].close
123    closes = closes.groupby(pd.Grouper(freq='MS')).last()
124    for time, close in closes.items():
125        self.data[symbol].update_daily_return(close)
126
127if self.market_ff not in self.data:
128    self.data[self.market_ff] = data_tools.SymbolData(self.period)
129    history:DataFrame = self.History(self.market_ff, self.period, Resolution.Daily)
130    if not history.empty:
131        values:pd.Series = history.loc[self.market_ff].value.groupby(pd.Grouper(freq='MS')).last()
132        for time, value in values.items():
133            self.data[self.market_ff].update_daily_return(value)
134    else:
135        self.Log(f"Not enough data for {symbol} yet.")
136
137selected:Dict[Symbol, Fundamental] = {x.Symbol: x for x in selected if self.data[x.Symbol].is_ready()}
138
139# second regression and prediction
140if len(self.convex_stocks) != 0 and len(self.concave_stocks) != 0:
141    self.xma_factor_returns.append(np.mean(np.array([self.data[x].get_last_return() for x in self.convex_stocks if self.data[x].is_ready()]), axis=0) - np.mean(np.array([self.data[x].get_last_return() for x in self.concave_stocks if self.data[x].is_ready()]), axis=0))
142
143if len(self.xma_factor_returns) >= self.threshold:
144    history_custom_data:DataFrame = self.History(self.custom_data, start=self.Time.date() - relativedelta(months=len(self.xma_factor_returns)), end=self.Time.date())['value'].unstack(level=0)
145    history_qc_data:DataFrame = self.History(self.qc_data, start=self.Time.date() - relativedelta(months=len(self.xma_factor_returns)), end=self.Time.date())['close'].unstack(level=0)
146    history_custom_data = history_custom_data.groupby(pd.Grouper(freq='MS')).last()
147    history_qc_data = history_qc_data.groupby(pd.Grouper(freq='MS')).last()
148    independent_variables:DataFrame = pd.concat([history_custom_data, history_qc_data], axis=1)[-len(self.xma_factor_returns):]
149
150    independent_variables = independent_variables.dropna(axis=1, how='any')
151
152    x:np.ndarray = independent_variables[:-1].values
153    y:np.ndarray = self.xma_factor_returns[1:]
154
155    model = self.multiple_linear_regression(x, y)
156    predicted_y:np.ndarray = model.predict(sm.add_constant(independent_variables[-1:].values, has_constant='add'))
157
158    self.traded_portfolio = self.convex_stocks if predicted_y > 0 else [self.world_market]
159
160    self.rebalance_flag = True
161
162if len(selected) != 0:
163    if not self.data[self.market_ff].is_ready():
164        return Universe.Unchanged
165
166    # stock returns
167    returns_by_stock:Dict[Symbol, List[Tuple[datetime, float]]] = {sym : sym_data.get_returns() for sym, sym_data in self.data.items() if sym_data.is_ready() and sym in selected and sym != self.market_ff}
168    stock_returns:List = list(zip(*[[i for i in x] for x in returns_by_stock.values()]))
169    
170    # FF returns
171    ff_returns:np.ndarray = np.array(self.data[self.market_ff].get_returns())
172    transformed_array = np.column_stack((
173        ff_returns,
174        ff_returns ** 2))
175
176    transformed_array = np.array(transformed_array, dtype=float)
177
178    # run stock regression
179    x:np.ndarray = transformed_array
180    y:np.ndarray = stock_returns
181    model = self.multiple_linear_regression(x, y)
182    gamma_values:np.ndarray = model.params[2]
183
184    gamma:Dict[Symbol, float] = {}
185    
186    # fetch gamma parameters for each stock
187    for i, asset in enumerate(returns_by_stock):
188        asset_s:Symbol = self.Symbol(asset)
189
190        # fill data
191        if asset in selected:
192            if gamma_values[i] is not None:
193                gamma[asset_s] = gamma_values[i]
194
195    # sort by gamma and divide into quantiles
196    if len(gamma) >= self.quantile:
197        sorted_gamma:List[Symbol] = sorted(gamma.items(), key=lambda x:x[1])
198        quantile:int = int(len(gamma) / self.quantile)
199        self.convex_stocks = [symbol for symbol, gamma in sorted_gamma][-quantile:]
200        self.concave_stocks = [symbol for symbol, gamma in sorted_gamma][:quantile]
201
202return self.convex_stocks + self.concave_stocks
203
204def OnData(self, data: Slice) -> None:
205if not self.rebalance_flag:
206    return Universe.Unchanged
207self.rebalance_flag = False
208
209# order execution
210# invested: List[Symbol] = [x.Key for x in self.Portfolio if x.Value.Invested]
211# for symbol in invested:
212#     if symbol not in self.traded_portfolio:
213#         self.Liquidate(symbol)
214
215# for symbol in self.traded_portfolio:
216#     if symbol in data and data[symbol]:
217#         self.SetHoldings(symbol, round(1 / len(self.traded_portfolio), 5))
218
219targets:List[PortfolioTarget] = [PortfolioTarget(symbol, round(1 / len(self.traded_portfolio), 5)) for symbol in self.traded_portfolio if symbol in data and data[symbol]]      
220self.SetHoldings(targets, True)
221
222def Selection(self) -> None:
223self.selection_flag = True
224
225def multiple_linear_regression(self, x:np.ndarray, y:np.ndarray):
226x = sm.add_constant(x, has_constant='add')
227result = sm.OLS(endog=y, exog=x).fit()
228return result