Conditional FX Correlation Risk

1from AlgorithmImports import *
2import data_tools
3import statsmodels.api as sm
4# endregion
5
6class ConditionalFXCorrelationRisk(QCAlgorithm):
7
8def Initialize(self):
9self.SetStartDate(2000, 7, 1)
10self.SetCash(100000)
11
12self.quantpedia_futures:bool = False
13self.leverage:int = 5
14
15self.tickers:list[str] = [
16    'CME_AD1', 'CME_BP1', 'CME_NE1',
17    'CME_CD1', 'CME_SF1', 'CME_JY1'
18]
19
20if not self.quantpedia_futures:
21    self.tickers = [
22        'AUDUSD', 'GBPUSD', 'NZDUSD', 'USDCAD',
23        'USDCHF', 'USDCNH', 'USDCZK', 'USDDKK',
24        'USDHUF', 'USDINR', 'USDJPY', 'USDNOK',
25        'USDPLN', 'USDSAR', 'USDSEK', 'USDTHB',
26        'USDTRY',
27    ]
28
29self.corr_period:int = 66 + 1       # need n days of daily closes
30self.regression_period:int = 36     # need m monthly values
31self.min_prices:int = 15            # need l daily prices in one month
32self.max_missing_days:int = 15
33
34self.quantile:int = 10
35self.traded_currency_count:int = 2
36
37self.fxc_beta_index:int = 1
38self.prev_fxc:float = None
39self.fxc_delta_values:RollingWindow = RollingWindow[float](self.regression_period)
40self.data:dict[Symbol, SymbolData] = {}
41
42for ticker in self.tickers:
43    security = self.AddData(data_tools.QuantpediaFutures, ticker, Resolution.Daily) if self.quantpedia_futures \
44            else self.AddForex(ticker, Resolution.Daily, Market.Oanda)
45    
46    if self.quantpedia_futures:
47        security.SetFeeModel(data_tools.CustomFeeModel())
48    security.SetLeverage(self.leverage)
49    
50    symbol:Symbol = security.Symbol
51    self.data[symbol] = data_tools.SymbolData(self.corr_period, self.regression_period)
52
53self.recent_month:int = -1
54
55def OnData(self, data: Slice):
56curr_date:datetime.date = self.Time.date()
57
58# update daily prices
59for symbol, symbol_obj in self.data.items():
60    if symbol in data and data[symbol] and data[symbol].Value != 0:
61        symbol_obj.update_prices(data[symbol].Value, curr_date)
62
63# rebalance monthly
64if self.recent_month == self.Time.month:
65    return
66self.recent_month = self.Time.month
67
68corr_ready_currencies:list[Symbol] = []
69for symbol, symbol_obj in self.data.items():
70    # reset currencies, which prices stopped coming, because data are required consecutive
71    if symbol_obj.data_still_coming(curr_date, self.max_missing_days) and symbol_obj.curr_monthly_prices_ready(self.min_prices):
72        symbol_obj.update_monthly_returns()
73    else:
74        # monthly prices have to be consecutive for regression
75        symbol_obj.reset_prices_and_monthly_returns()
76
77    # filter only currencies, which prices are ready for correlation calculation
78    if symbol_obj.correlation_prices_ready():
79        corr_ready_currencies.append(symbol)
80
81    symbol_obj.reset_curr_month_prices()
82
83total_ready_currencies:int = len(corr_ready_currencies)
84correlations:list[float] = []
85
86for i in range(total_ready_currencies):
87    symbol1:Symbol = corr_ready_currencies[i]
88    returns1:list[float] = self.data[symbol1].get_daily_returns()
89
90    for j in range(i + 1, total_ready_currencies, 1):
91        symbol2:Symbol = corr_ready_currencies[j]
92        returns2:list[float] = self.data[symbol2].get_daily_returns()
93
94        correlation:float = np.corrcoef(returns1, returns2)[0][1]
95        correlations.append(correlation)
96
97if len(correlations) < self.quantile:
98    # regression data has to be consecutive
99    self.fxc_delta_values.Reset()
100    self.prev_fxc = None
101    self.Liquidate()
102    return
103
104correlations.sort()
105quantile:int = int(len(correlations) / self.quantile)
106top_part_sum:float = sum(correlations[-quantile:])
107bottom_part_sum:float = sum(correlations[:quantile])
108
109fxc:float = top_part_sum - bottom_part_sum
110if self.prev_fxc != None: # previous value of FXC factor is required in FXC delta calculation
111    # take the innovation part of FXC
112    fxc_delta:float = fxc - self.prev_fxc
113    # fxc_delta:float = fxc / self.prev_fxc - 1
114    self.fxc_delta_values.Add(fxc_delta)
115self.prev_fxc = fxc
116
117if not self.fxc_delta_values.IsReady:
118    self.Liquidate()
119    return
120
121beta_fxc_by_symbols:dict[Symbol, float] = {}
122regression_x:list[float] = list(self.fxc_delta_values)
123
124for symbol, symbol_obj in self.data.items():
125    if not symbol_obj.monthly_returns_ready():
126        continue
127
128    regression_y:list[float] = symbol_obj.get_monthly_returns()
129
130    regression_model = self.MultipleLinearRegression(regression_x, regression_y)
131
132    beta_fxc:float = regression_model.params[self.fxc_beta_index]
133
134    beta_fxc_by_symbols[symbol] = beta_fxc
135
136if len(beta_fxc_by_symbols) < (self.traded_currency_count * 2):
137    self.Liquidate()
138    return
139
140# the low (high) beta currency pairs are in the long (short) position
141sorted_by_beta_fxc:list[Symbol] = [x[0] for x in sorted(beta_fxc_by_symbols.items(), key=lambda item: item[1])]
142long_leg:list[Symbol] = sorted_by_beta_fxc[:self.traded_currency_count]
143short_leg:list[Symbol] = sorted_by_beta_fxc[-self.traded_currency_count:]
144
145# trade execution
146invested:list[Symbol] = [x.Key for x in self.Portfolio if x.Value.Invested]
147for symbol in invested:
148    if symbol not in long_leg + short_leg:
149        self.Liquidate(symbol)
150
151for symbol in long_leg:
152    weight:float = 1 / self.traded_currency_count
153
154    if symbol.Value[:3] == 'USD':
155        weight = weight * -1
156
157    self.SetHoldings(symbol, weight)
158
159for symbol in short_leg:
160    weight:float = -1 / self.traded_currency_count
161    
162    if symbol.Value[:3] == 'USD':
163        weight = weight * -1
164
165    self.SetHoldings(symbol, weight)
166
167def MultipleLinearRegression(self, x:list, y:list):
168x:np.array = np.array(x).T
169x = sm.add_constant(x)
170result = sm.OLS(endog=y, exog=x).fit()
171return result