Traditional Carry in Cryptocurrencies

1from AlgorithmImports import *
2import data_tools
3from typing import List, Dict
4# endregion
5
6class TraditionalCarryinCryptocurrencies(QCAlgorithm):
7
8def Initialize(self):
9self.SetStartDate(2015, 1, 1) 
10self.SetCash(100000)
11self.market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
12
13self.crypto_tickers:List[str] = ['BTCUSD', 'ETHUSD', 'SOLUSD', 'ADAUSD', 'XRPUSD', 'DOTUSD', 'DOGEUSD', 'LUNAUSD', 'AVAXUSD', 'UNIUSD',
14                                'LINKUSD', 'LTCUSD', 'BCHABCUSD', 'BSVUSD', 'FILUSD', 'XLMUSD', 'XTZUSD', 'NEOUSD', 'ATOMUSD', 'IOTAUSD', 
15                                'ETCUSD', 'DASHUSD', 'EGLDUSD', 'AAVEUSD', 'ENJUSD', 'EOSUSD', 'MKRUSD', 'MANAUSD', 'SNXUSD', 'FTTUSD', 
16                                'OMGUSD', 'SUSHIUSD', 'YFIUSD', 'WBTCUSD', 'XMRUSD', 'ZECUSD', 'ZRXUSD', 'XRAUSD', 'AMPLUSD', 'GRTUSD', 
17                                'DGBUSD', '1INCHUSD']
18
19self.crypto_int_symbols: Dict[str, str] = {x : x + '_FRR' for x in self.crypto_tickers}                       
20
21self.IR_symbol_by_price_symbol:Dict[Symbol, Symbol] = {}
22self.rebalance_flag:bool = False
23self.quantile:int = 3
24self.leverage:int = 2
25self.period:int = 365
26self.percentage_traded:float = .2
27
28# data subscription
29for ticker, ticker_interest in self.crypto_int_symbols.items():
30    data = self.AddCrypto(ticker, Resolution.Daily, Market.Bitfinex, )
31    data.SetLeverage(self.leverage)
32    crypto_symbol:Symbol = data.Symbol
33
34    interest_symbol = self.AddData(data_tools.CryptoInterestRate, ticker_interest, Resolution.Daily).Symbol
35    self.IR_symbol_by_price_symbol[crypto_symbol] = interest_symbol
36
37self.Schedule.On(self.DateRules.WeekEnd(self.market), self.TimeRules.BeforeMarketClose(self.market), self.Selection)
38
39def OnData(self, data: Slice):
40# weekly rebalance
41if not self.rebalance_flag:
42    return
43self.rebalance_flag = False
44
45rates_last_update_date:Dict[Symbol, datetime.date] = data_tools.CryptoInterestRate.get_last_update_date()
46rates:Dict[Symbol, float] = {}
47
48for symbol, symbol_interest in self.IR_symbol_by_price_symbol.items(): 
49    # both crypto price data and IR data is still comming in
50    if symbol in data and data[symbol]:
51        if self.Securities[symbol_interest].GetLastData() and symbol_interest in rates_last_update_date and self.Time.date() <= rates_last_update_date[symbol_interest]: 
52            if self.Securities[symbol_interest].Price != 0:
53                rates[symbol] = self.Securities[symbol_interest].Price
54  
55# sort funding rate
56if len(rates) >= self.quantile:
57    sorted_rates:List[Symbols] = sorted(rates, key=rates.get)
58    quantile:int = len(sorted_rates) // self.quantile
59    long:List[str] = sorted_rates[-quantile:]
60    short:List[str] = sorted_rates[:quantile]
61
62    # trade execution
63    invested:List[Symbol] = [x.Key for x in self.Portfolio if x.Value.Invested]
64    for price_symbol in invested:
65        if price_symbol not in long + short:
66            self.Liquidate(price_symbol)
67
68    for symbol in long:
69        self.SetHoldings(symbol, self.percentage_traded / len(long))
70
71    for symbol in short:
72        self.SetHoldings(symbol, -self.percentage_traded / len(short))
73
74def Selection(self):
75self.rebalance_flag = True