Financial Uncertainty Explains Cryptocurrency Returns

1from AlgorithmImports import *
2from collections import deque
3from dateutil.relativedelta import relativedelta
4from pandas.core.frame import DataFrame
5import data_tools
6import statsmodels.api as sm
7# endregionx
8
9class FinancialUncertaintyExplainsCryptocurrencyReturns(QCAlgorithm):
10
11def Initialize(self):
12self.SetStartDate(2015, 1, 1)
13self.SetCash(100000)
14
15self.period:int = 24
16self.portion:float = 0.1
17self.rebalance_hour:int = 0 # rebalance at 00:00
18self.portfolio_percentage:float = .1
19self.quantile:int = 3
20# self.leverage:int = 2
21
22self.cryptos:Dict[str, str] = {
23    "XLMUSD": "XLM",    # Stellar
24    "XMRUSD": "XMR",    # Monero
25    "XRPUSD": "XRP",    # XRP
26    "ADAUSD": "ADA",    # Cardano
27    "DOTUSD": "DOT",    # Polkadot
28    "UNIUSD": "UNI",    # Uniswap
29    "LINKUSD": "LINK",  # Chainlink
30    "ANTUSD": "ANT",    # Aragon
31    "BATUSD": "BAT",    # Basic Attention Token
32    "BTCUSD": "BTC",    # Bitcoin
33    "BTGUSD": "BTG",    # Bitcoin Gold
34    "DASHUSD": "DASH",  # Dash
35    "DGBUSD": "DGB",    # Dogecoin
36    "ETCUSD": "ETC",    # Ethereum Classic
37    "ETHUSD": "ETH",    # Ethereum
38    # "FUNUSD": "FUN",    # FUN Token
39    "LTCUSD": "LTC",    # Litecoin
40    "MKRUSD": "MKR",    # Maker
41    "NEOUSD": "NEO",    # Neo
42    "PAXUSD": "PAX",    # Paxful
43    "SNTUSD": "SNT",    # Status
44    "TRXUSD": "TRX",    # Tron
45    "XRPUSD": "XRP",    # XRP
46    "XTZUSD": "XTZ",    # Tezos
47    "XVGUSD": "XVG",    # Verge
48    "ZECUSD": "ZEC",    # Zcash
49    "ZRXUSD": "ZRX"     # Ox
50}
51
52self.data:Dict[Symbol, SymbolData] = {}
53
54# data subscription
55for pair, ticker in self.cryptos.items():
56    data = self.AddCrypto(pair, Resolution.Hour, Market.Bitfinex)
57    data.SetFeeModel(data_tools.CustomFeeModel())
58    # data.SetLeverage(self.leverage)
59    network_data = self.AddData(data_tools.DailyCustomData, ticker, Resolution.Daily).Symbol
60
61    self.data[data.Symbol] = data_tools.SymbolData(network_data, self.period)
62
63self.fin_uncertainty:Symbol = self.AddData(data_tools.FinancialUncertaintyIndex, 'FIN_UN', Resolution.Daily).Symbol
64
65self.rebalance_flag:bool = False
66self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
67
68self.current_month:int = -1
69
70def OnData(self, data: Slice):
71# monthly rebalance
72if self.Time.month == self.current_month:
73    return
74
75if not self.Time.hour == self.rebalance_hour:
76    return
77self.current_month = self.Time.month
78
79# update monthly price
80for symbol, symbol_data in self.data.items():
81    network_symbol:Symbol = symbol_data._network_symbol
82
83    if symbol in data and data[symbol]:
84        self.data[symbol].update(data[symbol].Close, self.Time.date())
85
86    if network_symbol in data and data[network_symbol]:
87        cap_mrkt_cur_usd:float = data[network_symbol].Value
88        self.data[symbol].update_cap(cap_mrkt_cur_usd, self.Time.date())
89
90network_data_last_update_date:Dict[Symbol, datetime.date] = data_tools.DailyCustomData.get_last_update_date()
91financial_uncertainty_last_update_date:datetime.date = data_tools.FinancialUncertaintyIndex.get_last_update_date()
92
93# custom data still comming in
94if all([self.Securities[x].GetLastData() for x in list(self.data.keys())]) and any([self.Time.date() >= network_data_last_update_date[x] for x in network_data_last_update_date]) \
95    or self.Time.date() >= financial_uncertainty_last_update_date:
96    self.Liquidate()
97    return
98
99history_financial_uncertainty:DataFrame = self.History([self.fin_uncertainty], start=self.Time.date() - relativedelta(months=self.period), end=self.Time.date()).value.unstack(level=0)
100
101crypto_returns_dict:Dict[Symbol, np.ndarray] = {symbol : symbol_data.get_returns() for symbol, symbol_data in self.data.items() if symbol_data.is_ready()}
102crypto_returns:List[float] = list(zip(*[[i for i in x] for x in crypto_returns_dict.values()]))
103
104if len(crypto_returns) == 0:
105    return
106
107# run stock regression
108y:np.ndarray = np.array(crypto_returns)
109x:np.ndarray = history_financial_uncertainty[1:].values
110model = self.multiple_linear_regression(x, y)
111beta_values:np.ndarray = model.params[1]
112
113# store betas
114beta_by_symbol:Dict[Symbol, float] = {sym : beta_values[n] for n, sym in enumerate(list(crypto_returns_dict.keys()))}
115
116long:List[Symbol] = []
117short:List[Symbol] = []
118
119# sort by beta and divide into quantiles
120if len(beta_by_symbol) >= self.quantile:
121    sorted_gamma:List[Symbol] = sorted(beta_by_symbol.items(), key=lambda x:x[1])
122    quantile:int = int(len(beta_by_symbol) / self.quantile)
123    long = [symbol for symbol, beta in sorted_gamma][:quantile]
124    short = [symbol for symbol, beta in sorted_gamma][-quantile:]
125
126# weight:Dict[Symbol, float] = {}
127
128# total_cap_long:float = sum([symbol_data._cap_mrkt_cur_usd for symbol, symbol_data in self.data.items() if symbol in long])
129# for symbol in long:
130#     weight[symbol] = self.data[symbol]._cap_mrkt_cur_usd / total_cap_long
131
132# total_cap_short:float = sum([symbol_data._cap_mrkt_cur_usd for symbol, symbol_data in self.data.items() if symbol in short])
133# for symbol in short:
134#     weight[symbol] = -self.data[symbol]._cap_mrkt_cur_usd / total_cap_short
135
136invested = [x.Key for x in self.Portfolio if x.Value.Invested]
137for symbol in invested:
138    if symbol not in long + short:
139        self.Liquidate(symbol)
140
141# trade execution
142# for symbol, w in weight.items():
143#     if symbol in data and data[symbol]:
144#         self.SetHoldings(symbol, self.portfolio_percentage * w)
145
146for symbol in long:
147    if symbol in data and data[symbol]:
148        self.SetHoldings(symbol, 1/len(long) * self.portfolio_percentage)
149
150for symbol in short:
151    if symbol in data and data[symbol]:
152        self.SetHoldings(symbol, -1/len(short) * self.portfolio_percentage)
153
154def multiple_linear_regression(self, x:np.ndarray, y:np.ndarray):
155x = sm.add_constant(x, has_constant='add')
156result = sm.OLS(endog=y, exog=x).fit()
157return result