“该策略利用利率、动量和反转信号优化货币投资组合，通过滚动的20年训练窗口进行分析，并每年重新估算以更新预测，旨在最大化投资者效用。”
资产类别：差价合约（CFDs）、远期合约、期货、掉期 | 地区：全球 | 频率：每月 | 市场：外汇 | 关键词：投资组合
I. 策略概述
该策略关注来自17个国家的货币，以以下四个变量作为输入：
- Sign（利率差方向）
- fd（标准化的利率差）
- mom（3个月动量，标准化）
- rev（5年长期反转，排除近期动量影响）
通过参数化投资组合政策，确定最优投资组合权重，以最大化投资者效用，效用公式为：
U = (1 + r_p) ^ (1−γ) / (1−γ)
其中，r_p 是投资组合收益率，γ 是风险厌恶系数（设定为5）。模型采用滚动20年历史数据窗口进行训练，预测并设计未来12个月的投资组合，并每年使用扩展的历史数据重新估算，以适应市场变化。
II. 策略合理性
该策略通过分析利率差、动量和长期反转信号优化货币投资组合。基于效用最大化方法，计算货币的最优权重，兼顾收益和风险。投资者效用函数结合了投资组合的回报和相对风险厌恶，风险厌恶系数设定为5，反映了投资者对高回报与低风险的平衡需求。通过滚动窗口和年度重新估算，模型动态适应市场变化，确保投资组合与最新预测一致，并基于货币特定的风险与表现特征实现回报最大化。
III. 论文来源
Beyond the Carry Trade: Optimal Currency Portfolios [点击浏览原文]
- 作者：Pedro Barroso 和 Pedro Santa-Clara
- 机构：葡萄牙天主教大学里斯本商学院（CATÓLICA-LISBON School of Business & Economics）、诺瓦商学院（Nova School of Business and Economics）、美国国家经济研究局（NBER）、经济政策研究中心（CEPR）
<摘要>
我们测试了技术性和基本面变量在构建货币投资组合中的重要性。研究发现，套息交易、动量和反转对投资组合表现均有贡献，而实际汇率和经常账户的影响不显著。由此产生的最优投资组合显著优于单一套息交易和其他货币投资策略。这表明，通过结合不同的货币特征，投资者可以在动态市场中实现更高的回报，同时有效管理风险。
IV. 回测表现
| 年化收益率 | 19.2% |
| 波动率 | 22.2% |
| Beta | 0.05 |
| 夏普比率 | 1.4 |
| 索提诺比率 | -0.289 |
| 最大回撤 | N/A |
| 胜率 | 59% |
V. 完整python代码
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
