“该策略投资于10种外币，利用利率差异和协方差矩阵分析进行均值-方差优化，确定每月再平衡的最佳多头/空头投资组合权重，以美元为基准。”
资产类别: 差价合约、远期合约、期货、互换 | 地区: 全球 | 周期: 每月 | 市场: 外汇 | 关键词: 均值方差，套利
I. 策略概要
该策略以美元（USD）为本币，投资于10种外币：瑞士法郎（CHF）、欧元（EUR）、日元（JPY）、英镑（GBP）、澳元（AUD）、加元（CAD）、挪威克朗（NOK）、瑞典克朗（SEK）、新加坡元（SGD）和新西兰元（NZD）。每月，在随机游走假设下，每种货币的预期超额回报计算为与美国无风险利率的利率差。使用过去250个观测值重新估计每日协方差矩阵。投资者应用均值-方差优化来确定每种货币的最佳多头/空头投资组合权重。投资组合每月进行再平衡，利用利率差异和协方差动态来最大化风险调整后的回报。
II. 策略合理性
学术研究表明，无抛补利率平价预测汇率会进行调整以消除套利交易利润，但数据与此相悖。实证证据表明，短期汇率波动是不可预测的，并遵循随机游走。尽管存在汇率风险，但套利交易策略通常平均产生正的预期回报，使投资者受益。
III. 来源论文
套利交易的风险与回报研究 [点击查看论文]
- 阿克曼、波尔、施梅德斯、苏黎世州立银行、挪威经济学院、IMD
<摘要>
德米格尔、加拉皮和乌帕尔（《金融研究评论》，22（2009），1915-1953）表明，在股票市场中，由于估计误差，使用均值-方差分析构建的优化投资组合很难跑赢简单的等权重投资组合。在本文中，我们证明投资组合优化可以在货币市场中发挥作用。这两个设置之间的关键区别在于，在货币市场中，利率提供了对未来回报的预测，而这种预测没有估计误差，这允许应用均值-方差分析。我们表明，在过去的26年中，以这种方式构建的均值-方差有效投资组合的夏普比率为0.91，而等权重投资组合的夏普比率仅为0.15。我们还考虑了该策略的实际实施。
IV. 回测表现
| 年化回报 | 1.7% |
| 波动率 | 1.87% |
| β值 | 0.197 |
| 夏普比率 | 0.91 |
| 索提诺比率 | -0.298 |
| 最大回撤 | N/A |
| 胜率 | 49% |
V. 完整的 Python 代码
from AlgorithmImports import *
import pandas as pd
from collections import deque
import data_tools
class MeanVarianceCarryTradeStrategy(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2009, 1, 1)
self.SetCash(100000)
self.tickers:dict[str, str] = {
'CME_AD1' : Futures.Currencies.AUD, # Australian Dollar Futures, Continuous Contract #1
'CME_BP1' : Futures.Currencies.GBP, # British Pound Futures, Continuous Contract #1
'CME_CD1' : Futures.Currencies.CAD, # Canadian Dollar Futures, Continuous Contract #1
'CME_EC1' : Futures.Currencies.EUR, # Euro FX Futures, Continuous Contract #1
'CME_JY1' : Futures.Currencies.JPY, # Japanese Yen Futures, Continuous Contract #1
'CME_MP1' : Futures.Currencies.MXN, # Mexican Peso Futures, Continuous Contract #1
'CME_NE1' : Futures.Currencies.NZD, # New Zealand Dollar Futures, Continuous Contract #1
'CME_SF1' : Futures.Currencies.CHF, # Swiss Franc Futures, Continuous Contract #1
}
self.period:int = 250
self.max_missing_days:int = 5
self.min_expiration_days:int = 0
self.max_expiration_days:int = 360
self.futures_data:dict[str, FuturesData] = {}
for qp_ticker, qc_ticker in self.tickers.items():
# quantpedia #1 Contract
data = self.AddData(data_tools.QuantpediaFutures, qp_ticker, Resolution.Daily)
data.SetFeeModel(data_tools.CustomFeeModel())
data.SetLeverage(5)
quantpedia_future_symbol:Symbol = data.Symbol
# QC futures
future:Future = self.AddFuture(qc_ticker, Resolution.Daily, dataNormalizationMode=DataNormalizationMode.Raw)
future.SetFilter(timedelta(days=self.min_expiration_days), timedelta(days=self.max_expiration_days))
self.futures_data[future.Symbol.Value] = data_tools.FuturesData(quantpedia_future_symbol, self.period)
self.recent_month:int = -1
def FindAndUpdateContracts(self, futures_chain, ticker) -> None:
near_contract:FuturesContract = None
dist_contract:FuturesContract = None
if ticker in futures_chain:
contracts:list[:FuturesContract] = [contract for contract in futures_chain[ticker] if contract.Expiry.date() > self.Time.date()]
if len(contracts) >= 2:
contracts:list[:FuturesContract] = sorted(contracts, key=lambda x: x.Expiry, reverse=False)
near_contract = contracts[0]
dist_contract = contracts[1]
self.futures_data[ticker].update_contracts(near_contract, dist_contract)
def OnData(self, data):
# daily update QC futures data
if data.FutureChains.Count > 0:
for ticker, future_obj in self.futures_data.items():
# check if near contract is expired or is not initialized
if not future_obj.is_initialized() or \
(future_obj.is_initialized() and future_obj.near_contract.Expiry.date() == self.Time.date()):
self.FindAndUpdateContracts(data.FutureChains, ticker)
# update QC futures rolling return
if future_obj.is_initialized():
near_c:FuturesContract = future_obj.near_contract
dist_c:FuturesContract = future_obj.distant_contract
if near_c.Symbol in data and data[near_c.Symbol] and dist_c.Symbol in data and data[dist_c.Symbol]:
raw_price1:float = data[near_c.Symbol].Value * self.Securities[ticker].SymbolProperties.PriceMagnifier
raw_price2:float = data[dist_c.Symbol].Value * self.Securities[ticker].SymbolProperties.PriceMagnifier
if raw_price1 != 0 and raw_price2 != 0:
daily_return:float = raw_price1 / raw_price2 - 1
future_obj.update_roll_return(daily_return)
# check if quantpedia data still coming
for _, future_obj in self.futures_data.items():
quantpedia_future:Symbol = future_obj.quantpedia_future
# if quantpedia_future in data and data[quantpedia_future]:
# future_obj.update_quantpedia_last_update(self.Time.date())
# rebalance monthly
if self.recent_month != self.Time.month:
self.recent_month = self.Time.month
self.Rebalance()
def Rebalance(self):
custom_data_last_update_date: Dict[Symbol, datetime.date] = data_tools.QuantpediaFutures.get_last_update_date()
self.Liquidate()
ready_futures:dict[Symbol, list] = {}
# filter ready futures and reset futures, which do not recieve new data
for _, future_obj in self.futures_data.items():
data_ready_flag = future_obj.is_ready()
if self.securities[future_obj.quantpedia_future].get_last_data() and self.time.date() > custom_data_last_update_date[future_obj.quantpedia_future]:
self.liquidate()
return
if data_ready_flag:
ready_futures[future_obj.quantpedia_future] = future_obj.reverse_roll_return()
elif data_ready_flag:
future_obj.reset_data()
ready_futures_length:int = len(ready_futures)
if ready_futures_length == 0: return
df:pd.dataframe = pd.dataframe(ready_futures, columns=ready_futures.keys())
optimization:data_tools.PortfolioOptimization = data_tools.PortfolioOptimization(df, 0, ready_futures_length)
opt_weight:list[float] = optimization.opt_portfolio()
if sum(opt_weight) == 0: return
for index in range(ready_futures_length):
weight:float = opt_weight[index]
if weight >= 0.001:
self.SetHoldings(df.columns[index], weight)