Salience Theory and Cryptocurrency Returns

1from AlgorithmImports import *
2from typing import List, Dict
3
4class SalienceTheoryandCryptocurrencyReturns(QCAlgorithm):
5
6def Initialize(self):
7self.SetStartDate(2015, 1, 1)
8self.SetCash(100000)
9
10self.cryptos: Dict[str, str] = {
11    "XLMUSD": "XLM",    # Stellar
12    "XMRUSD": "XMR",    # Monero
13    "XRPUSD": "XRP",    # XRP
14    "ADAUSD": "ADA",    # Cardano
15    "DOTUSD": "DOT",    # Polkadot
16    "UNIUSD": "UNI",    # Uniswap
17    "LINKUSD": "LINK",  # Chainlink
18    "ANTUSD": "ANT",    # Aragon
19    "BATUSD": "BAT",    # Basic Attention Token
20    "BTCUSD": "BTC",    # Bitcoin
21    "BTGUSD": "BTG",    # Bitcoin Gold
22    "DAIUSD": "DAI",    # Dai
23    "DASHUSD": "DASH",  # Dash
24    "DGBUSD": "DGB",    # Dogecoin
25    "ETCUSD": "ETC",    # Ethereum Classic
26    "ETHUSD": "ETH",    # Ethereum
27    "FUNUSD": "FUN",    # FUN Token
28    "LTCUSD": "LTC",    # Litecoin
29    "MKRUSD": "MKR",    # Maker
30    "NEOUSD": "NEO",    # Neo
31    "PAXUSD": "PAX",    # Paxful
32    "SNTUSD": "SNT",    # Status
33    "TRXUSD": "TRX",    # Tron
34    "XRPUSD": "XRP",    # XRP
35    "XTZUSD": "XTZ",    # Tezos
36    "XVGUSD": "XVG",    # Verge
37    "ZECUSD": "ZEC",    # Zcash
38    "ZRXUSD": "ZRX"     # Ox
39}
40
41self.symbol_data: Dict[str, data_tools.SymbolData] = {}
42
43self.ranking_period: int = 21
44self.momentum_period: int = self.ranking_period * 2        # need n of daily prices
45self.quantile: int = 5
46self.portfolio_percentage: float = .2
47self.leverage: int = 3
48
49self.delta: float = .7
50self.theta: float = .1
51
52self.SetBrokerageModel(BrokerageName.Bitfinex)
53self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
54
55for ticker, crypto in self.cryptos.items():
56    data: Crypto = self.AddCrypto(ticker, Resolution.Daily, Market.Bitfinex)
57    data.SetFeeModel(data_tools.CustomFeeModel())
58    data.SetLeverage(self.leverage)
59    
60    network_symbol: Symbol = self.AddData(data_tools.CryptoNetworkData, crypto, Resolution.Daily).Symbol
61    self.symbol_data[ticker] = data_tools.SymbolData(network_symbol, self.momentum_period)
62  
63self.recent_month: int = -1
64
65def OnData(self, data: Slice) -> None:
66crypto_data_last_update_date: Dict[Symbol, datetime.date] = data_tools.CryptoNetworkData.get_last_update_date()
67
68# daily updating of crypto prices and market capitalization(CapMrktCurUSD)
69for crypto, symbol_obj in self.symbol_data.items():
70    network_symbol: Symbol = symbol_obj._network_symbol
71    
72    if crypto in data.Bars and data[crypto]:
73        # get crypto price
74        price: float = data.Bars[crypto].Value
75        self.symbol_data[crypto].update(price)
76    
77    if network_symbol in data and data[network_symbol]:
78        # get market capitalization
79        cap_mrkt_cur_usd: float = data[network_symbol].Value
80        self.symbol_data[crypto].update_cap(cap_mrkt_cur_usd)
81
82# monthly rebalance
83if self.recent_month == self.Time.month:
84    return
85
86prepared_symbols: List[Tuple[str, data_tools.CryptoNetworkData]] = [(crypto, symbol_obj) for crypto, symbol_obj in self.symbol_data.items() if symbol_obj.is_ready() and \
87                                                             self.Securities[symbol_obj._network_symbol].GetLastData() and self.Time.date() < crypto_data_last_update_date[symbol_obj._network_symbol] and \
88                                                             crypto in data and data[crypto]]
89ST: Dict[str, float] = {}
90
91if len(prepared_symbols) != 0:
92    self.recent_month = self.Time.month 
93    daily_returns: np.ndarray = np.array([symbol_obj.daily_returns() for _, symbol_obj in prepared_symbols])
94
95    # salience
96    mean_returns: np.ndarray = np.mean(daily_returns, axis=0)
97    salience: np.ndarray = abs(daily_returns - mean_returns) / (abs(daily_returns) + abs(mean_returns) + self.theta)
98    
99    # salience ranking
100    salience_order: np.ndarray = salience.argsort(axis=0)
101    salience_rank: np.ndarray = salience_order.argsort(axis=0)
102    pi: float = 1. / float(self.ranking_period)
103    salience_rank_sum: np.ndarray = salience_rank.sum(axis=0) * pi
104    
105    # decision weights
106    weights: np.ndarray = salience_rank / salience_rank_sum
107
108    # salience effect
109    ST = { prepared_symbols[i] : np.cov(stack)[0][1] for i, stack in enumerate(np.stack((weights, daily_returns), axis=1)) }
110
111long: List[str] = []
112short: List[str] = []
113
114if len(ST) >= self.quantile:
115    # sort the portfolio into quintiles, long the lowest quintile, short the highest
116    sorted_by_ST: List[Tuple[str, float]] = sorted(ST.items(), key=lambda x:x[1], reverse=True)
117    quantile: int = int(len(sorted_by_ST) / self.quantile)
118    long = [x[0] for x in sorted_by_ST[-quantile:]]
119    short = [x[0] for x in sorted_by_ST[:quantile]]
120
121# value weighting
122weight: Dict[str, float] = {}
123for i, portfolio in enumerate([long, short]):
124    mc_sum:float = sum(list(map(lambda x: x[1]._cap_mrkt_cur_usd, portfolio)))
125    for ticker, symbol_obj in portfolio:
126        weight[ticker] = ((-1)**i) * symbol_obj._cap_mrkt_cur_usd / mc_sum
127
128# trade execution    
129portfolio:List[PortfolioTarget] = [PortfolioTarget(symbol, self.portfolio_percentage * w) for symbol, w in weight.items() if symbol in data and data[symbol]]
130self.SetHoldings(portfolio, True)