Hedging Factor in Cryptocurrencies

1from AlgorithmImports import *
2import numpy as np
3from data_tools import CommitmentsOfTraders, CryptoCotPair
4from typing import Tuple, List
5import statsmodels.api as sm
6#endregion
7
8class HedgingFactorinCryptocurrencies(QCAlgorithm):
9
10def Initialize(self):
11self.SetStartDate(2015, 1, 1)
12self.SetCash(1000000)
13
14self.leverage:int = 5
15
16# subscribe crypto and COT data
17data:Crypto = self.AddCrypto('BTCUSD', Resolution.Minute, Market.Bitfinex)
18data.SetLeverage(self.leverage)
19crypto_symbol:Symbol = data.Symbol
20
21cot_symbol:Symbol = self.AddData(CommitmentsOfTraders, 'QBT', Resolution.Daily).Symbol
22
23data:Equity = self.AddEquity('BIL', Resolution.Minute)
24data.SetLeverage(self.leverage)
25self.t_bill:Symbol = data.Symbol
26
27self.risk_aversion:float = 3.
28self.allocation_limits:List[float] = [-0.5, 1.5]
29
30# weekly hedging factor data
31hedging_factor_w_period:int = 8
32price_w_period:int = hedging_factor_w_period + 1
33self.SetWarmup(max(price_w_period, hedging_factor_w_period) * 7, Resolution.Daily)
34
35max_missing_crypto_days:int = 5
36max_missing_cot_days:int = 7
37
38self.crypto_cot_pair:CryptoCotPair = CryptoCotPair(crypto_symbol,               \
39                                                    cot_symbol,                 \
40                                                    price_w_period,             \
41                                                    hedging_factor_w_period,    \
42                                                    max_missing_crypto_days,    \
43                                                    max_missing_cot_days
44                                                    )
45
46def OnData(self, data:Slice) -> None:
47# COT data is present in the algo
48if self.crypto_cot_pair._cot_symbol in data and data[self.crypto_cot_pair._cot_symbol] and \
49    self.crypto_cot_pair._crypto_symbol in data and data[self.crypto_cot_pair._crypto_symbol]:
50
51    # update weekly COT data
52    cot_data = data[self.crypto_cot_pair._cot_symbol]
53    comm_hedgers_interest:float = cot_data.GetProperty("COMMERCIAL_HEDGER_LONG") + cot_data.GetProperty("COMMERCIAL_HEDGER_SHORT")
54    if comm_hedgers_interest != 0:
55        hedging_factor_value:float = cot_data.GetProperty("COMMERCIAL_HEDGER_SHORT") / comm_hedgers_interest
56
57        price:float = data[self.crypto_cot_pair._crypto_symbol].Value
58
59        self.crypto_cot_pair.update_data(price, hedging_factor_value)
60
61        if self.crypto_cot_pair.is_ready():
62            x:Tuple[np.ndarray, np.ndarray] = self.crypto_cot_pair.get_regression_data()
63            model = self.multiple_linear_regression(x[1][:-1], x[0][1:])
64            forecast_return:float = model.predict([1, x[1][-1]])[0]
65            forecast_variance:float = np.std(x[0]) ** 2 # * np.sqrt(52)
66            w_t:float = (1. / self.risk_aversion) * (forecast_return / forecast_variance)
67            w_t = min(max(w_t, self.allocation_limits[0]), self.allocation_limits[1])
68            t_bill_w:float = 1. - w_t
69
70            self.SetHoldings(self.crypto_cot_pair._crypto_symbol, w_t)
71            self.SetHoldings(self.t_bill, t_bill_w)
72else:
73    # COT data is still comming in
74    if not self.crypto_cot_pair.cot_updated(self):
75        self.crypto_cot_pair.reset_data()
76        self.Liquidate()
77
78def multiple_linear_regression(self, x:np.ndarray, y:np.ndarray):
79x = np.array(x).T
80x = sm.add_constant(x)
81result = sm.OLS(endog=y, exog=x).fit()
82return result