“该策略使用八种流动性外汇期货，结合利率套利、动量和均值回归等标准化指标，进行风险预算分配，等权重配置，并每月重新平衡。”
资产类别: 差价合约、远期、期货、掉期 | 地区: 美国 | 周期: 每月 | 市场: 外汇 | 关键词: 外汇
I. 策略概要
该策略投资于芝加哥商品交易所（CME）八种流动性最强的当月外汇期货，涉及多种货币（澳元、英镑、加元、欧元、日元、墨西哥比索、新西兰元和瑞士法郎）。使用多种指标来创建单独的策略，包括利率套利、动量、均值回归、股票动量和商品动量。这些指标通过计算相对于历史值的百分位数得分，并进行-0.5的偏差调整，标准化到-0.5到+0.5的范围。采用风险预算，根据每个指标标准化分数的绝对值进行分配，所有指标的年化总风险目标为10%。这些策略以等权重组合，并每月重新平衡，允许在优化过程中出现负权重。论文中还提供了用于优化的Python代码。
II. 策略合理性
该策略的功能源于使用一套从各种策略中提取的标准化指标，这些指标成功预测了外汇期货回报。套利策略是一种行之有效的方法，涉及在低利率经济体借款并在高利率经济体放贷。这在股票市场回报和外汇回报之间建立了联系，解释了股票动量的使用。此外，商品价格，特别是农产品、石油和金属，与外汇回报密切相关。发达货币之间的交叉汇率趋于均值回归，进一步支持了该策略的有效性。多策略方法被证明更优越，与表现最佳的单一策略相比，提供了更稳定的一致回报并提高了夏普比率，展示了系统化、多元化方法的优势。
III. 来源论文
Realized Semibetas: Signs of Things to Come [点击查看论文]
- 索南·斯里瓦斯塔瓦（Sonam Srivastava）、高拉夫·查克拉沃尔蒂（Gaurav Chakravorty）、桑奇特·古普塔（Sanchit Gupta）、安基特·阿瓦斯提（Ankit Awasthi），Wright Research，Qplum，Qplum。
<摘要>
在本文中，我们提出了一种系统化的多策略方法，用于管理期货投资组合中的外汇期货交易。我们的核心发现是，与手工设计每个指标相比，结合不同的指标可以获得更多的阿尔法。我们表明，将动量和均值回归等技术指标与外汇套利指标相结合，可以显著优于单个指标。通过端到端的系统化投资组合构建方法，包括指标构建、标准化和组合，我们在无偏的前向回测中评估时，能够将所得投资组合的夏普比率比表现最佳的单一指标提高60%。
IV. 回测表现
| 年化回报 | 9.76% |
| 波动率 | 9.3% |
| β值 | 0.033 |
| 夏普比率 | 0.5 |
| 索提诺比率 | -0.965 |
| 最大回撤 | N/A |
| 胜率 | 48% |
V. 完整的 Python 代码
from AlgorithmImports import *
import numpy as np
from typing import List, Dict, Tuple
from scipy import stats
class AMultiStrategyApproachtoTradingForeignExchangeFutures(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2000, 1, 1)
self.SetCash(100000)
# Symbols - Currency future, equity future, 10Y bond yield, cash rate data.
# Cash rate source: https://www.quandl.com/data/OECD-Organisation-for-Economic-Co-operation-and-Development
# 10Y bond yield source: www.investing.com
self.symbols:List[Tuple[str, str, str, str]] = [
('CME_AD1', 'ASX_YAP1', 'AU10YT', 'IR3TIB01AUM156N'), # Australian Dollar Futures, Continuous Contract #1
('CME_CD1', 'LIFFE_FCE1', 'CA10YT', 'IR3TIB01CAM156N'), # Canadian Dollar Futures, Continuous Contract #1
('CME_SF1', 'EUREX_FSMI1', 'CH10YT', 'IR3TIB01CHM156N'), # Swiss Franc Futures, Continuous Contract #1
('CME_EC1', 'EUREX_FSTX1', 'DE10YT', 'IR3TIB01EZM156N'), # Euro FX Futures, Continuous Contract #1
('CME_BP1', 'LIFFE_Z1', 'GB10YT', 'LIOR3MUKM'), # British Pound Futures, Continuous Contract #1
('CME_JY1', 'SGX_NK1', 'JP10YT', 'IR3TIB01JPM156N'), # Japanese Yen Futures, Continuous Contract #1
]
# Symbol data.
self.data:Dict[Symbol, SymbolData] = {}
self.short_period:int = 3
self.long_period:int = 12
self.leverage:int = 20
# Target risk of the allocation.
self.target_risk:float = 0.1
for currency_future, equity_future, bond_yield_symbol, cash_rate_symbol in self.symbols:
# Currency future data.
data = self.AddData(data_tools.QuantpediaFutures, currency_future, Resolution.Daily)
data.SetFeeModel(data_tools.CustomFeeModel())
data.SetLeverage(self.leverage)
self.data[currency_future] = data_tools.SymbolData(currency_future, self, self.short_period*21, self.long_period*21, True)
# Equity future data.
self.AddData(data_tools.QuantpediaFutures, equity_future, Resolution.Daily)
self.data[equity_future] = data_tools.SymbolData(equity_future, self, self.short_period*21, self.long_period*21, False)
# Bond yield data.
self.AddData(data_tools.QuantpediaBondYield, bond_yield_symbol, Resolution.Daily)
# Interbank rate data.
self.AddData(data_tools.InterestRate3M, cash_rate_symbol, Resolution.Daily)
self.usa_10Y_yield:str = 'US10Y'
self.usa_cash_rate:str = 'IR3TIB01USM156N'
self.AddData(data_tools.QuantpediaBondYield, self.usa_10Y_yield, Resolution.Daily)
self.AddData(data_tools.InterestRate3M, self.usa_cash_rate, Resolution.Daily)
self.commodity_index:Symbol = self.AddEquity('DBC', Resolution.Daily).Symbol
self.data[self.commodity_index] = data_tools.SymbolData(self.commodity_index, self, self.short_period*21, self.long_period*21, False)
self.last_month:int = -1
def OnData(self, data:Slice) -> None:
# Rebalance once a month.
if self.last_month != self.Time.month:
self.last_month = self.Time.month
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()
# Create indicators.
weight:Dict[Symbol, float] = {}
for currency_future, equity_future, bond_yield_symbol, cash_rate_symbol in self.symbols:
# data is still coming
if self.Securities[currency_future].GetLastData() and qp_futures_last_update_date[currency_future] <= self.Time.date() \
or self.Securities[cash_rate_symbol].GetLastData() and ir_last_update_date[cash_rate_symbol] <= self.Time.date():
continue
# If data needed is ready (Momentum and SMA's).
if self.data[currency_future].is_ready() and self.data[equity_future].is_ready() and self.data[self.commodity_index].is_ready():
long_interest_rate_carry:Union[float,None] = self.Securities[bond_yield_symbol].Price - self.Securities[self.usa_10Y_yield].Price if self.Securities.ContainsKey(bond_yield_symbol) and self.Securities.ContainsKey(self.usa_10Y_yield) else None
short_interest_rate_carry:Union[float,None] = self.Securities[cash_rate_symbol].Price - self.Securities[self.usa_cash_rate].Price if self.Securities.ContainsKey(cash_rate_symbol) and self.Securities.ContainsKey(self.usa_cash_rate) else None
curr_short_term_momentum:float = self.data[currency_future].short_term_momentum()
curr_long_term_momentum:float = self.data[currency_future].long_term_momentum()
curr_long_term_reversal:Union[float,None] = self.Securities[currency_future].Price / self.data[currency_future].long_term_sma() if self.Securities.ContainsKey(currency_future) else None
curr_short_term_reversal:Union[float,None] = self.Securities[currency_future].Price / self.data[currency_future].short_term_sma() if self.Securities.ContainsKey(currency_future) else None
equity_short_term_momentum:float = self.data[equity_future].short_term_momentum()
equity_long_term_momentum:float = self.data[equity_future].long_term_momentum()
commodity_short_term_momentum:float = self.data[self.commodity_index].short_term_momentum()
commodity_long_term_momentum:float = self.data[self.commodity_index].long_term_momentum()
if long_interest_rate_carry and short_interest_rate_carry and curr_long_term_reversal and curr_short_term_reversal:
indicator_value:Tuple[float] = (long_interest_rate_carry, short_interest_rate_carry, curr_long_term_momentum, curr_short_term_momentum, \
curr_long_term_reversal, curr_short_term_reversal, equity_long_term_momentum, equity_short_term_momentum, \
commodity_long_term_momentum, commodity_short_term_momentum)
self.data[currency_future].add_indicator_value(indicator_value)
if self.data[currency_future].indicator_history_is_ready():
num_of_properties:int = len(indicator_value)
scores:List[float] = []
for propert_index in range(num_of_properties):
scores.append(abs((stats.percentileofscore([x[propert_index] for x in self.data[currency_future].IndicatorsHistory], indicator_value[propert_index]) / 100 ) - 0.5))
score_sum:float = sum(scores)
scores:np.ndarray = np.array(scores)
scores = (scores / score_sum) * self.target_risk
risk_budget:float = sum(scores)
weight[currency_future] = risk_budget
# Trade execution.
invested:List[Symbol] = [x.Key for x in self.Portfolio if x.Value.Invested]
for symbol in invested:
if symbol not in weight:
self.Liquidate(symbol)
for symbol, w in weight.items():
self.SetHoldings(symbol, w)