Growth of Addresses with Balance Factor in Cryptos

1from AlgorithmImports import *
2from typing import List, Dict
3#endregion
4
5import data_tools
6
7class CrosssectionalMomentumInLargeCryptos(QCAlgorithm):
8
9def Initialize(self) -> None:
10self.SetStartDate(2015, 1, 1)
11self.SetCash(1_000_000)
12
13self.period: int = 7         # need n of daily addresses
14self.quantile: int = 5
15self.portfolio_percentage: float = .1
16self.leverage: int = 10
17
18self.cryptos: List[str] = [
19    "ANT",  # Aragon
20    "BAT",  # Basic Attention Token
21    "BTC",  # Bitcoin
22    "DAI",  # Dai
23    "DASH", # Dash
24    "DOGE", # Dogecoin
25    "EOS",  # EOS
26    "ETH",  # Ethereum
27    "FUN",  # FUN Token
28    "LTC",  # Litecoin
29    "MKR",  # Maker
30    "OMG",  # OMG Network
31    "SNT",  # Status
32    "ZEC",  # Zcash
33    "ZRX"   # Ox
34]
35
36self.data: Dict[str, data_tools.SymbolData] = {}
37self.weight: Dict[str, float] = {}
38self.SetBrokerageModel(BrokerageName.Bitfinex)
39
40for crypto in self.cryptos:
41    crypto_symbol: str = crypto + 'USD'
42    data: Securities = self.AddCrypto(crypto_symbol, Resolution.Daily, Market.Bitfinex)
43    data.SetLeverage(self.leverage)
44    
45    network_symbol: Symbol = self.AddData(data_tools.CryptoNetworkData, crypto, Resolution.Daily).Symbol
46    address_symbol: Symbol = self.AddData(data_tools.CryptoAddressesData, crypto, Resolution.Daily).Symbol
47    self.data[crypto_symbol] = data_tools.SymbolData(network_symbol, address_symbol, self.period)
48
49self.rebalance_flag: bool = False
50self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
51self.Schedule.On(self.DateRules.WeekStart("BTCUSD"), self.TimeRules.At(9, 30), self.Rebalance)
52
53def OnData(self, data: Slice) -> None:
54curr_date: datetime.date = self.Time.date()
55
56# daily updating of crypto prices and market capitalization(CapMrktCurUSD)
57for crypto, symbol_obj in self.data.items():
58    network_symbol: Symbol = symbol_obj.network_symbol
59    address_symbol: Symbol = symbol_obj.address_symbol
60    
61    if network_symbol in data and data[network_symbol]:
62        # get market capitalization
63        cap_mrkt_cur_usd: float = data[network_symbol].Value
64        if cap_mrkt_cur_usd != 0.:
65            self.data[crypto].update_cap(cap_mrkt_cur_usd, curr_date)
66        
67    if address_symbol in data and data[address_symbol]:
68        # get value of total with balance addresses 
69        addresses_count: int = data[address_symbol].Value
70        if addresses_count != 0.:
71            self.data[crypto].update_addresses_count_values(addresses_count, curr_date)
72
73if not self.rebalance_flag:
74    return
75
76# trade execution
77invested: List[str] = [x.Key.Value for x in self.Portfolio if x.Value.Invested]
78for ticker in invested:
79    if ticker not in self.weight:
80        self.Liquidate(ticker)
81
82for ticker, w in self.weight.items():
83    if ticker in data and data[ticker]:
84        self.SetHoldings(ticker, w * self.portfolio_percentage)
85
86self.rebalance_flag = False
87self.weight.clear()
88
89def Rebalance(self) -> None:
90self.rebalance_flag = True
91
92curr_date: datetime.date = self.Time.date()
93crypto_data_last_update_date: Dict[Symbol, datetime.date] = data_tools.CryptoNetworkData.get_last_update_date()
94crypto_addresses_last_update_date: Dict[Symbol, datetime.date] = data_tools.CryptoAddressesData.get_last_update_date()
95
96address_growth: Dict[str, float] = {}
97
98for crypto, symbol_obj in self.data.items():
99    network_symbol: Symbol = symbol_obj.network_symbol
100    address_symbol: Symbol = symbol_obj.address_symbol
101
102    if network_symbol not in crypto_data_last_update_date or address_symbol not in crypto_addresses_last_update_date:
103        continue
104
105    # crypto doesn't have enough data
106    if self.Securities[network_symbol].GetLastData() and self.Time.date() > crypto_data_last_update_date[network_symbol] or \
107        self.Securities[address_symbol].GetLastData() and self.Time.date() > crypto_addresses_last_update_date[address_symbol]:
108        continue
109
110    if symbol_obj.is_ready():
111        # calculate address growth for current crypto
112        if symbol_obj.cap_mrkt_cur_usd != 0:
113            address_growth[crypto] = symbol_obj.address_growth()
114
115# not enough cryptos for selection    
116if len(address_growth) < self.quantile:
117    self.Liquidate()    
118    return
119
120# perform selection
121quantile: int = int(len(address_growth) / self.quantile)
122sorted_by_perf: List[str] = [x[0] for x in sorted(address_growth.items(), key=lambda item: item[1])]
123
124# long highest quantile
125long: List[str] = sorted_by_perf[-quantile:]
126# short lowest quantile
127short: List[str] = sorted_by_perf[:quantile]
128
129# value weighting
130for i, portfolio in enumerate([long, short]):
131    mc_sum:float = sum(list(map(lambda ticker: self.data[ticker].cap_mrkt_cur_usd, portfolio)))
132    for ticker in portfolio:
133        self.weight[ticker] = ((-1) ** i) * self.data[ticker].cap_mrkt_cur_usd / mc_sum