Timing Carry Trade v2

1from AlgorithmImports import *
2import data_tools
3import numpy as np
4import pandas as pd
5import statsmodels.formula.api as sm
6#endregion
7class TimingCarryTradeV2(QCAlgorithm):
8def Initialize(self):
9self.SetStartDate(2009, 1, 1)
10self.SetCash(100000)
11self.tickers:dict[str, str] = {
12    "CME_AD1" : Futures.Currencies.AUD, # Australian Dollar Futures, Continuous Contract #1
13    "CME_BP1" : Futures.Currencies.GBP, # British Pound Futures, Continuous Contract #1
14    "CME_CD1" : Futures.Currencies.CAD, # Canadian Dollar Futures, Continuous Contract #1
15    "CME_EC1" : Futures.Currencies.EUR, # Euro FX Futures, Continuous Contract #1
16    "CME_JY1" : Futures.Currencies.JPY, # Japanese Yen Futures, Continuous Contract #1
17    "CME_MP1" : Futures.Currencies.MXN, # Mexican Peso Futures, Continuous Contract #1
18    "CME_NE1" : Futures.Currencies.NZD, # New Zealand Dollar Futures, Continuous Contract #1
19    "CME_SF1" : Futures.Currencies.CHF, # Swiss Franc Futures, Continuous Contract #1
20}
21                
22self.month_period:int = 21
23self.avg_vol_period:int = 4
24self.regression_period:int = 12
25self.max_missing_days:int = 5
26self.min_expiration_days:int = 2
27self.max_expiration_days:int = 360
28self.SetWarmUp(12 * self.month_period, Resolution.Daily) # one year warm up
29
30self.data:dict[Symbol, data_tools.SymbolData] = {}
31self.futures_data:dict[str, data_tools.FuturesData] = {}
32
33# Monthly average currency volatility.
34self.avg_vol:RollingWindow = RollingWindow[float](self.avg_vol_period)
35
36# MSCI world equity price index
37self.msci:Symbol = self.AddEquity('URTH',  Resolution.Daily).Symbol
38self.data[self.msci] = data_tools.SymbolData(4 * self.month_period) # 4 months
39
40# 12 months of regression data.
41self.regression_data:data_tools.RegressionData = data_tools.RegressionData(self.regression_period)
42
43# Raw Industrials Spot Commodity Index.
44self.dbb:Symbol = self.AddEquity('DBB',  Resolution.Daily).Symbol
45self.data[self.dbb] = data_tools.SymbolData(4 * self.month_period) # 4 months
46
47for qp_ticker, qc_ticker in self.tickers.items():
48    security:Security = self.AddData(data_tools.QuantpediaFutures, qp_ticker, Resolution.Daily)
49    security.SetFeeModel(data_tools.CustomFeeModel())
50    security.SetLeverage(5)
51    qp_symbol:Symbol = security.Symbol
52    self.data[qp_symbol] = data_tools.SymbolData(12 * self.month_period) # one year
53    # QC futures
54    future:Future = self.AddFuture(qc_ticker, Resolution.Daily, dataNormalizationMode=DataNormalizationMode.Raw)
55    future.SetFilter(timedelta(days=self.min_expiration_days), timedelta(days=self.max_expiration_days))
56    self.futures_data[future.Symbol.Value] = data_tools.FuturesData(qp_symbol)
57self.recent_month:int = -1
58self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
59def FindAndUpdateContracts(self, futures_chain, ticker) -> None:
60near_contract:FuturesContract = None
61dist_contract:FuturesContract = None
62if ticker in futures_chain:
63    contracts:list[:FuturesContract] = [contract for contract in futures_chain[ticker] if contract.Expiry.date() > self.Time.date()]
64    if len(contracts) >= 2:
65        contracts:list[:FuturesContract] = sorted(contracts, key=lambda x: x.Expiry, reverse=False)
66        near_contract = contracts[0]
67        dist_contract = contracts[1]
68self.futures_data[ticker].update_contracts(near_contract, dist_contract) 
69def OnData(self, data):
70curr_date:datetime.date = self.Time.date()
71# daily update qc future data
72if data.FutureChains.Count > 0:
73    for ticker, future_obj in self.futures_data.items():
74        # check if near contract is expired or is not initialized
75        if not future_obj.is_initialized() or \
76            (future_obj.is_initialized() and future_obj.near_contract.Expiry.date() == curr_date):
77            self.FindAndUpdateContracts(data.FutureChains, ticker)
78        # update QC futures rolling return
79        if future_obj.is_initialized():
80            near_c:FuturesContract = future_obj.near_contract
81            dist_c:FuturesContract = future_obj.distant_contract
82            if near_c.Symbol in data and data[near_c.Symbol] and dist_c.Symbol in data and data[dist_c.Symbol]:
83                raw_price1:float = data[near_c.Symbol].Value * self.Securities[ticker].SymbolProperties.PriceMagnifier
84                raw_price2:float = data[dist_c.Symbol].Value * self.Securities[ticker].SymbolProperties.PriceMagnifier
85                if raw_price1 != 0 and raw_price2 != 0:
86                    future_obj.update_roll_return(raw_price1, raw_price2, curr_date)
87# update daily prices of QP futures and indexes
88for symbol, symbol_obj in self.data.items():
89    if symbol in data and data[symbol]:
90        price:float = data[symbol].Value
91        symbol_obj.update(price, curr_date)
92# rebalacne monthly
93if self.IsWarmingUp or (self.recent_month == self.Time.month):
94    return
95self.recent_month = self.Time.month
96roll_return:dict[Symbol, float] = {}
97volatility:dict[Symbol, float] = {}
98last_update_date:Dict[str, datetime.date] = data_tools.QuantpediaFutures.get_last_update_date()
99for ticker, future_obj in self.futures_data.items():
100    qp_symbol:Symbol = future_obj.quantpedia_future
101    
102    if qp_symbol.Value in last_update_date and self.Time.date() < last_update_date[qp_symbol.Value]:
103        if self.data[qp_symbol].is_ready() and future_obj.is_ready():
104            # last month of daily closes
105            closes:list[float] = self.data[qp_symbol].get_prices(self.month_period)
106            volatility[qp_symbol] = data_tools.Volatility(closes)
107            roll_return[qp_symbol] = future_obj.get_roll_return()
108    else:
109        future_obj.reset_data()
110# volatility is required for avg_vol calculation
111if len(volatility) == 0:
112    self.regression_data.reset_data()
113    self.avg_vol.Reset()
114    self.Liquidate()
115    return
116avg_vol:float = np.mean([x[1] for x in volatility.items()])
117self.avg_vol.Add(avg_vol)
118# these data are required for regression variables
119if not self.avg_vol.IsReady or not self.data[self.msci].is_ready() or not self.data[self.dbb].is_ready():
120    self.regression_data.reset_data()
121    self.Liquidate()
122    return
123
124# change in average currency volatility.
125monthly_diff:np.ndarray = np.diff([x for x in self.avg_vol])
126# change in average currency volatility two months ago.
127vol_change_two_months:float = monthly_diff[1]
128# change in average currency volatility three months ago.
129vol_change_three_months:float = monthly_diff[2]
130# monthly change in commodity index three months ago.
131msci_monthly_returns:list[float] = self.MonthlyReturns(self.data[self.msci].prices)
132msci_change:float = msci_monthly_returns[2]
133
134# monthly change in equity index three months ago.
135dbb_monthly_returns:list[float] = self.MonthlyReturns(self.data[self.dbb].prices)
136dbb_change:float = dbb_monthly_returns[2]
137long_part:list[Symbol] = []
138short_part:list[Symbol] = []
139# regression data has to be ready
140if self.regression_data.is_ready():
141    # regression data
142    msci_changes, dbb_changes, vol_two_months_changes, vol_three_months_changes = self.regression_data.get_series()
143    for qp_symbol, roll_return_value in roll_return.items():
144        monthly_returns:list[float] = self.MonthlyReturns(self.data[qp_symbol].prices)
145        monthly_returns:pd.Series = pd.Series(monthly_returns)
146        
147        if roll_return_value > 0:
148            Y:float = self.RegressionPrediction(
149                list_of_series=[msci_changes, vol_three_months_changes, monthly_returns],
150                columns=['msci_changes', 'vol_three_months_changes', 'monthly_returns'],
151                formula='monthly_returns ~ msci_changes + vol_three_months_changes',
152                X1=msci_change,
153                X2=vol_change_three_months
154            )
155            if Y > 0:
156                long_part.append(qp_symbol)
157            
158        else:
159            Y:float = self.RegressionPrediction(
160                list_of_series=[dbb_changes, vol_two_months_changes, monthly_returns],
161                columns=['dbb_changes', 'vol_two_months_changes', 'monthly_returns'],
162                formula='monthly_returns ~ dbb_changes + vol_two_months_changes',
163                X1=dbb_change,
164                X2=vol_change_two_months
165            )
166            if Y < 0:
167                short_part.append(qp_symbol)
168    
169# update regression data for current month
170self.regression_data.update(msci_change, dbb_change, vol_change_two_months, vol_change_three_months)
171       
172# trade execution
173invested = [x.Key for x in self.Portfolio if x.Value.Invested]
174for symbol in invested:
175    if symbol not in long_part + short_part:
176        self.Liquidate(symbol)
177for i, portfolio in enumerate([long_part, short_part]):
178    for symbol in portfolio:
179        if symbol in data and data[symbol]:
180            self.SetHoldings(symbol, ((-1) ** i) / len(portfolio))
181def MonthlyReturns(self, prices_roll_window: RollingWindow) -> list:
182prices:list[float] = [x for x in prices_roll_window]
183monthly_returns:list[float] = [data_tools.Return(prices[x:x+self.month_period]) for x in range(0, len(prices), self.month_period)]
184return monthly_returns
185def RegressionPrediction(self, list_of_series:list, columns:list, formula:str, X1:pd.Series, X2:pd.Series) -> float:
186data_frame:pd.DataFrame = pd.concat(list_of_series, axis=1).dropna()
187data_frame.columns = columns
188model = sm.ols(formula=formula, data=data_frame).fit()
189
190alpha:float = model.params[0]
191beta:float = model.params[1]
192
193# Expected symbol return.
194return alpha + (beta * X1) + (beta * X2)