The strategy optimizes currency portfolios using interest rates, momentum, and reversals, maximizing investor utility with a rolling 20-year training window and annual re-estimation for updated predictions.
ASSET CLASS: CFDs, forwards, futures, swaps | REGION: Global| FREQUENCY:
Monthly | MARKET: currencies | KEYWORD: Optimized Currency, Portfolios
I. STRATEGY IN A NUTSHELL
The strategy constructs currency portfolios for 17 countries using interest rate differentials, 3-month momentum, and 5-year long-term reversals. Optimal portfolio weights are derived via a utility-maximizing model (γ = 5), trained on a rolling 20-year window and updated annually for the next 12 months.
II. ECONOMIC RATIONALE
By combining interest rate signals, short-term momentum, and long-term reversals, the strategy predicts currency performance and allocates capital to maximize investor utility. Annual re-estimation ensures adaptation to evolving market conditions and risk-return dynamics.
III. SOURCE PAPER
Beyond the Carry Trade: Optimal Currency Portfolios [Click to Open PDF]
Pedro Barroso and Pedro Santa-Clara.CATÓLICA-LISBON School of Business & Economics.Nova School of Business and Economics; National Bureau of Economic Research (NBER); Centre for Economic Policy Research (CEPR).
<Abstract>
We test the relevance of technical and fundamental variables in forming currency portfolios. Carry, momentum and reversal all contribute to portfolio performance, whereas the real exchange rate and the current account do not. The resulting optimal portfolio outperforms the carry trade and other naive benchmarks in an extensive 16 year out-of-sample test. Its returns are not explained by risk and are valuable to diversified investors holding stocks and bonds. Exposure to currencies increases the Sharpe ratio of diversified portfolios by 0.5 on average, while reducing crash risk. We argue that currency returns are an anomaly which is gradually being corrected as hedge fund capital increases.
V. BACKTEST PERFORMANCE
| Annualised Return | 19.2% |
| Volatility | 22.2% |
| Beta | 0.05 |
| Sharpe Ratio | 1.4 |
| Sortino Ratio | -0.289 |
| Maximum Drawdown | N/A |
| Win Rate | 59% |
V. FULL PYTHON CODE
from AlgorithmImports import *
import numpy as np
import data_tools
from scipy.optimize import minimize
#endregion
class OptimizedCurrencyPortfolios(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2007, 1, 1)
self.SetCash(100000)
# currency future symbol and interbank rate
self.symbols:dict[str, str] = {
'CME_AD1' : 'IR3TIB01AUM156N', # Australian Dollar Futures, Continuous Contract #1
'CME_CD1' : 'IR3TIB01CAM156N', # Canadian Dollar Futures, Continuous Contract #1
'CME_SF1' : 'IR3TIB01CHM156N', # Swiss Franc Futures, Continuous Contract #1
'CME_EC1' : 'IR3TIB01EZM156N', # Euro FX Futures, Continuous Contract #1
'CME_BP1' : 'LIOR3MUKM', # British Pound Futures, Continuous Contract #1
'CME_JY1' : 'IR3TIB01JPM156N', # Japanese Yen Futures, Continuous Contract #1
'CME_NE1' : 'IR3TIB01NZM156N', # New Zealand Dollar Futures, Continuous Contract #1 # data are from 2006
'CME_MP1' : 'IR3TIB01MXM156N', # Mexican Peso Futures, Continuous Contract #1
}
self.m_momentum_period:int = 3+1
self.m_reversal_period:int = 5*12+1
self.SetWarmUp(self.m_reversal_period*21, Resolution.Daily)
self.monthly_price_data:dict = {}
self.gamma:float = 5.
self.futures_max_missing_days:int = 5
self.ir_max_missing_days:int = 31
# characteristics names
self.characteristics:list[str] = ['sign', 'fd', 'mom', 'rev']
self.characteristics_by_symbol:dict[str, pd.DataFrame] = {}
self.performance_df:pd.dataframe = pd.dataframe(columns=list(self.symbols.keys()))
# optimalization setup
self.opt_period:int = 60
for currency_future, cash_rate_symbol in self.symbols.items():
# currency futures data
data:Security = self.AddData(data_tools.QuantpediaFutures, currency_future, Resolution.Daily)
data.SetFeeModel(data_tools.CustomFeeModel())
data.SetLeverage(10)
# interbank rate data
self.AddData(data_tools.InterestRate3M, cash_rate_symbol, Resolution.Daily)
# price data
self.monthly_price_data[currency_future] = RollingWindow[float](self.m_reversal_period)
# characteristics by symbol
self.characteristics_by_symbol[currency_future] = pd.dataframe(columns=self.characteristics)
# USD interbank rate data
self.usd_ir:Symbol = self.AddData(data_tools.InterestRate3M, 'IR3TIB01USM156N', Resolution.Daily).Symbol
self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
self.recent_month:int = -1
self.month_counter:int = 0
def OnData(self, data):
rebalance_flag:bool = False
sign:dict[str, int] = {}
fd_by_currency:dict[str, float] = {}
fd:dict[str, float] = {}
mom:dict[str, float] = {}
rev:dict[str, float] = {}
performance:dict[str, float] = {}
ir_last_update_date:Dict[str, datetime.date] = data_tools.InterestRate3M.get_last_update_date()
qp_futures_last_update_date:Dict[str, datetime.date] = data_tools.QuantpediaFutures.get_last_update_date()
for currency, rate in self.symbols.items():
if currency in data and data[currency]:
foo=3
if self.usd_ir in data and data[self.usd_ir]:
# rebalance once a month, once new interbank data comes in
if self.Time.month != self.recent_month:
rebalance_flag = True
for currency_future, cash_rate_symbol in self.symbols.items():
# interbank rate data is present in the algorithm
if cash_rate_symbol in data and data[cash_rate_symbol]:
# making sure custom futures data is still comming in
if not self.Securities[currency_future].GetLastData() and qp_futures_last_update_date[currency_future] > self.Time.date():
continue
# store monthly price data
if self.Securities[currency_future].Price != 0:
self.monthly_price_data[currency_future].Add(self.Securities[currency_future].Price)
if self.monthly_price_data[currency_future].IsReady:
# calculate variables
sign[currency_future] = 1 if data[cash_rate_symbol].Value > data[self.usd_ir].Value else -1
# only partial information for fd variable
fd_by_currency[currency_future] = data[cash_rate_symbol].Value
mom[currency_future] = self.monthly_price_data[currency_future][0] / self.monthly_price_data[currency_future][self.m_momentum_period-1] - 1
# three months is excluded to avoid unnecessary correlation with the momentum variable
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
performance[currency_future] = self.monthly_price_data[currency_future][0] / self.monthly_price_data[currency_future][1] - 1
else:
# IR data stopped comming in
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())):
self.Liquidate()
return
if self.IsWarmingUp:
return
if not rebalance_flag:
return
self.recent_month = self.Time.month
# every symbol has data filled
# if len(sign) == len(self.symbols):
if len(sign) != 0:
# complete fd variable
fd_values:list[float] = list(fd_by_currency.values())
fd = { currency_future : (fd - np.mean(fd_values)) / np.std(fd_values) for currency_future, fd in fd_by_currency.items() }
# standardize every variable
sign = self.standardize_dict(sign)
fd = self.standardize_dict(fd)
mom = self.standardize_dict(mom)
rev = self.standardize_dict(rev)
# append new row to characteristics dataframe
for symbol in sign:
eval_dicts:list[dict] = []
for c in self.characteristics:
eval_dict:dict = eval(c)
eval_dicts.append(eval_dict)
self.characteristics_by_symbol[symbol].loc[len(self.characteristics_by_symbol[symbol].index)] = [d[symbol] for d in eval_dicts]
# self.performance_df.loc[len(self.performance_df.index)] = [performance.get(symbol, 0) for symbol in self.symbols]
if len(performance) != len(self.symbols):
return
self.performance_df.loc[len(self.performance_df.index)] = [performance[symbol] for symbol in self.symbols]
# optimize once a year
self.month_counter += 1
if self.month_counter != 12:
return
self.month_counter = 0
weights:dict[str, float] = {}
# dataframes are ready for optimization
if len(self.characteristics_by_symbol[list(self.symbols.keys())[0]]) >= self.opt_period:
# optimize portfolio policy and calculate weights for the next year
opt = self.optimization_method(self.performance_df, self.characteristics_by_symbol, self.characteristics, self.opt_period, self.gamma)[1]
weights = { symbol : np.sum(self.characteristics_by_symbol[symbol].iloc[-1].values * opt.values) / len(self.symbols) for symbol in self.symbols }
# trade execution
invested = [x.Key.Value for x in self.Portfolio if x.Value.Invested]
for symbol in invested:
if symbol not in weights:
self.Liquidate(symbol)
for symbol, w in weights.items():
self.SetHoldings(symbol, w)
def standardize_dict(self, dict_to_standardize:dict) -> dict:
''' cross-sectional standardization '''
dict_values:list[float] = list(dict_to_standardize.values())
result:dict = { key : (value - np.mean(dict_values)) / np.std(dict_values) for key, value in dict_to_standardize.items() }
return result
def optimization_method(self, returns:pd.dataframe, characteristics_dict:dict, characteristics:list, opt_period:int, gamma:float):
''' parametric portfolio policy optimization '''
size:int = len(characteristics)
# objective function
fun = lambda weights: self.opt_fun(returns, characteristics_dict, weights, gamma, opt_period)
# Constraint #1: The weights adds up to 1
# constraints:list = [{'type': 'eq', 'fun': lambda w: 1 - np.sum(w)}]
constraints:list = []
bounds:tuple = tuple((-1, 1) for x in range(size))
# initial parametric portfolio policy
x0:np.ndarray = np.array(size * [1. / size])
opt = minimize(fun, # Objective function
x0, # Initial guess
method='SLSQP', # Optimization method: Sequential Least Squares Programming
bounds = bounds, # Bounds for variables
constraints = constraints) # Constraints definition
return opt, pd.Series(opt['x'], index = characteristics)
def opt_fun(self, returns:pd.dataframe, characteristics:dict, weights:np.ndarray, gamma:float, opt_period:int):
symbols:list[str] = list(returns.columns)
df_dict:dict[str, pd.Series] = {}
for symbol in symbols:
relevant_c:pd.dataframe = characteristics[symbol].iloc[-opt_period:-1]
# product of weights and performance
adj_w:np.ndarray = (relevant_c.multiply(weights, axis=1).sum(axis=1) / len(symbols)).reset_index(drop=True)
relevant_perf:pd.Series = returns[symbol].iloc[-(opt_period-1):].reset_index(drop=True)
monthly_returns:pd.Series = adj_w * relevant_perf
df_dict[symbol] = monthly_returns
adj_perf_df:pd.dataframe = pd.dataframe(df_dict)
portfolio_perf:float = np.sum(adj_perf_df, axis=1)
# calculate equity
init_eq:float = 1.
eq:float = init_eq
for ret in portfolio_perf.values:
eq = eq * (1 + ret)
portfolio_return:float = eq / init_eq - 1.
# power utility as the objective function
utility = ((portfolio_return) ** (1-gamma)) / (1-gamma)
return -utility