Price-based Value in Cryptocurrencies

1from AlgorithmImports import *
2import data_tools
3from typing import List, Dict
4#endregion
5
6class PricebasedValueinCryptocurrencies(QCAlgorithm):
7
8def Initialize(self) -> None:
9self.SetStartDate(2015, 1, 1)
10self.SetCash(100000)
11
12self.period: int = 52 * 7 
13self.quantiles: int = 5
14self.percentage_traded: int = .2
15self.leverage: int = 10
16
17self.cryptos: Dict[str, str] = {
18    "ANTUSD": "ANT", # Aragon
19    "BATUSD": "BAT", # Basic Attention Token
20    "BTCUSD": "BTC", # Bitcoin
21    "DGBUSD": "DGB", # Dogecoin
22    "EOSUSD": "EOS", # EOS
23    "ETCUSD": "ETC", # Ethereum Classic
24    "ETHUSD": "ETH", # Ethereum
25    "FUNUSD": "FUN", # FUN Token
26    "LTCUSD": "LTC", # Litecoin
27    "MKRUSD": "MKR", # Maker
28    "NEOUSD": "NEO", # Neo
29    "OMGUSD": "OMG", # OMG Network
30    "SNTUSD": "SNT", # Status
31    "TRXUSD": "TRX", # Tron
32    "XLMUSD": "XLM", # Stellar
33    "XMRUSD": "XMR", # Monero
34    "XRPUSD": "XRP", # XRP
35    "XTZUSD": "XTZ", # Tezos
36    "XVGUSD": "XVG", # Verge
37    "ZECUSD": "ZEC", # Zcash
38    "ZRXUSD": "ZRX"  # Ox
39}
40
41self.data: Dict[str, data_tools.SymbolData] = {}
42self.SetBrokerageModel(BrokerageName.Bitfinex)
43
44for crypto, ticker in self.cryptos.items():
45    data: Securities = self.AddCrypto(crypto, Resolution.Daily, Market.Bitfinex)
46    data.SetFeeModel(data_tools.CustomFeeModel())
47    data.SetLeverage(self.leverage)
48    
49    network_symbol: Symbol = self.AddData(data_tools.CryptoNetworkData, ticker, Resolution.Daily).Symbol
50    self.data[crypto] = data_tools.SymbolData(network_symbol, self.period)
51
52self.rebalance_flag: bool = False
53self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
54self.Schedule.On(self.DateRules.WeekStart('BTCUSD'), self.TimeRules.At(0,0), self.Rebalance)
55
56def OnData(self, data: Slice) -> None:
57crypto_data_last_update_date: Dict[Symbol, datetime.date] = data_tools.CryptoNetworkData.get_last_update_date()
58
59# daily updating of crypto prices and market capitalization(CapMrktCurUSD)
60for crypto, symbol_obj in self.data.items():
61    network_symbol: Symbol = symbol_obj.network_symbol
62    
63    if crypto in data.Bars and data[crypto]:
64        # get crypto price
65        price: float = data.Bars[crypto].Value
66        self.data[crypto].update(price)
67    
68    if network_symbol in data and data[network_symbol]:
69        # get market capitalization
70        cap_mrkt_cur_usd: float = data[network_symbol].Value
71        self.data[crypto].update_cap(cap_mrkt_cur_usd)
72
73if not self.rebalance_flag:
74    return
75performance: Dict[str, float] = {}
76
77for crypto, symbol_obj in self.data.items():
78    if crypto in data and data[crypto]:
79        # crypto doesn't have enough data
80        if self.Securities[symbol_obj.network_symbol].GetLastData() and self.Time.date() > crypto_data_last_update_date[symbol_obj.network_symbol]:
81            self.Liquidate()
82            continue
83
84        if not symbol_obj.is_ready():
85            continue
86    
87        # calculate performance for current crypto
88        performance[crypto] = symbol_obj.performance()
89
90# not enough cryptos for selection    
91if len(performance) < self.quantiles:  
92    return
93
94self.rebalance_flag = False
95# perform selection
96quantile: int = int(len(performance) / self.quantiles)
97sorted_by_perf: List[str] = [x[0] for x in sorted(performance.items(), key=lambda item: item[1])]
98
99# long top quantile
100long: List[str] = sorted_by_perf[-quantile:]
101# short bottom quantile
102short: List[str] = sorted_by_perf[:quantile]
103
104# trade execution
105for i, portfolio in enumerate([long, short]):
106    for symbol in portfolio:
107        if symbol in data and data[symbol]:
108            self.SetHoldings(symbol, round(((-1) ** i) / len(portfolio) * self.percentage_traded, 2))
109
110def Rebalance(self) -> None:
111self.rebalance_flag = True