High-Momentum in Liquid Cryptocurrencies

1from AlgorithmImports import *
2from typing import List, Dict
3import numpy as np
4# endregion
5
6class HighMomentuminLiquidCryptocurrencies(QCAlgorithm):
7
8def Initialize(self):
9self.SetStartDate(2015, 1, 1)
10self.SetCash(100000)
11
12self.market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
13
14self.crypto_tickers:List[str] = ['BTCUSD', 'ETHUSD', 'SOLUSD', 'ADAUSD', 'XRPUSD', 'DOTUSD', 'DOGEUSD', 'LUNAUSD', 'AVAXUSD', 'UNIUSD',
15                                'LINKUSD', 'LTCUSD', 'BCHABCUSD', 'BSVUSD', 'FILUSD', 'XLMUSD', 'XTZUSD', 'NEOUSD', 'ATOMUSD', 'IOTAUSD', 
16                                'ETCUSD', 'DASHUSD', 'EGLDUSD', 'AAVEUSD', 'ENJUSD', 'EOSUSD', 'MKRUSD', 'MANAUSD', 'SNXUSD', 'FTTUSD', 
17                                'OMGUSD', 'SUSHIUSD', 'YFIUSD', 'WBTCUSD', 'XMRUSD', 'ZECUSD', 'ZRXUSD', 'XRAUSD', 'AMPLUSD', 'GRTUSD', 
18                                'DGBUSD', '1INCHUSD']
19
20self.data:Dict[Symbol, SymbolData] = {}
21
22self.week_periods:List[int] = [7, 14, 21]
23self.selection_flag:bool = False
24self.quantile:int = 5
25self.leverage:int = 2
26self.portion:float = .33
27self.percentage_traded:float = .2
28
29self.SetWarmup(self.week_periods[2], Resolution.Daily)
30
31# data subscription
32for ticker in self.crypto_tickers:
33    data = self.AddCrypto(ticker, Resolution.Daily, Market.Bitfinex)
34    data.SetLeverage(self.leverage)
35
36    self.data[ticker] = SymbolData(self.week_periods[2])
37
38self.Schedule.On(self.DateRules.WeekEnd(self.market), self.TimeRules.BeforeMarketClose(self.market), self.Selection)
39
40def OnData(self, data: Slice) -> None:
41# store daily prices
42for ticker in self.crypto_tickers:
43    if ticker in self.data:
44        if ticker in data and data[ticker]:
45            self.data[ticker].update_price(data[ticker].Close, data[ticker].High)
46
47if self.IsWarmingUp: return
48
49if not self.selection_flag:
50    return
51self.selection_flag = False
52
53hmom_by_period_long:Dict[int, List[Symbol]] = {}
54hmom_by_period_short:Dict[int, List[Symbol]] = {}
55
56# sort and divide into periods and quantiles
57for period in self.week_periods:
58    hmom:dict[Symbol, float] = {}
59    for ticker, item in self.data.items():
60        if self.data[ticker].is_ready():
61            hmom[ticker] = self.data[ticker].get_week_highmomentum(period)
62            
63    if len(hmom) >= self.quantile:
64        sorted_hmom = sorted(hmom, key=hmom.get, reverse=True)
65        quantile:int = len(sorted_hmom) // self.quantile
66        hmom_by_period_long[period] = sorted_hmom[:quantile]
67        hmom_by_period_short[period] = sorted_hmom[-quantile:]
68
69trade_quantities:Dict[Symbol, float] = {}
70
71# trade quantities calculation
72for i, hmom_lst in enumerate( [list(hmom_by_period_long.values()), list(hmom_by_period_short.values())] ):
73    for lst in hmom_lst:
74        for ticker in lst:
75            if ticker in data and data[ticker]:
76                quantity:float = ((self.Portfolio.TotalPortfolioValue / len(lst)) * self.portion) // data[ticker].Price
77                if ticker not in trade_quantities:
78                    trade_quantities[ticker] = 0
79
80                # long
81                if i == 0: trade_quantities[ticker] += quantity
82                # short
83                else: trade_quantities[ticker] -= quantity
84
85# trade execution
86stocks_invested:List[Symbol] = [x.Key.Value for x in self.Portfolio if x.Value.Invested]
87for ticker in stocks_invested:
88    if ticker not in trade_quantities:
89        self.Liquidate(ticker)
90
91for ticker, new_quantity in trade_quantities.items():
92    if self.Portfolio[ticker].Invested:
93        quantity:float = new_quantity * self.percentage_traded - self.Portfolio[ticker].Quantity
94        if abs(quantity) >= 1.:
95            self.MarketOrder(ticker, quantity)
96    else:
97        self.MarketOrder(ticker, new_quantity * self.percentage_traded)
98
99def Selection(self) -> None:
100self.selection_flag = True
101
102class SymbolData():
103def __init__(self, period:int) -> None:
104self._period:int = period
105self._daily_price:RollingWindow = RollingWindow[float](period)
106self._high_price:RollingWindow = RollingWindow[float](period)
107
108def update_price(self, price:float, high:float) -> None:
109self._daily_price.Add(price)
110self._high_price.Add(high)
111
112def is_ready(self) -> bool:
113return self._daily_price.IsReady and self._high_price.IsReady
114
115def get_week_highmomentum(self, period:int) -> float:
116prices:np.ndarray = np.array([x for x in self._daily_price])
117high:np.ndarray = np.array([x for x in self._high_price])
118return np.log(prices[0]) - np.log(max(high[:period]))