Trading Based on Levered ETFs Speculation Sentiment

1from AlgorithmImports import *
2import statsmodels.api as sm
3from scipy import stats
4# endregion
5class TradingBasedonLeveredETFsSpeculationSentiment(QCAlgorithm):
6def Initialize(self):
7 self.SetStartDate(2000, 1, 1)
8 self.SetCash(100_000)
9 self.long_etfs: List[str] = ['QLD', 'SSO', 'DDM']
10 self.short_etfs: List[str] = ['QID', 'SDS', 'DXD']
11 
12 # previous month's SO
13 self.previous_so: Dict[str, float] = {
14     ticker : 0 for ticker in self.long_etfs + self.short_etfs
15 }
16 
17 # monthly prices
18 self.m_price_data: Dict[Symbol, RollingWindow] = {}
19 # market data
20 self.m_regression_period: int = 36 + 1
21 self.market: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
22 self.m_price_data[self.market]: RollingWindow = RollingWindow[float](self.m_regression_period)
23 
24 self.net_percent_share_change_series: RollingWindow = RollingWindow[float](self.m_regression_period)
25 self.shares_out_data:Symbol = self.AddData(ETFSharesOutstanding, 'ETFSharesOutstanding', Resolution.Daily).Symbol
26 self.weight: Dict[Symbol, float] = {}   # traded weight for a given month
27 # universe selection
28 self.quantile: int = 5
29 self.leverage: int = 5
30 self.leverage_cap: int = 5
31 self.exchange_codes: List[str] = ['NYS']
32 
33 self.fundamental_count: int = 500
34 self.fundamental_sorting_key = lambda x: x.DollarVolume
35 self.recent_month: int = -1
36 self.rebalance_flag: bool = False
37 self.selection_flag: bool = False
38 self.UniverseSettings.Resolution = Resolution.Daily
39 self.AddUniverse(self.FundamentalSelectionFunction)
40 self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
41 self.settings.daily_precise_end_time = False
42def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
43 for security in changes.AddedSecurities:
44     security.SetFeeModel(CustomFeeModel())
45     security.SetLeverage(self.leverage)
46def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
47 if not self.selection_flag:
48     return Universe.Unchanged
49 # update rolling windows every month
50 for stock in fundamental:
51     symbol: Symbol = stock.Symbol
52     if symbol in self.m_price_data:
53         self.m_price_data[symbol].Add(stock.AdjustedPrice)
54 self.selection_flag = False
55 
56 selected: List[Fundamental] = [
57     f for f in fundamental if f.HasFundamentalData 
58     and f.Market == 'usa'
59     and f.SecurityReference.ExchangeId in self.exchange_codes
60     and f.MarketCap != 0
61 ]
62 if len(selected) > self.fundamental_count:
63     selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
64     
65 # warmup price rolling windows
66 for stock in selected:
67     symbol: Symbol = stock.Symbol
68     if symbol not in self.m_price_data:
69         self.m_price_data[symbol] = RollingWindow[float](self.m_regression_period)
70         history: DataFrame = self.History(symbol, self.m_regression_period * 30, Resolution.Daily)
71         if history.empty:
72             self.Log(f"Not enough data for {symbol} yet.")
73             continue
74         
75         closes: Series = history.loc[symbol].close
76         closes_len = len(closes.keys())
77         # find monthly closes
78         for index, time_close in enumerate(closes.items()):
79             # index out of bounds check.
80             if index + 1 < closes_len:
81                 date_month: int = time_close[0].date().month
82                 next_date_month: int = closes.keys()[index + 1].month
83             
84                 # found last day of month
85                 if date_month != next_date_month:
86                     self.m_price_data[symbol].Add(time_close[1])
87 if not self.net_percent_share_change_series.IsReady:
88     return Universe.Unchanged
89 # forming the SSI index
90 # at this point, both market price data and net share change series data are ready
91 market_prices: np.ndarray = np.array([x for x in self.m_price_data[self.market]])
92 market_returns: np.ndarray = market_prices[:-1] / market_prices[1:] - 1
93 share_changes: np.ndarray = np.array([x for x in self.net_percent_share_change_series])[:-1]
94 ols_model = self.multiple_linear_regression(share_changes, market_returns)
95 
96 # time series of residuals form the Speculation Sentiment Index
97 SSI: np.ndarray = ols_model.resid
98 SSI_sensitivity: Dict[Fundamental, float] = {}
99 self.daily_return: DataFrame = pd.DataFrame()
100 for stock in selected :
101     symbol: Symbol = stock.Symbol
102     if self.m_price_data[symbol].IsReady:
103         # at this point, both stock price data and net share change series data are ready
104         stock_prices: np.ndarray = np.array([x for x in self.m_price_data[symbol]])
105         stock_returns: np.ndarray = stock_prices[:-1] / stock_prices[1:] - 1
106         self.daily_return[stock] = stock_returns
107         # estimate monthly sensitivity to lagged SSI
108         ols_model = self.multiple_linear_regression(SSI[1:], stock_returns[:-1])
109         SSI_sensitivity[stock] = ols_model.params[1]
110 
111 if len(SSI_sensitivity) >= self.quantile:
112     # sorting by SSI sensitivity
113     sorted_by_sensitivity: List[Fundamental] = sorted(SSI_sensitivity, key = SSI_sensitivity.get, reverse=True)
114     quantile: int = int(len(sorted_by_sensitivity) / self.quantile)
115     
116     long: List[Fundamental] = []
117     short: List[Fundamental] = []
118     if SSI[0] > 0:
119         long = sorted_by_sensitivity[-quantile:]
120         short = sorted_by_sensitivity[:quantile]
121     else:
122         long = sorted_by_sensitivity[:quantile]
123         short = sorted_by_sensitivity[-quantile:]
124     # value weighting
125     for i, portfolio in enumerate([long, short]):
126         mc_sum: float = sum(map(lambda x: x.MarketCap, portfolio))
127         for stock in portfolio:
128             self.weight[stock.Symbol] = ((-1) ** i) * stock.MarketCap / mc_sum
129     # magnitude of SSI scales the portfolio itself; average leverage is equal to zero
130     abs_ssi_values: np.ndarray = abs(SSI)
131     percentile: float = stats.percentileofscore(abs_ssi_values[1:], abs_ssi_values[0]) / 100.
132     leverage: float = 1 - ((0.5 - percentile) * 2) if percentile < 0.5 else 1 + ((percentile - 0.5) * 2)
133     self.weight = {
134         x : w * leverage for x, w in self.weight.items()
135     }
136     
137     self.rebalance_flag = True
138 return list(self.weight.keys())
139def OnData(self, data: Slice) -> None:
140 if (self.Time.date() > ETFSharesOutstanding.get_last_update_date()):
141     self.selection_flag = False
142     self.Liquidate()
143     return
144 # SO data came in 
145 if self.shares_out_data in data and data[self.shares_out_data]:
146     # universe selection and rebalance is done once a month when new SO data comes
147     if self.Time.month != self.recent_month:
148         self.selection_flag = True
149         self.recent_month = self.Time.month
150         long_psc_sum: float = 0
151         short_psc_sum: float = 0
152         data_ready_ticker_cnt: int = 0    # n of tickers with SO data ready
153         for ticker in self.long_etfs + self.short_etfs:
154             so: float = data[self.shares_out_data].GetProperty(ticker)
155             
156             # ticker has previous month's SO data
157             if self.previous_so[ticker] != 0:
158                 data_ready_ticker_cnt += 1
159                 percent_share_change: float = -1 + (so / self.previous_so[ticker])
160                 if ticker in self.long_etfs:
161                     long_psc_sum += percent_share_change
162                 else:
163                     short_psc_sum += percent_share_change
164             
165             self.previous_so[ticker] = so
166         
167         # every ticker had SO data ready
168         if data_ready_ticker_cnt == len(self.long_etfs + self.short_etfs):
169             psc_net: float = long_psc_sum - short_psc_sum
170             self.net_percent_share_change_series.Add(psc_net)
171             
172 # rebalance once a month
173 if not self.rebalance_flag:
174     return
175 self.rebalance_flag = False
176 portfolio: List[PortfolioTarget] = [PortfolioTarget(symbol, w) for symbol, w in self.weight.items() if symbol in data and data[symbol]]
177 self.SetHoldings(portfolio, True)
178 self.weight.clear()
179
180def multiple_linear_regression(self, x:np.ndarray, y:np.ndarray, add_residual:bool = True):
181 x:np.ndarray = np.array(x).T
182 
183 if add_residual:
184     x = sm.add_constant(x)
185 result = sm.OLS(endog=y, exog=x).fit()
186 return result
187# Custom data
188class ETFSharesOutstanding(PythonData):
189_last_update_date: datetime.date = datetime(1,1,1).date()
190@staticmethod
191def get_last_update_date() -> datetime.date:
192return ETFSharesOutstanding._last_update_date
193def GetSource(self, config, date, isLiveMode):
194 return SubscriptionDataSource("data.quantpedia.com/backtesting_data/equity/etf_shares_outstanding.csv", SubscriptionTransportMedium.RemoteFile, FileFormat.Csv)
195def Reader(self, config, line, date, isLiveMode):
196 data = ETFSharesOutstanding()
197 data.Symbol = config.Symbol
198 
199 if not line[0].isdigit():
200     self.header_columns = line.split(';')
201     return None
202     
203 split = line.split(';')
204 data.Time = datetime.strptime(split[0], "%Y-%m-%d") + timedelta(days=1)
205 data.Value = float(split[1])
206 for i, col in enumerate(self.header_columns[1:]): # skip date
207     data[col] = float(split[i+1])
208 # store last update date
209 if data.Time.date() > ETFSharesOutstanding._last_update_date:
210     ETFSharesOutstanding._last_update_date = data.Time.date()
211 return data
212# Custom fee model.
213class CustomFeeModel():
214def GetOrderFee(self, parameters):
215 fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
216 return OrderFee(CashAmount(fee, "USD"))