FX Value v2 – Real Exchange Rate Changes

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