Arbitraging Levered ETFs

1from AlgorithmImports import *
2from typing import Dict, List, Set
3from data_tools import CustomFeeModel, TradePair, SymbolData
4# endregion
5
6class ArbitragingLeveredETFs(QCAlgorithm):
7
8def Initialize(self):
9self.SetStartDate(2010, 1, 1)
10self.SetCash(100000)
11
12self.leverage:int = 5
13self.corr_threshold:float = 0.5
14
15self.period:int = 12 * 21
16
17self.data:Dict[Symbol, SymbolData] = {}
18self.trades:List[List[Symbol, float]] = []
19
20self.l_etfs:Dict[str, float] = {}
21
22l_etf_tickers_csv:str = self.Download('data.quantpedia.com/backtesting_data/equity/leveraged_etf_tickers.csv')
23lines:List[str] = l_etf_tickers_csv.split('\r\n')
24
25for line in lines[1:]:
26    if line == '':
27        continue
28
29    line_split:List[str] = line.split(';')
30    ticker:str = line_split[0]
31    leverage:float = float(line_split[-1])
32
33    self.l_etfs[ticker] = leverage
34
35etf_tickers_csv:str = self.Download('data.quantpedia.com/backtesting_data/equity/not_leveraged_etf_tickers.csv')
36lines:List[str] = etf_tickers_csv.split('\r\n')
37self.etfs:List[str] = { ticker: 1 for ticker in lines[1:] if ticker != '' }
38
39self.market_symbol:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
40
41self.selection_flag:bool = False
42self.UniverseSettings.Resolution = Resolution.Daily
43self.AddUniverse(self.FundamentalSelectionFunction)
44
45self.Schedule.On(self.DateRules.MonthStart(self.market_symbol), self.TimeRules.BeforeMarketClose(self.market_symbol, 0), self.Selection)
46
47def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
48for security in changes.AddedSecurities:
49    security.SetFeeModel(CustomFeeModel())
50    security.SetLeverage(self.leverage)
51
52def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
53for stock in fundamental:
54    symbol:Symbol = stock.Symbol
55
56    if symbol in self.data:
57        self.data[symbol].update_prices(stock.AdjustedPrice)
58
59if not self.selection_flag:
60    return Universe.Unchanged
61
62selected_leveraged_symbols:List[Symbol] = []
63selected_symbols:List[Symbol] = []
64
65for stock in fundamental:
66    symbol:Symbol = stock.Symbol
67    ticker:str = symbol.Value
68
69    if ticker in self.etfs:
70        selected_symbols.append(symbol)
71
72    elif ticker in self.l_etfs:
73        selected_leveraged_symbols.append(symbol)
74
75    if symbol not in self.data:
76        self.data[symbol] = SymbolData(self.period)
77        history:pd.DataFrame = self.History(symbol, self.period, Resolution.Daily)
78
79        if history.empty:
80            continue
81
82        closes:pd.Series = history.loc[symbol].close
83
84        for _, close in closes.items():
85            self.data[symbol].update_prices(close)
86
87trade_pairs:List[TradePair] = []
88symbols_to_trade:Set[Symbol] = set()
89abs_corr_by_symbol:Dict[Symbol, Tuple] = {}
90
91# pair each LETF to the ETF and calculate their correlation based on daily returns in one year
92for l_symbol in selected_leveraged_symbols:
93    if not self.data[l_symbol].is_ready():
94        continue
95
96    leveraged_daily_returns:np.array = self.data[l_symbol].get_daily_returns()
97
98    for symbol in selected_symbols:
99        if not self.data[symbol].is_ready():
100            continue
101
102        daily_returns:np.array = self.data[symbol].get_daily_returns()
103
104        correlation:float = np.corrcoef(leveraged_daily_returns, daily_returns)[0][-1]
105
106        # make sure correlation is greater than threshold
107        abs_corr:float = abs(correlation)
108        if abs_corr >= self.corr_threshold:
109            if (l_symbol not in abs_corr_by_symbol) or \
110                (l_symbol in abs_corr_by_symbol and abs_corr > abs_corr_by_symbol[l_symbol][1]):
111                # go short (long) on ETF, when correlation is negative (positive)
112                short_signal:bool = True if correlation < 0 else False
113                abs_corr_by_symbol[l_symbol] = (symbol, abs_corr, short_signal)
114
115    # create trade pairs
116    for l_symbol, pair_tuple in abs_corr_by_symbol.items():
117        symbol:Symbol = pair_tuple[0]
118        short_signal:bool = pair_tuple[2]
119
120        symbols_to_trade.add(symbol)
121        symbols_to_trade.add(l_symbol)
122        
123        trade_pairs.append(TradePair(symbol, l_symbol, short_signal))
124
125if len(trade_pairs) == 0:
126    return Universe.Unchanged
127
128total_trades:int = len(trade_pairs) * 2
129portfolio_partition:float = self.Portfolio.TotalPortfolioValue / total_trades
130
131# calculate quantity for each ETF and LETF in trade pairs
132for trade_pair in trade_pairs:
133    l_etf_price:float = self.data[trade_pair.l_etf_symbol].get_last_price()
134    l_etf_leverage:float = self.l_etfs[trade_pair.l_etf_symbol.Value]
135    etf_price:float = self.data[trade_pair.etf_symbol].get_last_price()
136
137    l_etf_quantity:float = np.floor(portfolio_partition / l_etf_price / l_etf_leverage)
138    etf_quantity:float = np.floor(portfolio_partition / etf_price)
139
140    self.trades.append([trade_pair.l_etf_symbol, -l_etf_quantity])
141    self.trades.append([trade_pair.etf_symbol, -etf_quantity if trade_pair.short_signal else etf_quantity])
142
143return list(symbols_to_trade)
144
145def OnData(self, data: Slice) -> None:
146if not self.selection_flag:
147    return
148self.selection_flag = False
149
150self.Liquidate()
151
152for symbol, quantity in self.trades:
153    self.MarketOrder(symbol, quantity)
154
155self.trades.clear()
156
157def Selection(self) -> None:
158self.selection_flag = True