“该策略利用货币远期、股票指数和债券期货之间的关系，在资产类别之间平均分配风险，优化权重，并根据3-4年内的溢出效应和累计回报每周重新平衡。”
资产类别: 差价合约、远期、期货、掉期 | 地区: 全球 | 周期: 每周 | 市场: 债券、外汇、股票 | 关键词: 趋势跟踪、溢出效应
I. 策略概要
该策略涉及九种汇率的货币远期、11个发达国家股票指数以及多个国家的各种债券期货。资产类别之间的关系被识别：债券对汇率产生负面影响，对股票产生正面影响；股票对债券和汇率都产生负面影响；汇率对债券和股票都产生正面影响。该策略使用3到4年的回溯期内的累计回报来定义信号，其正负效应取决于资产之间的关系。该策略将总风险预算的三分之一分配给每个资产类别，确保资产类别内部和之间的风险贡献相等。权重经过优化以最大化绝对权重的自然对数之和。本文考虑了资产类别之间的溢出效应，从六种潜在的溢出情景中形成策略。该策略每周重新平衡，确保根据不断变化的市场动态及时调整。
II. 策略合理性
该策略基于债券、股票和外汇之间的溢出效应。债券对汇率产生负面影响，因为紧缩的货币政策导致美元升值，而套利交易则证明了这一点。债券对股票产生正面影响，因为较低的利率会促进企业融资、复苏和盈利能力。股票对债券产生负面影响，因为强劲的股票表现预示着通货膨胀，从而提高利率并降低债券价值。尽管股票和外汇通常具有正的短期相关性，但表现良好的股票可能导致货币贬值。外汇对债券产生正面影响，因为货币贬值缓解了央行降息压力。最后，外汇对股票产生正面影响，货币升值预示着经济健康。尽管某些关系在数据中无法观察到，但该策略是盈利的，特别是由于三个有效的溢出关系：债券对股票、股票对外汇以及外汇对股票。综合策略也证明是盈利的，利用了这些联系。
III. 来源论文
Trend-Following and Spillover Effects [点击查看论文]
- 菲利普·德克莱克（Philippe Declerck），汇丰环球资产管理公司（HSBC Global Asset Management）
<摘要>
我们首先记录了政府债券、货币和股票指数（所有发达国家）在资产类别层面和多资产层面的趋势跟踪（或时间序列动量），使用29种流动性工具，回溯期从1到60个月不等。典型的多资产趋势跟踪策略在短期到中期回溯期内能带来强劲回报。我记录了趋势会溢出到其他资产类别：资产的过去趋势有助于使用其他相关资产构建投资策略。这种溢出效应在使用比趋势跟踪最佳点更长的回溯期时效果更好。
IV. 回测表现
| 年化回报 | 3.2% |
| 波动率 | 4.8% |
| β值 | -0.001 |
| 夏普比率 | 0.67 |
| 索提诺比率 | -0.811 |
| 最大回撤 | -12.3% |
| 胜率 | 53% |
V. 完整的 Python 代码
from AlgorithmImports import *
import numpy as np
class TrendFollowingandSpilloverEffect(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2000, 1, 1)
self.SetCash(100000)
# Symbols - currency, index and bond futures.
self.symbols = [
('CME_AD1', 'ASX_YAP1', 'ASX_XT1'), # Australian Dollar Futures, Continuous Contract #1
('CME_BP1', 'LIFFE_Z1', 'LIFFE_R1'), # British Pound Futures, Continuous Contract #1
('CME_CD1', 'LIFFE_FCE1', 'MX_CGB1'), # Canadian Dollar Futures, Continuous Contract #1
('CME_EC1', 'EUREX_FSTX1', 'EUREX_FGBL1'), # Euro FX Futures, Continuous Contract #1
('CME_JY1', 'SGX_NK1', 'SGX_JB1'), # Japanese Yen Futures, Continuous Contract #1
('CME_DX1', 'CME_ES1', 'CME_TY1') # US Dollar Index Futures, Continuous Contract #1
# ('CME_SF1', 'EUREX_FSMI1', '') # Swiss Franc Futures, Continuous Contract #1
# ('CME_MP1', '', '') # Mexican Peso Futures, Continuous Contract #1
# ('CME_NE1', '', '') # New Zealand Dollar Futures, Continuous Contract #
]
self.symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
# Daily ROC data.
self.data = {}
self.period = 36 * 21
self.SetWarmUp(self.period)
for futures_symbols in self.symbols:
for symbol in futures_symbols:
data = self.AddData(QuantpediaFutures, symbol, Resolution.Daily)
self.data[symbol] = SymbolData(self.period)
data.SetFeeModel(CustomFeeModel())
data.SetLeverage(5)
self.rebalance_flag: bool = False
self.Schedule.On(self.DateRules.WeekStart(self.symbol), self.TimeRules.AfterMarketOpen(self.symbol), self.Rebalance)
self.settings.daily_precise_end_time = False
self.settings.minimum_order_margin_portfolio_percentage = 0.
def OnData(self, data):
for futures_symbols in self.symbols:
for symbol in futures_symbols:
if self.securities[symbol].get_last_data() and self.time.date() > QuantpediaFutures.get_last_update_date()[symbol]:
self.liquidate(symbol)
self.data[symbol].reset()
continue
if symbol in data and data[symbol]:
price = data[symbol].Value
self.data[symbol].update(price)
if not self.rebalance_flag:
return
self.rebalance_flag = False
weight = {}
traded_asset_classs_count = 0
for futures_symbols in self.symbols:
fx = futures_symbols[0]
eq = futures_symbols[1]
bond = futures_symbols[2]
if self.data[fx].is_ready() and self.data[eq].is_ready() and self.data[bond].is_ready():
fx_perf = self.data[fx].performance()
eq_perf = self.data[eq].performance()
bond_perf = self.data[bond].performance()
bond_w = 0
fx_w = 0
eq_w = 0
# Bonds have a negative effect on FX and positive effect on Equities
bond_signum = np.sign(bond_perf)
fx_w -= bond_signum
eq_w += bond_signum
# Equities have a negative effect on Bonds and negative effect on FX
eq_signum = np.sign(eq_perf)
bond_w -= eq_signum
fx_w -= eq_signum
# FX has a positive effect on Equities and positive effect on Bonds
fx_signum = np.sign(fx_perf)
eq_w += fx_signum
bond_w += fx_signum
# inverse volatility sum of traded symbols
total_volatility = sum([ 1/self.data[x[0]].volatility() for x in [(fx,fx_w), (eq,eq_w), (bond, bond_w)] if x[1] != 0 ])
# volatility weighting
if total_volatility != 0:
weight[fx] = ((1/self.data[fx].volatility()) / total_volatility) * np.sign(fx_w)
weight[eq] = ((1/self.data[eq].volatility()) / total_volatility) * np.sign(eq_w)
weight[bond] = ((1/self.data[bond].volatility()) / total_volatility) * np.sign(bond_w)
traded_asset_classs_count += 1
portfolio: List[PortfolioTarget] = []
if traded_asset_classs_count != 0:
weight_ratio = 1 / traded_asset_classs_count
portfolio = [PortfolioTarget(symbol, weight_ratio * w) for symbol, w in weight.items() if data.contains_key(symbol) and data[symbol]]
self.SetHoldings(portfolio, True)
def Rebalance(self):
self.rebalance_flag = True
class SymbolData():
def __init__(self, period):
self.price = RollingWindow[float](period)
self.period = period
def update(self, value) -> None:
self.price.Add(value)
def performance(self) -> float:
result = self.price[0] / self.price[self.period-1] - 1
return result
def volatility(self) -> float:
prices = np.array([x for x in self.price][:60])
result = prices[:-1] / prices[1:] - 1
result = np.std(result) * np.sqrt(252)
return result
def reset(self) -> None:
self.price.reset()
def is_ready(self) -> bool:
return self.price.IsReady
# Quantpedia data.
# NOTE: IMPORTANT: Data order must be ascending (datewise)
class QuantpediaFutures(PythonData):
_last_update_date:Dict[Symbol, datetime.date] = {}
@staticmethod
def get_last_update_date() -> Dict[Symbol, datetime.date]:
return QuantpediaFutures._last_update_date
def GetSource(self, config, date, isLiveMode):
return SubscriptionDataSource("data.quantpedia.com/backtesting_data/futures/{0}.csv".format(config.Symbol.Value), SubscriptionTransportMedium.RemoteFile, FileFormat.Csv)
def Reader(self, config, line, date, isLiveMode):
data = QuantpediaFutures()
data.Symbol = config.Symbol
if not line[0].isdigit(): return None
split = line.split(';')
data.Time = datetime.strptime(split[0], "%d.%m.%Y") + timedelta(days=1)
data['back_adjusted'] = float(split[1])
data['spliced'] = float(split[2])
data.Value = float(split[1])
if config.Symbol.Value not in QuantpediaFutures._last_update_date:
QuantpediaFutures._last_update_date[config.Symbol.Value] = datetime(1,1,1).date()
if data.Time.date() > QuantpediaFutures._last_update_date[config.Symbol.Value]:
QuantpediaFutures._last_update_date[config.Symbol.Value] = data.Time.date()
return data
# Custom fee model
class CustomFeeModel(FeeModel):
def GetOrderFee(self, parameters):
fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
return OrderFee(CashAmount(fee, "USD"))