“该策略使用15种发达国家货币，应用均值-方差优化，预期回报来自远期贴现，协方差矩阵根据过去的汇率波动估算。投资组合每月重新平衡。”
资产类别: 远期、期货、互换 | 地区: 全球 | 周期: 每月 | 市场: 外汇 | 关键词: 均值、方差
I. 策略概要
投资范围包括15种发达国家货币。为了确定投资组合权重，投资者使用均值-方差优化，预期回报等于远期贴现，协方差矩阵基于六个月内平方汇率增长的指数加权移动平均值。协方差矩阵被特征分解为特征向量（W矩阵）和特征值（lambda矩阵），并移除解释方差小于1%的主成分。修改后的协方差矩阵用于计算投资组合权重，并根据投资者的相对风险厌恶进行调整。投资组合每月重新平衡。
II. 策略合理性
该策略通过利用远期贴现（近似等于预期回报）在外汇市场中使用均值-方差优化，有助于减少估计误差。本文还解决了协方差矩阵估计误差的问题，并建议使用主成分分析（PCA）来更好地捕捉外汇市场风险。PCA移除了低方差成分，提高了协方差矩阵的稳健性，并有助于避免接近套利的机会。这种方法提高了样本外夏普比率，并将偏度转为正值。本文还强调了市场择时在提高业绩方面的重要性，表明风险限制（如保持恒定的名义价值）对回报产生负面影响。与其他学术策略相比，该方法在考虑交易成本时表现更优，并且在包括NBER衰退和欧元推出后的各个时期内都保持盈利。此外，在1983年至2016年的较短样本期内，它仍然表现良好。
III. 来源论文
Market Timing and Predictability in FX Markets [点击查看论文]
- 毛瑞尔（Thomas Andreas Maurer），杜思源（Thuy Duong To）和陈玉香（Ngoc-Khanh Tran），香港大学；圣路易斯华盛顿大学 – 约翰·M·奥林商学院；伦敦政治经济学院（LSE），新南威尔士大学，悉尼；金融研究网络（FIRN），弗吉尼亚理工大学帕姆普林商学院金融系；奥林商学院 – 圣路易斯华盛顿大学
<摘要>
我们研究了外汇市场中市场择时的经济价值，即利用关于条件夏普比率的信息来调整条件均值-方差有效货币投资组合的名义价值。当条件风险回报权衡更有利（不利）时，我们的策略交易更积极（消极）。这导致样本外无条件夏普比率、偏度和每1%预期超额回报的最大回撤显著改善。该策略的市场择时预测外汇市场的回报、波动性和偏度。流行的货币定价因子无法解释该策略的高平均超额回报。我们的研究结果表明，在构建货币交易策略时，施加杠杆或风险（即条件波动率）限制或其他较差的市场择时政策是代价高昂的。
IV. 回测表现
| 年化回报 | 7.44% |
| 波动率 | 8.12% |
| β值 | 0.065 |
| 夏普比率 | 0.92 |
| 索提诺比率 | -0.597 |
| 最大回撤 | 16.87% |
| 胜率 | 66% |
V. 完整的 Python 代码
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)