Conditional Currency Momentum Portfolios

1import data_tools
2import numpy as np
3import pandas as pd
4from AlgorithmImports import *
5from pandas.core.frame import DataFrame
6from typing import List, Dict
7# endregion
8
9class ConditionalCurrencyMomentumPortfolios(QCAlgorithm):
10
11def Initialize(self):
12self.SetStartDate(2000, 1, 1)
13self.SetCash(100000)
14
15self.us_ir:Symbol = self.AddData(data_tools.InterestRate3M, 'IR3TIB01USM156N', Resolution.Daily).Symbol
16
17self.leverage:int = 5
18self.quantile:int = 4
19self.monthly_period:int = 36
20self.daily_period:int = 21
21
22self.data:Dict[Symbol, data_tools.SymbolData] = {}
23self.top_portfolio:List[Symbol] = []
24self.bottom_portfolio:List[Symbol] = []
25
26# Cash rate source: https://fred.stlouisfed.org/series/IR3TIB01USM156N
27self.symbols:Dict[str, str] = {
28    "AUDUSD" : "IR3TIB01AUM156N",   # Australian Dollar Futures, Continuous Contract #1
29    "GBPUSD" : "LIOR3MUKM",         # British Pound Futures, Continuous Contract #1
30    "CADUSD" : "IR3TIB01CAM156N",   # Canadian Dollar Futures, Continuous Contract #1
31    "EURUSD" : "IR3TIB01EZM156N",   # Euro FX Futures, Continuous Contract #1
32    "JPYUSD" : "IR3TIB01JPM156N",   # Japanese Yen Futures, Continuous Contract #1
33    "MXNUSD" : "IR3TIB01MXM156N",   # Mexican Peso Futures, Continuous Contract #1
34    "NZDUSD" : "IR3TIB01NZM156N",   # New Zealand Dollar Futures, Continuous Contract #1
35    "CHFUSD" : "IR3TIB01CHM156N"    # Swiss Franc Futures, Continuous Contract #1
36}
37
38# data subscription
39for symbol, rate_symbol in self.symbols.items():
40    data:Security = self.AddForex(symbol, Resolution.Daily, Market.Oanda)
41    data.SetFeeModel(data_tools.CustomFeeModel())
42    data.SetLeverage(self.leverage)
43
44    ir_symbol:Symbol = self.AddData(data_tools.InterestRate3M, rate_symbol, Resolution.Daily).Symbol
45
46    self.data[data.Symbol] = data_tools.SymbolData(self.daily_period, ir_symbol)
47
48self.return_dispersion:RollingWindow = RollingWindow[float](self.monthly_period)
49self.volatility:RollingWindow = RollingWindow[float](self.monthly_period)
50
51self.SetWarmup(self.monthly_period * self.daily_period, Resolution.Daily)
52
53self.recent_month:int = -1
54self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
55
56def OnData(self, data: Slice) -> None:
57rebalance_flag:bool = False
58
59# store daily prices
60for symbol, symbol_data in self.data.items():
61    if symbol in data and data[symbol]:
62        symbol_data.update_price(data[symbol].Price)
63
64        # monthly rebalance
65        if self.recent_month != self.Time.month and not rebalance_flag:
66            self.recent_month = self.Time.month
67            rebalance_flag = True
68
69if not rebalance_flag:
70    return
71
72ir_last_update_date:Dict[str, datetime.date] = data_tools.InterestRate3M.get_last_update_date()
73curr_us_ir_value:Union[float, None] = self.Securities[self.us_ir].Price if self.Securities[self.us_ir].GetLastData() and ir_last_update_date[self.us_ir.Value] > self.Time.date() else None
74
75if curr_us_ir_value is None:
76    self.Liquidate()
77    return
78
79fd_by_symbol:Dict[Symbol, float] = {}
80
81# save daily prices lists
82for symbol, symbol_data in self.data.items():
83    if self.Securities[symbol_data._ir_symbol].GetLastData():
84        if ir_last_update_date[symbol_data._ir_symbol.Value] < self.Time.date():
85            continue
86            
87        if not self.data[symbol].is_ready():
88            continue
89        
90        fd_by_symbol[symbol] = self.Securities[symbol_data._ir_symbol].Price - curr_us_ir_value
91
92self.return_dispersion.Add(np.sqrt((np.sum(([self.data[x].get_monthly_return() for x in fd_by_symbol] - np.mean([self.data[x].get_monthly_return() for x in fd_by_symbol])) ** 2) / (len(fd_by_symbol) - 1))))
93self.volatility.Add(np.mean(np.mean(np.abs([self.data[x].get_daily_returns() for x in fd_by_symbol]), axis=0)))
94
95if self.IsWarmingUp:
96    return
97
98# signal calculation
99if len(fd_by_symbol) < self.quantile or (not self.return_dispersion.IsReady and not self.volatility.IsReady):
100    self.Liquidate()
101    return
102
103afd:float = np.mean(list(fd_by_symbol.values()))
104afd_signal:int = 1 if afd > 0 else 0
105ord_signal = 1 if self.return_dispersion[0] > np.percentile(np.array(list(self.return_dispersion)[1:]), 90) else 0
106vol_signal = 1 if self.volatility[0] > np.percentile(np.array(list(self.volatility)[1:]), 90) else 0
107
108signals:List[int] = [afd_signal, ord_signal, vol_signal]
109traded_direction:int = 1 if all(x==1 for x in signals) else -1
110
111# sort and divide to quantiles
112sorted_by_momentum:List[Symbol] = sorted([(symbol, self.data[symbol].get_monthly_return()) for symbol, value in fd_by_symbol.items()], key=lambda x: x[1])
113quantile:int = len(sorted_by_momentum) // self.quantile
114top_portfolio:List[Symbol] = [i[0] for i in sorted_by_momentum][-quantile:]
115bottom_portfolio:List[Symbol] = [i[0] for i in sorted_by_momentum][:quantile]
116
117# trade execution
118invested:List[Symbol] = [x.Key for x in self.Portfolio if x.Value.Invested]
119for symbol in invested:
120    if symbol not in top_portfolio + bottom_portfolio:
121        self.Liquidate(symbol)
122
123for i, portfolio in enumerate([top_portfolio, bottom_portfolio]):
124    for symbol in portfolio:
125        if symbol in data and data[symbol]:
126            self.SetHoldings(symbol, ((-1) ** i) * traded_direction / len(top_portfolio))