Cross-sectional Momentum in Large Cryptos

1from AlgorithmImports import *
2from typing import List, Dict
3
4class CrosssectionalMomentumInLargeCryptos(QCAlgorithm):
5
6def Initialize(self) -> None:
7self.SetStartDate(2015, 1, 1)
8self.SetCash(1_000_000)
9
10self.period: int = 14 # need n of daily prices
11self.quantile: int = 3
12self.portfolio_percentage: float = .1
13self.leverage: int = 10
14
15self.cryptos: Dict[str, str] = {
16    "ANTUSD": "ANT", # Aragon
17    "BATUSD": "BAT", # Basic Attention Token
18    "BTCUSD": "BTC", # Bitcoin
19    "BTGUSD": "BTG", # Bitcoin Gold
20    "DAIUSD": "DAI", # Dai
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.weight: Dict[str, float] = {}
43
44self.SetBrokerageModel(BrokerageName.Bitfinex)
45
46for crypto, ticker in self.cryptos.items():
47    # GDAX is coinmarket, but it doesn't support this many cryptos, so we choose Bitfinex
48    data: Securities = self.AddCrypto(crypto, Resolution.Daily, Market.Bitfinex)
49    data.SetLeverage(self.leverage)
50    
51    network_symbol: Symbol = self.AddData(data_tools.CryptoNetworkData, ticker, Resolution.Daily).Symbol
52    self.data[crypto] = data_tools.SymbolData(network_symbol, self.period)
53
54self.rebalance_flag: bool = False
55self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
56self.Schedule.On(self.DateRules.WeekStart("BTCUSD"), self.TimeRules.At(0,0), self.Rebalance)
57
58def OnData(self, data: Slice) -> None:
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        if cap_mrkt_cur_usd != 0:
72            self.data[crypto].update_cap(cap_mrkt_cur_usd)
73
74if not self.rebalance_flag:
75    return
76
77# trade execution
78invested: List[str] = [x.Key.Value for x in self.Portfolio if x.Value.Invested]
79for ticker in invested:
80    if ticker not in self.weight:
81        self.Liquidate(ticker)
82
83for ticker, w in self.weight.items():
84    self.SetHoldings(ticker, w)
85
86self.rebalance_flag = False
87self.weight.clear()
88
89def Rebalance(self) -> None:
90self.rebalance_flag = True
91
92crypto_data_last_update_date: Dict[Symbol, datetime.date] = data_tools.CryptoNetworkData.get_last_update_date()
93
94performance: Dict[str, float] = {}
95
96for crypto, symbol_obj in self.data.items():
97    network_symbol: Symbol = symbol_obj.network_symbol
98    if network_symbol not in crypto_data_last_update_date:
99        continue
100
101    # crypto doesn't have enough data
102    if self.Securities[network_symbol].GetLastData() and self.Time.date() > crypto_data_last_update_date[network_symbol]:
103        self.Liquidate()
104        return
105
106    if symbol_obj.is_ready():
107        # calculate performance for current crypto
108        performance[crypto] = symbol_obj.performance()
109
110# not enough cryptos for selection    
111if len(performance) < self.quantile:
112    self.Liquidate()    
113    return
114
115# perform selection
116quantile: int = int(len(performance) / self.quantile)
117sorted_by_perf: List[str] = [x[0] for x in sorted(performance.items(), key=lambda item: item[1])]
118
119# long top quantile
120long: List[str] = sorted_by_perf[-quantile:]
121# short bottom quantile
122short: List[str] = sorted_by_perf[:quantile]
123
124# value weighting
125for i, portfolio in enumerate([long, short]):
126    mc_sum:float = sum(list(map(lambda ticker: self.data[ticker].cap_mrkt_cur_usd, portfolio)))
127    for ticker in portfolio:
128        self.weight[ticker] = ((-1) ** i) * (self.data[ticker].cap_mrkt_cur_usd / mc_sum)