Optimized Currency Portfolios

1from AlgorithmImports import *
2import numpy as np
3import data_tools
4from scipy.optimize import minimize
5#endregion
6class OptimizedCurrencyPortfolios(QCAlgorithm):
7def Initialize(self):
8self.SetStartDate(2007, 1, 1)
9self.SetCash(100000)
10
11# currency future symbol and interbank rate
12self.symbols:dict[str, str] = {
13    'CME_AD1' : 'IR3TIB01AUM156N',   # Australian Dollar Futures, Continuous Contract #1
14    'CME_CD1' : 'IR3TIB01CAM156N',   # Canadian Dollar Futures, Continuous Contract #1
15    'CME_SF1' : 'IR3TIB01CHM156N',   # Swiss Franc Futures, Continuous Contract #1
16    'CME_EC1' : 'IR3TIB01EZM156N',   # Euro FX Futures, Continuous Contract #1
17    'CME_BP1' : 'LIOR3MUKM',         # British Pound Futures, Continuous Contract #1
18    'CME_JY1' : 'IR3TIB01JPM156N',   # Japanese Yen Futures, Continuous Contract #1
19    'CME_NE1' : 'IR3TIB01NZM156N',   # New Zealand Dollar Futures, Continuous Contract #1 # data are from 2006
20    'CME_MP1' : 'IR3TIB01MXM156N',   # Mexican Peso Futures, Continuous Contract #1
21    }
22self.m_momentum_period:int = 3+1
23self.m_reversal_period:int = 5*12+1
24self.SetWarmUp(self.m_reversal_period*21, Resolution.Daily)
25self.monthly_price_data:dict = {}
26self.gamma:float = 5.
27self.futures_max_missing_days:int = 5
28self.ir_max_missing_days:int = 31
29
30# characteristics names
31self.characteristics:list[str] = ['sign', 'fd', 'mom', 'rev']
32self.characteristics_by_symbol:dict[str, pd.DataFrame] = {}
33self.performance_df:pd.DataFrame = pd.DataFrame(columns=list(self.symbols.keys()))
34# optimalization setup
35self.opt_period:int = 60
36for currency_future, cash_rate_symbol in self.symbols.items():
37    # currency futures data
38    data:Security = self.AddData(data_tools.QuantpediaFutures, currency_future, Resolution.Daily)
39    data.SetFeeModel(data_tools.CustomFeeModel())
40    data.SetLeverage(10)
41    
42    # interbank rate data
43    self.AddData(data_tools.InterestRate3M, cash_rate_symbol, Resolution.Daily)
44    # price data
45    self.monthly_price_data[currency_future] = RollingWindow[float](self.m_reversal_period)
46    
47    # characteristics by symbol
48    self.characteristics_by_symbol[currency_future] = pd.DataFrame(columns=self.characteristics)
49# USD interbank rate data
50self.usd_ir:Symbol = self.AddData(data_tools.InterestRate3M, 'IR3TIB01USM156N', Resolution.Daily).Symbol
51self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
52self.recent_month:int = -1
53self.month_counter:int = 0
54def OnData(self, data):
55rebalance_flag:bool = False
56sign:dict[str, int] = {}
57fd_by_currency:dict[str, float] = {}
58fd:dict[str, float] = {}
59mom:dict[str, float] = {}
60rev:dict[str, float] = {}
61performance:dict[str, float] = {}
62ir_last_update_date:Dict[str, datetime.date] = data_tools.InterestRate3M.get_last_update_date()
63qp_futures_last_update_date:Dict[str, datetime.date] = data_tools.QuantpediaFutures.get_last_update_date()
64for currency, rate in self.symbols.items():
65    if currency in data and data[currency]:
66        foo=3
67if self.usd_ir in data and data[self.usd_ir]:
68    # rebalance once a month, once new interbank data comes in
69    if self.Time.month != self.recent_month:
70        rebalance_flag = True
71        for currency_future, cash_rate_symbol in self.symbols.items():
72            # interbank rate data is present in the algorithm
73            if cash_rate_symbol in data and data[cash_rate_symbol]:
74                # making sure custom futures data is still comming in
75                if not self.Securities[currency_future].GetLastData() and qp_futures_last_update_date[currency_future] > self.Time.date():
76                    continue
77                
78                # store monthly price data
79                if self.Securities[currency_future].Price != 0:
80                    self.monthly_price_data[currency_future].Add(self.Securities[currency_future].Price)
81                if self.monthly_price_data[currency_future].IsReady:
82                    # calculate variables
83                    sign[currency_future] = 1 if data[cash_rate_symbol].Value > data[self.usd_ir].Value else -1
84                    # only partial information for fd variable
85                    fd_by_currency[currency_future] = data[cash_rate_symbol].Value
86                    mom[currency_future] = self.monthly_price_data[currency_future][0] / self.monthly_price_data[currency_future][self.m_momentum_period-1] - 1
87                    
88                    # three months is excluded to avoid unnecessary correlation with the momentum variable
89                    rev[currency_future] = self.monthly_price_data[currency_future][self.m_momentum_period-1] / self.monthly_price_data[currency_future][self.m_reversal_period-1] - 1
90                    
91                    performance[currency_future] = self.monthly_price_data[currency_future][0] / self.monthly_price_data[currency_future][1] - 1
92        
93else:
94    # IR data stopped comming in
95    if not any(self.Securities[x] for x in list(self.symbols.keys())) and any(ir_futures_last_update_date[x] <= self.Time.date() for x in list(self.symbols.keys())):
96        self.Liquidate()
97        return
98
99if self.IsWarmingUp:
100    return
101  
102if not rebalance_flag:
103    return
104self.recent_month = self.Time.month
105# every symbol has data filled
106# if len(sign) == len(self.symbols):
107if len(sign) != 0:
108    # complete fd variable
109    fd_values:list[float] = list(fd_by_currency.values())
110    fd = { currency_future : (fd - np.mean(fd_values)) / np.std(fd_values) for currency_future, fd in fd_by_currency.items() }
111    # standardize every variable
112    sign = self.standardize_dict(sign)
113    fd = self.standardize_dict(fd)
114    mom = self.standardize_dict(mom)
115    rev = self.standardize_dict(rev)
116    
117    # append new row to characteristics dataframe
118    for symbol in sign:
119        eval_dicts:list[dict] = []
120        for c in self.characteristics:
121            eval_dict:dict = eval(c)
122            eval_dicts.append(eval_dict)
123        self.characteristics_by_symbol[symbol].loc[len(self.characteristics_by_symbol[symbol].index)] = [d[symbol] for d in eval_dicts]
124    # self.performance_df.loc[len(self.performance_df.index)] = [performance.get(symbol, 0) for symbol in self.symbols]
125    if len(performance) != len(self.symbols):
126        return
127    self.performance_df.loc[len(self.performance_df.index)] = [performance[symbol] for symbol in self.symbols]
128    
129    # optimize once a year
130    self.month_counter += 1
131    if self.month_counter != 12:
132        return
133    self.month_counter = 0
134    
135    weights:dict[str, float] = {}
136    # dataframes are ready for optimization
137    if len(self.characteristics_by_symbol[list(self.symbols.keys())[0]]) >= self.opt_period:
138        # optimize portfolio policy and calculate weights for the next year
139        opt = self.optimization_method(self.performance_df, self.characteristics_by_symbol, self.characteristics, self.opt_period, self.gamma)[1]
140        weights = { symbol : np.sum(self.characteristics_by_symbol[symbol].iloc[-1].values * opt.values) / len(self.symbols) for symbol in self.symbols }
141    # trade execution
142    invested = [x.Key.Value for x in self.Portfolio if x.Value.Invested]
143    for symbol in invested:
144        if symbol not in weights:
145            self.Liquidate(symbol)
146    
147    for symbol, w in weights.items():
148        self.SetHoldings(symbol, w)
149def standardize_dict(self, dict_to_standardize:dict) -> dict:
150''' cross-sectional standardization '''
151dict_values:list[float] = list(dict_to_standardize.values())
152result:dict = { key : (value - np.mean(dict_values)) / np.std(dict_values) for key, value in dict_to_standardize.items() }
153return result
154def optimization_method(self, returns:pd.DataFrame, characteristics_dict:dict, characteristics:list, opt_period:int, gamma:float):
155''' parametric portfolio policy optimization '''
156size:int = len(characteristics)
157
158# objective function
159fun = lambda weights: self.opt_fun(returns, characteristics_dict, weights, gamma, opt_period)
160# Constraint #1: The weights adds up to 1
161# constraints:list = [{'type': 'eq', 'fun': lambda w: 1 - np.sum(w)}]
162constraints:list = []
163bounds:tuple = tuple((-1, 1) for x in range(size))
164# initial parametric portfolio policy
165x0:np.ndarray = np.array(size * [1. / size])
166opt = minimize(fun,                         # Objective function
167               x0,                          # Initial guess
168               method='SLSQP',              # Optimization method:  Sequential Least Squares Programming
169               bounds = bounds,             # Bounds for variables 
170               constraints = constraints)   # Constraints definition
171return opt, pd.Series(opt['x'], index = characteristics)
172
173def opt_fun(self, returns:pd.DataFrame, characteristics:dict, weights:np.ndarray, gamma:float, opt_period:int):
174symbols:list[str] = list(returns.columns)
175df_dict:dict[str, pd.Series] = {}
176for symbol in symbols:
177    relevant_c:pd.DataFrame = characteristics[symbol].iloc[-opt_period:-1]
178    # product of weights and performance
179    adj_w:np.ndarray = (relevant_c.multiply(weights, axis=1).sum(axis=1) / len(symbols)).reset_index(drop=True)
180    relevant_perf:pd.Series = returns[symbol].iloc[-(opt_period-1):].reset_index(drop=True)
181    monthly_returns:pd.Series = adj_w * relevant_perf
182    df_dict[symbol] = monthly_returns
183adj_perf_df:pd.DataFrame = pd.DataFrame(df_dict)
184portfolio_perf:float = np.sum(adj_perf_df, axis=1)
185
186# calculate equity
187init_eq:float = 1.
188eq:float = init_eq
189for ret in portfolio_perf.values:
190    eq = eq * (1 + ret)
191portfolio_return:float = eq / init_eq - 1.
192
193# power utility as the objective function
194utility = ((portfolio_return) ** (1-gamma)) / (1-gamma)
195return -utility