Pairs Trading in Cryptocurrencies

1from AlgorithmImports import *
2from numpy import floor
3from datetime import datetime
4from itertools import combinations
5# endregion
6
7class PairsTradinginCryptocurrencies(QCAlgorithm):
8
9def Initialize(self):
10self.SetStartDate(2018, 1, 1)
11self.SetCash(100000)
12
13# formation-trading ratio of 2:1
14self.formation_period:int = int((datetime.now() - self.Time).days / 3 * 2)
15self.z_score_period:int = 6 * 21
16
17self.market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
18
19self.crypto_tickers:List[str] = [
20    'LINKUSD', 'QTUMUSD', 'NEOUSD', 'OMGUSD', 'BATUSD', 'ETHUSD', 'BNTUSD', 'XLMUSD', 'ADAUSD', 'XRPUSD',
21    'EOSUSD', 'ETCUSD', 'ZECUSD', 'DASHUSD', 'XMRUSD', 'TRXUSD', 'ENJUSD', 'MANAUSD' # 'BNBUSD'
22]
23
24self.possible_pairs = set(combinations(self.crypto_tickers, 2))
25
26self.pair_count:int = 6
27self.leverage:int = 2
28self.price:Dict[str, RollingWindow] = {}
29self.percentage_traded:float = .2
30
31self.threshold:float = 2.
32self.pair_z_score:Dict[TickerPair, RollingWindow] = {}
33
34self.SetWarmup(self.formation_period, Resolution.Daily)
35self.traded_pairs:List[TickerPair] = list()
36
37# data subscription
38for ticker in self.crypto_tickers:
39    data = self.AddCrypto(ticker, Resolution.Daily, Market.Bitfinex)
40    data.SetLeverage(self.leverage)
41
42    self.price[ticker] = RollingWindow[float](self.formation_period)
43
44self.recent_month:int = -1
45
46def OnData(self, data: Slice) -> None:
47# store daily prices
48for ticker in self.crypto_tickers:
49    if ticker in data and data[ticker]:
50        self.price[ticker].Add(data[ticker].Value)
51
52if self.IsWarmingUp: return
53
54for pair in self.traded_pairs:
55    price_i:np.ndarray = np.array(list(self.price[pair._ticker_i])[::-1])
56    price_j:np.ndarray = np.array(list(self.price[pair._ticker_j])[::-1])
57
58    St:np.ndarray = price_i - price_j
59    
60    # price residuals, rolling mean and std
61    St_rolling:np.ndarray = [St[i - self.z_score_period : i] for i in range(self.z_score_period, self.formation_period)]
62    Dt_rolling:np.ndarray = [np.mean(st) for st in St_rolling]
63    STDt_rolling:np.ndarray = [np.std(st) for st in St_rolling]
64
65    # z-score
66    Zt:np.ndarray = [(st[-1] - Dt_rolling[i]) / STDt_rolling[i] for i, st in enumerate(St_rolling)]
67
68    Zt_mean:float = np.mean(Zt)
69    Zt_std:float = np.std(Zt)
70    
71    quantity_i:float = floor((self.Portfolio.TotalPortfolioValue * self.percentage_traded) / self.pair_count / price_i[-1])
72    quantity_j:float = floor((self.Portfolio.TotalPortfolioValue * self.percentage_traded) / self.pair_count / price_j[-1])
73
74    # if the z-score touches the threshold from bellow, the spread is overpriced and shorted by selling coin i and buying coin j
75    upper_threshold:float = Zt_mean + (self.threshold * Zt_std)
76    lower_threshold:float = Zt_mean - (self.threshold * Zt_std)
77
78    if Zt[-1] > upper_threshold:
79        if pair.spread_is_bought():
80            self.liquidate_spread(pair)
81        else:
82            if not pair.spread_is_sold():
83                self.trade_spread(pair, -quantity_i, quantity_j)
84        
85    # when threshold is hit from above, then the portfolio value is below its long-run value so that the spread is bought, which means buying coin i and selling coin j
86    elif Zt[-1] < lower_threshold:
87        if pair.spread_is_sold():
88            self.liquidate_spread(pair)
89        else:
90            if not pair.spread_is_bought():
91                self.trade_spread(pair, quantity_i, -quantity_j)
92    
93    elif Zt[-1] < Zt_mean and Zt[-1] > lower_threshold:
94        if pair.spread_is_sold():
95            self.liquidate_spread(pair)
96
97    elif Zt[-1] > Zt_mean and Zt[-1] < upper_threshold:
98        if pair.spread_is_bought():
99            self.liquidate_spread(pair)
100
101# find new pairs to trade
102if len([x for x in self.crypto_tickers if self.price[x].IsReady]) < self.pair_count:
103    return
104else:
105    if len(self.traded_pairs) != 0:
106        return
107
108ssd_by_pair:Dict[TickerPair, float] = {}
109
110for ticker_i, ticker_j in self.possible_pairs:
111    if not self.price[ticker_i].IsReady or not self.price[ticker_j].IsReady: continue
112
113    price_i:np.ndarray = np.array(list(self.price[ticker_i])[::-1])
114    price_j:np.ndarray = np.array(list(self.price[ticker_j])[::-1])
115    St:np.ndarray = price_i - price_j
116    ssd_by_pair[TickerPair(ticker_i, ticker_j)] = sum(St ** 2)
117
118if len(ssd_by_pair) != 0:
119    sorted_by_ssd:List[TickerPair] = sorted(ssd_by_pair, key=ssd_by_pair.get, reverse=True)
120    self.traded_pairs = sorted_by_ssd[-self.pair_count:]
121
122def trade_spread(self, pair, quantity_i:float, quantity_j:float) -> None:
123self.MarketOrder(pair._ticker_i, quantity_i)
124self.MarketOrder(pair._ticker_j, quantity_j)
125pair.set_quantity(quantity_i, quantity_j)
126
127def liquidate_spread(self, pair) -> None:
128self.MarketOrder(pair._ticker_i, -pair._quantity_i)
129self.MarketOrder(pair._ticker_j, -pair._quantity_j)
130pair.set_quantity(0., 0.)
131
132class TickerPair():
133def __init__(self, ticker_i:str, ticker_j:str) -> None:
134self._ticker_i:str = ticker_i
135self._ticker_j:str = ticker_j
136
137self._quantity_i:float = 0.
138self._quantity_j:float = 0.
139
140def set_quantity(self, quantity_i:float, quantity_j:float) -> None:
141self._quantity_i = quantity_i
142self._quantity_j = quantity_j
143
144def spread_is_bought(self) -> bool:
145return self._quantity_i > 0. and self._quantity_j < 0.
146
147def spread_is_sold(self) -> bool:
148return self._quantity_i < 0. and self._quantity_j > 0.