FX Value v3 – Real Exchange Rate Levels

1from AlgorithmImports import *
2import data_tools
3from typing import Dict, List
4from dateutil.relativedelta import relativedelta
5class FXValuev3RealExchangeRateLevels(QCAlgorithm):
6def Initialize(self):
7self.SetStartDate(2000, 1, 1)
8self.SetCash(100000)
9
10self.leverage:int = 3
11self.selected_currencies:int = 3 # We select these many of forex futures for long and short each rebalance.
12
13# Currency future symbol and PPP yearly symbol.
14self.symbols:Dict[str, str] = {
15    "CME_AD1" : "AUS_PPP", # Australian Dollar Futures, Continuous Contract #1
16    "CME_BP1" : "GBR_PPP", # British Pound Futures, Continuous Contract #1
17    "CME_CD1" : "CAN_PPP", # Canadian Dollar Futures, Continuous Contract #1
18    "CME_EC1" : "DEU_PPP", # Euro FX Futures, Continuous Contract #1
19    "CME_JY1" : "JPN_PPP", # Japanese Yen Futures, Continuous Contract #1
20    "CME_NE1" : "NZL_PPP", # New Zealand Dollar Futures, Continuous Contract #1
21    "CME_SF1" : "CHE_PPP"  # Swiss Franc Futures, Continuous Contract #1
22}
23
24self.last_ppp:Dict[str, Union[None, str]] = {
25    "AUS_PPP" : None,
26    "GBR_PPP" : None,
27    "CAN_PPP" : None,
28    "DEU_PPP" : None,
29    "JPN_PPP" : None,
30    "NZL_PPP" : None,
31    "CHE_PPP" : None,
32    "USA_PPP" : None
33}
34
35for symbol, ppp_symbol in self.symbols.items():
36    data = self.AddData(data_tools.QuantpediaFutures, symbol, Resolution.Daily)
37    data.SetFeeModel(data_tools.CustomFeeModel())
38    data.SetLeverage(self.leverage)
39    
40    # PPP quandl data.
41    self.AddData(data_tools.PPPData, ppp_symbol, Resolution.Daily)
42    
43self.AddData(data_tools.PPPData, 'USA_PPP', Resolution.Daily).Symbol
44
45self.selection_flag:bool = False
46self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
47self.Schedule.On(self.DateRules.WeekStart('CME_AD1'), self.TimeRules.At(0, 0), self.Selection)
48def OnData(self, data: Slice) -> None:
49ppp_last_update_date:Dict[str, datetime.date] = data_tools.PPPData.get_last_update_date()
50future_last_update_date:Dict[str, datetime.date] = data_tools.QuantpediaFutures.get_last_update_date()
51# Store PPP values, when they are available.
52for ppp_symbol, _ in self.last_ppp.items():
53    if ppp_symbol in data and data[ppp_symbol]:
54        # Only last PPP for each country is stored.
55        self.last_ppp[ppp_symbol] = data[ppp_symbol].Value
56
57if not self.selection_flag:
58    return
59self.selection_flag = False
60
61exchange_ratio:Dict[str, float] = {}
62symbols_to_delete:List[str] = []
63
64for symbol, ppp_symbol in self.symbols.items():
65    # data is still coming
66    if self.Securities[symbol].GetLastData() and self.Time.date() > future_last_update_date[symbol] \
67        or self.Securities[ppp_symbol].GetLastData() and self.Time.date() > ppp_last_update_date[ppp_symbol] + relativedelta(months=12):
68        symbols_to_delete.append(symbol)
69        continue
70    # Check if all data are ready for calculation
71    if symbol in data and data[symbol] and self.last_ppp[ppp_symbol] and self.last_ppp['USA_PPP']:
72        
73        # real exchange rate = forex future price * (forex future PPP / USA PPP)
74        real_exchange_rate:float = data[symbol].Value * (self.last_ppp[ppp_symbol] / self.last_ppp['USA_PPP'])
75        nominal_exchange_rate:float = data[symbol].Value
76            
77        exchange_ratio[symbol] = nominal_exchange_rate / real_exchange_rate
78if len(symbols_to_delete) != 0:
79    [self.symbols.pop(symbol) for symbol in symbols_to_delete]
80# currency with the highest (lowest) past 5-year return is considered most overvalued (undervalued)
81sorted_by_ratio:List[str] = [x[0] for x in sorted(exchange_ratio.items(), key=lambda item: item[1])]
82# Investor ranks all currencies from the most undervalued to the most overvalued and goes long top 20% (the most undervalued) currencies and
83# goes short 20% (the most overvalued) currencies.
84
85# Go long self.selected_currencies with the lowest ratio
86long:List[str] = sorted_by_ratio[:self.selected_currencies]
87
88# Go short self.selected_currencies with the highest ratio
89short:List[str] = sorted_by_ratio[-self.selected_currencies:]
90
91# Trade execution
92invested:List[str] = [x.Key.Value for x in self.Portfolio if x.Value.Invested]
93for symbol in invested:
94    if symbol not in long + short:
95        self.Liquidate(symbol)
96        
97for i, portfolio in enumerate([long, short]):
98    for symbol in portfolio:
99        if symbol in data and data[symbol]:
100            self.SetHoldings(symbol, ((-1) ** i) / self.selected_currencies) 
101
102def Selection(self) -> None:
103self.selection_flag = True