Timing Carry Trade with Central Banks’ Announcements

1from AlgorithmImports import *
2import data_tools
3from pandas.tseries.offsets import BDay
4from pandas.core.frame import DataFrame
5# endregion
6
7class TimingCarryTradewithCentralBanksAnnouncements(QCAlgorithm):
8
9def Initialize(self) -> None:
10self.SetStartDate(2000, 1, 1)
11self.SetCash(100000)
12
13self.market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
14self.bond:Symbol = self.AddEquity('IEF', Resolution.Daily).Symbol
15# self.dollar:Symbol = self.AddEquity('UUP', Resolution.Daily).Symbol
16
17# Cash rate source: https://fred.stlouisfed.org/series/IR3TIB01USM156N
18self.symbols:Dict[str, str] = {
19    "CME_AD1" : "IR3TIB01AUM156N",   # Australian Dollar Futures, Continuous Contract #1
20    "CME_BP1" : "LIOR3MUKM",         # British Pound Futures, Continuous Contract #1
21    "CME_CD1" : "IR3TIB01CAM156N",   # Canadian Dollar Futures, Continuous Contract #1
22    "CME_EC1" : "IR3TIB01EZM156N",   # Euro FX Futures, Continuous Contract #1
23    "CME_JY1" : "IR3TIB01JPM156N",   # Japanese Yen Futures, Continuous Contract #1
24    "CME_MP1" : "IR3TIB01MXM156N",   # Mexican Peso Futures, Continuous Contract #1
25    "CME_NE1" : "IR3TIB01NZM156N",   # New Zealand Dollar Futures, Continuous Contract #1
26    "CME_SF1" : "IR3TIB01CHM156N"    # Swiss Franc Futures, Continuous Contract #1
27}
28
29self.long:List[str] = []
30self.short:List[str] = []
31
32# load Federal Open Market Committee dates
33csv_string_file:str = self.Download('data.quantpedia.com/backtesting_data/economic/fed_days.csv')
34dates:str = csv_string_file.split('\r\n')
35self.fomc_dates:List[datetime.date] = [datetime.strptime(x, "%Y-%m-%d") + BDay(1) for x in dates]
36
37self.traded_count:int = 3       # fx carry long and short count
38self.leverage:int = 3
39self.period:int = 2
40
41# data subscription
42for symbol, rate_symbol in self.symbols.items():
43    self.AddData(data_tools.InterestRate3M, rate_symbol, Resolution.Daily)
44
45    data = self.AddData(data_tools.QuantpediaFutures, symbol, Resolution.Daily)
46    data.SetFeeModel(data_tools.CustomFeeModel())
47    data.SetLeverage(self.leverage)
48    
49self.fx_carry_flag:bool = None
50self.lag_flag:bool = False      # handles currency futures daily close data availability
51
52def OnData(self, data: Slice) -> None:
53if self.lag_flag:
54    self.lag_flag = False
55    if not self.fx_carry_flag:
56        
57        self.fx_carry_flag = None
58
59        # sell futures to gain exposure in USD
60        for symbol in self.symbols.keys():
61            if symbol in data and data[symbol]:
62                self.SetHoldings(symbol, -1 / len(self.symbols))
63  
64    if self.fx_carry_flag:
65        self.fx_carry_flag = None
66
67        for symbol in self.long:
68            self.SetHoldings(symbol, 1 / len(self.long))
69        for symbol in self.short:
70            self.SetHoldings(symbol, -1 / len(self.short))
71        
72        self.long.clear()
73        self.short.clear()
74
75# not FED day
76if self.Time.timestamp() not in [i.timestamp() for i in self.fomc_dates]:
77    return
78
79self.Liquidate()
80
81# signal check
82history:DataFrame = self.History([self.market] + [self.bond], self.period, Resolution.Daily)['close'].unstack(level=0)
83if len(history.columns) > 1:
84    if history[self.market].iloc[1] < history[self.market].iloc[0] and history[self.bond].iloc[1] < history[self.bond].iloc[0]:
85        self.fx_carry_flag = False
86    else:
87        self.fx_carry_flag = True
88
89qp_futures_last_update_date:Dict[str, datetime.date] = data_tools.QuantpediaFutures.get_last_update_date()  
90ir_last_update_date:Dict[str, datetime.date] = data_tools.InterestRate3M.get_last_update_date()
91
92if self.fx_carry_flag is not None and not self.fx_carry_flag and not self.Portfolio.Invested:
93    self.lag_flag = True
94   
95    # # long UUP
96    # if self.dollar in data and data[self.dollar]:
97    #     self.SetHoldings(self.dollar, 1)
98    # self.fx_carry_flag = None
99
100elif self.fx_carry_flag is not None and self.fx_carry_flag and not self.Portfolio.Invested:
101    self.lag_flag = True
102    rebalance_flag:bool = False
103    rate:dict[str, float] = {}
104
105    for symbol, int_rate in self.symbols.items():
106        # futures data is present in the algorithm
107        if symbol in data and data[symbol]:
108            rebalance_flag = True
109
110            # IR data is still comming in
111            if self.Securities[int_rate].GetLastData() and ir_last_update_date[int_rate] > self.Time.date() \
112                and self.Securities[symbol].GetLastData() and qp_futures_last_update_date[symbol] > self.Time.date():
113                rate[symbol] = self.Securities[int_rate].Price
114
115    if rebalance_flag:
116        if len(rate) >= self.traded_count:
117            # interbank rate sorting
118            sorted_by_rate = sorted(rate.items(), key = lambda x: x[1], reverse = True)
119            self.long = [x[0] for x in sorted_by_rate[:self.traded_count]]
120            self.short = [x[0] for x in sorted_by_rate[-self.traded_count:]]