“该策略使用过去的股票和债券回报来指导20个工业国家的股票或债券投资。头寸持有一个月，目标波动率为10%，国家份额等权重。”
资产类别: ETF、期货 | 地区: 全球 | 周期: 每月 | 市场: 债券、股票 | 关键词: 动量
I. 策略概要
投资范围包括20个主要工业国家的债券和股票指数。该策略依赖于过去的回报作为预测指标。如果过去12个月股票回报为负，债券回报为正，投资者做多债券。如果股票和债券回报均为正，投资者做多股票。在其他情况下，资本存入以美元计价的保证金账户，赚取美国无风险利率。头寸持有一个月，目标波动率为10%，最终投资组合中各国份额等权重。
II. 策略合理性
跨资产时间序列动量效应受到债券和股票市场中缓慢移动的资本的影响，这些资本受到诸如注意力不集中、决策延迟和资本市场摩擦等因素的驱动。正的债券回报通过降低借贷成本和增加贷款抵押品来帮助受约束的投资者，这反过来又刺激了股票需求和回报。当利率下降时，投资者可以利用更多的杠杆投资股票。股票正回报与未来贷款活动相关，推高债券收益率，导致债券负回报。相反，股票市场回报与无风险利率的变化相关，这在股票市场表现良好后对债券回报产生负面影响。
III. 来源论文
Cross-Asset Signals and Time Series Momentum [点击查看论文]
- 皮特卡耶尔维（Pitkäjärvi）、苏奥米嫩（Suominen）、瓦伊蒂宁（Vaittinen），阿姆斯特丹自由大学；丁伯根研究所，阿尔托大学商学院，独立学者。
<摘要>
我们记录了债券和股票市场中的一种新现象，我们称之为跨资产时间序列动量。使用来自20个国家的数据，我们表明过去的债券市场回报是未来股票市场回报的正预测指标，而过去的股票市场回报是未来债券市场回报的负预测指标。我们利用这种可预测性构建了一个多元化的跨资产时间序列动量投资组合，其夏普比率比标准时间序列动量投资组合高45%。我们提供的证据表明，时间序列动量和跨资产时间序列动量是由债券和股票市场中缓慢移动的资本驱动的。
IV. 回测表现
| 年化回报 | 6.5% |
| 波动率 | 10% |
| β值 | 0.093 |
| 夏普比率 | 0.65 |
| 索提诺比率 | 0.077 |
| 最大回撤 | N/A |
| 胜率 | 59% |
V. 完整的 Python 代码
import numpy as np
from AlgorithmImports import *
class GlobalCrossAssetTimeSeriesMomentum(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2000, 1, 1)
self.SetCash(100000)
self.symbols = {
"ASX_YAP1" : "ASX_XT1", # 10 Year Commonwealth Treasury Bond Futures, Continuous Contract #1 (Australia)
"LIFFE_FCE1" : "MX_CGB1", # Ten-Year Government of Canada Bond Futures, Continuous Contract #1 (Canada)
"EUREX_FSTX1" : "EUREX_FGBL1", # Euro-Bund (10Y) Futures, Continuous Contract #1 (Germany)
"SGX_NK1" : "SGX_JB1", # SGX 10-Year Mini Japanese Government Bond Futures, Continuous Contract #1 (Japan)
"LIFFE_Z1" : "LIFFE_R1", # Long Gilt Futures, Continuous Contract #1 (U.K.)
"CME_ES1" : "CME_TY1" # 10 Yr Note Futures, Continuous Contract #1 (USA)
}
self.data = {}
self.period = 12*21
self.SetWarmUp(self.period)
self.leverage_cap = 5
for eq in self.symbols:
bond = self.symbols[eq]
data = self.AddData(QuantpediaFutures, eq, Resolution.Daily)
self.data[eq] = RollingWindow[float](self.period)
data.SetFeeModel(CustomFeeModel())
data.SetLeverage(self.leverage_cap)
data = self.AddData(QuantpediaFutures, bond, Resolution.Daily)
self.data[bond] = RollingWindow[float](self.period)
data.SetFeeModel(CustomFeeModel())
data.SetLeverage(self.leverage_cap)
first_key = [x for x in self.symbols.keys()][0]
self.rebalance_flag: bool = False
self.Schedule.On(self.DateRules.MonthStart(first_key), self.TimeRules.At(0, 0), self.Rebalance)
def OnData(self, data):
for eq in self.symbols:
bond = self.symbols[eq]
if eq in data and bond in data:
if data[eq] and data[bond]:
eq_price = data[eq].Value
bond_price = data[bond].Value
if eq_price != 0 and bond_price != 0:
self.data[eq].Add(eq_price)
self.data[bond].Add(bond_price)
if not self.rebalance_flag:
return
self.rebalance_flag = False
volatility = {}
for eq in self.symbols:
bond = self.symbols[eq]
if all([self.data[x].IsReady and self.Securities[x].GetLastData() and self.Time.date() < QuantpediaFutures.get_last_update_date()[x] for x in [eq, bond]]):
eq_prices = np.array([x for x in self.data[eq]])
bond_prices = np.array([x for x in self.data[bond]])
eq_return = eq_prices[0] / eq_prices[-1] - 1
bond_return = bond_prices[0] / bond_prices[-1] - 1
if eq_return < 0 and bond_return > 0:
bond_returns = bond_prices[:-1] / bond_prices[1:] - 1
volatility[bond] = np.std(bond_returns) * np.sqrt(252)
elif eq_return > 0 and bond_return > 0:
eq_returns = eq_prices[:-1] / eq_prices[1:] - 1
volatility[eq] = np.std(eq_returns) * np.sqrt(252)
if len(volatility) == 0: return
mean_vol = np.mean([x[1] for x in volatility.items()])
# leverage = (0.0833 / total_vol_annualized) * 100
leverage = min((0.1 / mean_vol), self.leverage_cap)
self.Liquidate()
count = len(volatility)
for symbol in volatility:
if data.contains_key(symbol) and data[symbol]:
# self.SetHoldings(symbol, 0.1667 * (1/count) * leverage)
self.SetHoldings(symbol, leverage / count)
def Rebalance(self):
self.rebalance_flag = True
# 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"))