PCA-Enhanced Mean-Variance Optimized Global Currency Portfolio Strategy

The strategy uses 15 developed currencies, applying mean-variance optimization with expected returns from forward discounts and covariance matrix estimated from past exchange rate movements. The portfolio is rebalanced monthly.

ASSET CLASS: forwards, futures, swaps | REGION: Global | FREQUENCY:
Monthly | MARKET: currencies | KEYWORD: Mean-Variance

I. STRATEGY IN A NUTSHELL

The strategy invests in 15 developed currencies, determining portfolio weights via mean-variance optimization. Expected returns are set equal to forward discounts, while the covariance matrix is estimated using exponentially weighted moving averages of squared exchange rate growths. Principal components explaining less than 1% of variance are removed, and the adjusted covariance matrix is scaled by the investor’s relative risk aversion. The portfolio is rebalanced monthly.

II. ECONOMIC RATIONALE

By leveraging forward discounts as expected returns, the strategy reduces estimation errors in the FX market. PCA improves covariance matrix robustness by removing low-variance components, mitigating near-arbitrage effects. This enhances out-of-sample Sharpe ratios, generates positive skewness, and outperforms other strategies even after transaction costs. Market timing and flexible risk scaling further boost performance, maintaining profitability across recessions, Euro adoption, and both long and shorter historical periods.

III. SOURCE PAPER

Market Timing and Predictability in FX Markets [Click to Open PDF]

Maurer, Thomas Andreas and To, Thuy Duong and Tran, Ngoc-Khanh The University of Hong Kong; Washington University in St. Louis – John M. Olin Business School; London School of Economics & Political Science (LSE), University of New South Wales, Sydney; Financial Research Network (FIRN), Finance Dept., Pamplin College of Business, Virginia Tech; Olin Business School- Washington University in St. Louis

<Abstract>

We study the economic value of market timing in FX markets, i.e., using information about the conditional Sharpe ratio to adjust the notional value of a conditionally mean-variance efficient currency portfolio. Our strategy trades more (less) aggressively when the conditional risk-return trade-off is more (less) favorable. This leads to a significant improvement in the out-of-sample unconditional Sharpe ratio, skewness and maximum drawdown per 1% expected excess return. The strategy’s market timing predicts returns, volatility and skewness in FX markets. Popular currency pricing factors do not explain the strategy’s high average excess returns. Our findings suggest that it is costly to impose leverage or risk (i.e., conditional volatility) limits or other inferior market timing policies when constructing currency trading strategies.

IV. BACKTEST PERFORMANCE

Annualised Return	7.44%
Volatility	8.12%
Beta	0.065
Sharpe Ratio	0.92
Sortino Ratio	-0.597
Maximum Drawdown	16.87%
Win Rate	66%

V. FULL PYTHON CODE

from AlgorithmImports import *
import data_tools
from scipy.optimize import minimize
from enum import Enum
# endregion
class TradedUniverse(Enum):
    FX = 1
    FX_FUTURES = 2
class MeanVarianceMarketTimingInTheFXMarket(QCAlgorithm):
    def initialize(self) -> None:
        self.set_start_date(2000, 1, 1)
        self.set_cash(1_000_000)
        self._us_ir: Symbol = self.AddData(data_tools.InterestRate3M, 'IR3TIB01USM156N', Resolution.Daily).Symbol
        self._traded_universe: TradedUniverse = TradedUniverse.FX_FUTURES
        period: int = 6
        self._min_weight: float = .01
        self._EWMA_lambda: float = .95
        self._data: Dict[Symbol, data_tools.SymbolData] = {}
        # Cash rate source: https://fred.stlouisfed.org/series/IR3TIB01USM156N
        if self._traded_universe == TradedUniverse.FX:
            symbols: Dict[str, str] = {
                "AUDUSD" : "IR3TIB01AUM156N",   # Australian Dollar Futures, Continuous Contract #1
                "GBPUSD" : "LIOR3MUKM",         # British Pound Futures, Continuous Contract #1
                "CADUSD" : "IR3TIB01CAM156N",   # Canadian Dollar Futures, Continuous Contract #1
                "EURUSD" : "IR3TIB01EZM156N",   # Euro FX Futures, Continuous Contract #1
                "JPYUSD" : "IR3TIB01JPM156N",   # Japanese Yen Futures, Continuous Contract #1
                "MXNUSD" : "IR3TIB01MXM156N",   # Mexican Peso Futures, Continuous Contract #1
                "NZDUSD" : "IR3TIB01NZM156N",   # New Zealand Dollar Futures, Continuous Contract #1
                "CHFUSD" : "IR3TIB01CHM156N"    # Swiss Franc Futures, Continuous Contract #1
            }
        elif self._traded_universe == TradedUniverse.FX_FUTURES:
            symbols: Dict[str, str] = {
                "CME_AD1" : "IR3TIB01AUM156N",   # Australian Dollar Futures, Continuous Contract #1
                "CME_BP1" : "LIOR3MUKM",         # British Pound Futures, Continuous Contract #1
                "CME_CD1" : "IR3TIB01CAM156N",   # Canadian Dollar Futures, Continuous Contract #1
                "CME_EC1" : "IR3TIB01EZM156N",   # Euro FX Futures, Continuous Contract #1
                "CME_JY1" : "IR3TIB01JPM156N",   # Japanese Yen Futures, Continuous Contract #1
                "CME_MP1" : "IR3TIB01MXM156N",   # Mexican Peso Futures, Continuous Contract #1
                "CME_NE1" : "IR3TIB01NZM156N",   # New Zealand Dollar Futures, Continuous Contract #1
                "CME_SF1" : "IR3TIB01CHM156N"    # Swiss Franc Futures, Continuous Contract #1
            }
        # data subscription
        for symbol, rate_symbol in symbols.items():
            if self._traded_universe == TradedUniverse.FX:
                data: Security = self.add_forex(symbol, Resolution.MINUTE, Market.OANDA)
            elif self._traded_universe == TradedUniverse.FX_FUTURES:
                data: Security = self.add_data(data_tools.QuantpediaFutures, symbol, Resolution.DAILY)
            data.set_fee_model(data_tools.CustomFeeModel())
            ir_symbol: Symbol = self.add_data(data_tools.InterestRate3M, rate_symbol, Resolution.DAILY).symbol
            self._data[data.symbol] = data_tools.SymbolData(period, ir_symbol)
        self.settings.daily_precise_end_time = False
        self.settings.minimum_order_margin_portfolio_percentage = 0.
        self._recent_month: int = -1
    def on_data(self, slice: Slice) -> None:
        if slice.contains_key(self._us_ir) and slice[self._us_ir]:
            for symbol, symbol_data in self._data.items():
                if slice.contains_key(symbol_data._ir_symbol) and slice[symbol_data._ir_symbol]:
                    symbol_data.update_values(
                        slice[symbol_data._ir_symbol].value - slice[self._us_ir].value, slice[symbol_data._ir_symbol].value
                    )
        if self._traded_universe == TradedUniverse.FX:
            if not self.securities[list(self._data.keys())[0]].exchange.hours.is_open(self.time, extended_market_hours=False):
                return
        # monthly rebalance
        if self._recent_month == self.time.month:
            return
        self._recent_month = self.time.month
        last_update_date: Dict[str, datetime.date] = data_tools.QuantpediaFutures.get_last_update_date()
        
        EWMA: Dict[Symbol, float] = {
            symbol: data_tools.EWMA_Volatility(symbol_data.get_rate_diff(), self._EWMA_lambda) 
            for symbol, symbol_data in self._data.items() 
            if symbol_data.is_ready()
            and (symbol.value in last_update_date and last_update_date[symbol.value] > self.time.date() if self._traded_universe == TradedUniverse.FX_FUTURES else True)
        }
        
        if len(EWMA) == 0:
            self.log('Not enough data for further calculation.')
            return
        
        expected_ret_df: dataframe = pd.concat(
            [symbol_data.get_expected_returns() for _, symbol_data in self._data.items() if symbol_data.is_ready()], axis=1
        )
        port_opt = data_tools.PortfolioOptimization(expected_ret_df, 0, len(expected_ret_df.columns), np.mean(list(EWMA.values())))
        w: np.ndarray = port_opt.opt_portfolio()
        targets: List[PortfolioTarget] = []
        for i, symbol in enumerate(EWMA):
            if w[i] > self._min_weight:
                if slice.contains_key(symbol) and slice[symbol]:
                    targets.append(PortfolioTarget(symbol, w[i]))
        
        self.set_holdings(targets, True)