Geopolitical Risk and the Cross-Section of Cryptocurrency Returns

1from AlgorithmImports import *
2import data_tools
3from typing import List, Dict
4# endregion
5
6class CrawlingYellowBarracuda(QCAlgorithm):
7
8def Initialize(self) -> None:
9self.SetStartDate(2015, 1, 1)
10self.SetCash(1000000)
11
12self.period: int = 21 + 1 # need n of daily data
13self.quantile: int = 5
14self.leverage: int = 5
15self.portfolio_percentage: float = .5
16
17cryptos: Dict[str, str] = {
18    "ANTUSD": "ANT", # Aragon
19    "BATUSD": "BAT", # Basic Attention Token
20    "BTCUSD": "BTC", # Bitcoin
21    "BTGUSD": "BTG", # Bitcoin Gold
22    "DAIUSD": "DAI", # Dai
23    "DGBUSD": "DGB", # Dogecoin
24    "EOSUSD": "EOS", # EOS
25    "ETCUSD": "ETC", # Ethereum Classic
26    "ETHUSD": "ETH", # Ethereum
27    "FUNUSD": "FUN", # FUNToken
28    "LTCUSD": "LTC", # Litecoin
29    "MKRUSD": "MKR", # Maker
30    "NEOUSD": "NEO", # Neo
31    "OMGUSD": "OMG", # OMG Network
32    "SNTUSD": "SNT", # Status
33    "TRXUSD": "TRX", # Tron
34    "XLMUSD": "XLM", # Stellar
35    "XMRUSD": "XMR", # Monero
36    "XRPUSD": "XRP", # XRP
37    "XTZUSD": "XTZ", # Tezos
38    "XVGUSD": "XVG", # Verge
39    "ZECUSD": "ZEC", # Zcash
40    "ZRXUSD": "ZRX", # Ox
41}
42
43self.data: Dict[str, data_tools.SymbolData] = {}
44
45self.SetBrokerageModel(BrokerageName.Bitfinex)
46
47for crypto, ticker in cryptos.items():
48    # GDAX is coinmarket, but it doesn't support this many cryptos, so we choose Bitfinex
49    data: Securities = self.AddCrypto(crypto, Resolution.Daily, Market.Bitfinex)
50    data.SetFeeModel(data_tools.CustomFeeModel())
51    data.SetLeverage(self.leverage)
52    
53    network_symbol: Symbol = self.AddData(data_tools.CryptoNetworkData, ticker, Resolution.Daily).Symbol
54    
55    self.data[crypto] = data_tools.SymbolData(network_symbol, self.period)
56
57self.geo_risk_index: Symbol = self.AddData(data_tools.QuantpediaGeopoliticalRisk, 'GeopoliticalRiskIndex', Resolution.Daily).Symbol
58self.geo_risk_index_values: RollingWindow = RollingWindow[float](self.period)
59self.geo_risk_beta_value_index: int = 1
60
61self.value_weighted: bool = True
62self.selection_flag: bool = False
63self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
64self.Schedule.On(self.DateRules.WeekStart('BTCUSD'), self.TimeRules.At(9, 30), self.Selection)
65
66def OnData(self, data: Slice) -> None:
67curr_date: datetime.date = self.Time.date()
68crypto_data_last_update_date: Dict[Symbol, datetime.date] = data_tools.CryptoNetworkData.get_last_update_date()
69qp_data_last_update_date: Dict[Symbol, datetime.date] = data_tools.QuantpediaGeopoliticalRisk.get_last_update_date()
70
71# daily updating of crypto prices and market capitalization(CapMrktCurUSD)
72if self.Securities[self.geo_risk_index].GetLastData() and self.Time.date() <= qp_data_last_update_date[self.geo_risk_index]:
73    for crypto, symbol_obj in self.data.items():
74    # if self.geo_risk_index in data and data[self.geo_risk_index]:
75        network_symbol:Symbol = symbol_obj.network_symbol
76        
77        if data.ContainsKey(crypto):
78            price: float = data[crypto].Value
79            self.data[crypto].update_prices(price)
80
81            # GPR_value:float = data[self.geo_risk_index].Value
82            GPR_value:float = self.Securities[self.geo_risk_index].Price
83            self.geo_risk_index_values.Add(GPR_value)
84        
85        if data.ContainsKey(network_symbol):
86            cap_mrkt_cur_usd: float = data[network_symbol].Value
87            self.data[crypto].update_cap(cap_mrkt_cur_usd)
88else:
89    self.geo_risk_index_values.Reset()
90
91# weekly rebalance
92if not self.selection_flag:
93    return
94self.selection_flag = False
95
96if not self.geo_risk_index_values.IsReady:
97    self.Liquidate()
98    return
99
100# calculate monthly performance series
101monthly_returns_by_symbol: Dict[str, float] = {}
102for crypto, symbol_obj in self.data.items():
103    network_symbol: Symbol = symbol_obj.network_symbol
104    if self.Securities[network_symbol].GetLastData() and self.Time.date() > crypto_data_last_update_date[network_symbol]:
105        self.Liquidate()
106        return
107
108    if not symbol_obj.is_ready():
109        continue
110
111    monthly_returns_by_symbol[crypto] = symbol_obj.get_daily_returns()
112
113if len(monthly_returns_by_symbol) < self.quantile:
114    self.Liquidate()
115    return
116
117# create market factor
118crypto_c: int = len(monthly_returns_by_symbol)
119weights: np.ndarray = np.array([1/crypto_c] * crypto_c)
120market_factor: np.ndarray = np.matmul(np.array(list(monthly_returns_by_symbol.values())).T, weights)
121
122# create GPR factor
123GPR_index_values: np.ndarray = np.array(list(self.geo_risk_index_values))
124GPR_factor: np.ndarray = (GPR_index_values[:-1] - GPR_index_values[1:]) / GPR_index_values[1:]
125regression_x: List[np.ndarray] = [
126    GPR_factor,
127    market_factor
128]
129
130beta_by_ticker: Dict[str, float] = {}
131
132for crypto, monthly_return in monthly_returns_by_symbol.items():
133    regression_y: np.ndarray = monthly_return
134    regression_model: RegressionResultsWrapper = data_tools.MultipleLinearRegression(regression_x, regression_y)
135    geo_risk_beat: float = regression_model.params[self.geo_risk_beta_value_index]
136
137    beta_by_ticker[crypto] = geo_risk_beat
138
139if len(beta_by_ticker) < self.quantile:
140    self.Liquidate()
141    return
142
143# long and short selection
144quantile: int = int(len(beta_by_ticker) / self.quantile)
145sorted_by_beta: List[list[str, float]] = [x[0] for x in sorted(beta_by_ticker.items(), key=lambda item: item[1])]
146
147long_leg: List[List[str, float]] = sorted_by_beta[:quantile]
148short_leg: List[List[str, float]] = sorted_by_beta[-quantile:]
149
150# weights calculation
151weights: Dict[str, float] = {}
152
153if self.value_weighted:
154    for i, portfolio in enumerate([long_leg, short_leg]):
155        mc_sum: float = sum(list(map(lambda ticker: self.data[ticker].cap_mrkt_cur_usd, portfolio)))
156        for ticker in portfolio:
157            weights[ticker] = ((-1)**i) * self.data[ticker].cap_mrkt_cur_usd / mc_sum
158
159else:
160    for i, portfolio in enumerate([long_leg, short_leg]):
161        for ticker in portfolio:
162            weights[ticker] = ((-1) ** i) / len(portfolio)
163
164# trade execution
165portfolio: List[PortfolioTarget] = [PortfolioTarget(ticker, self.portfolio_percentage * w) for ticker, w in weights.items() if ticker in data and data[ticker]]
166self.SetHoldings(portfolio, True)
167
168def Selection(self) -> None:
169self.selection_flag = True