“该策略涉及根据12个月股票回报的z分数交易10年期债券。头寸根据z分数的符号和大小做多或做空，每月重新平衡。”
资产类别: 债券、期货 | 地区: 全球 | 周期: 每月 | 市场: 债券 | 关键词: 债券、股票
I. 策略概要
该投资范围包括来自美国、英国、德国、日本、加拿大和澳大利亚的10年期债券。头寸基于过去12个月股票回报的z分数，计算方法是：12个月回报减去10年平均回报，然后除以标准差。z分数上限为1（-1）以避免极端情况。正z分数导致债券空头头寸，而负z分数导致多头头寸。头寸规模由z分数的绝对值决定。投资组合等权重，每月重新平衡。还提出了仅使用z分数符号或无上限分数的变体。
II. 策略合理性
股票和债券在市场上充当替代品：当股票表现良好时，债券需求下降，导致其价格下跌；反之，在股市崩盘期间则相反。研究证实，这种可预测性在不同的经济条件下都表现稳健，包括衰退、扩张和不同通胀时期。研究结果表明，债券市场择时策略并非由于结构性债券风险，而是反映了真实的市场行为，在市场波动较大时回报更为显著。这突出了根据股市表现趋势择时进入和退出债券市场的能力。
III. 来源论文
Predicting Bond Returns: 70 years of International Evidence [点击查看论文]
- Guido Baltussen, Martin Martens, Olaf Penninga，鹿特丹伊拉斯姆斯大学（EUR）；北方信托公司 – 北方信托资产管理，鹿特丹伊拉斯姆斯大学，鹿博资产管理
<摘要>
我们通过对主要债券市场70年国际数据的深入研究，考察了政府债券回报的可预测性。利用基于经济交易的测试框架，我们发现了债券回报可预测性的强有力经济和统计证据，自1950年以来夏普比率为0.87。这一发现对市场和时间段都具有稳健性，包括30年的国际债券市场样本外数据和另外九个国家的数据。此外，结果在各种经济环境中保持一致，包括利率长期上升或下降的时期，并且在扣除交易成本后仍可利用。可预测性与通胀和经济增长的可预测性相关。总的来说，政府债券溢价显示出可预测的动态，国际债券市场回报的择时为投资者提供了可利用的机会。
IV. 回测表现
| 年化回报 | 3.8% |
| 波动率 | 10% |
| β值 | -0.068 |
| 夏普比率 | 0.38 |
| 索提诺比率 | -0.101 |
| 最大回撤 | N/A |
| 胜率 | 69% |
V. 完整的 Python 代码
import numpy as np
from collections import deque
from AlgorithmImports import *
class PredictingBondReturnswithEquityReturn(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2000, 1, 1)
self.SetCash(100000)
# Bond future and equity etf.
self.symbols = [
("ASX_XT1", 'EWA'), # 10 Year Commonwealth Treasury Bond Futures, Continuous Contract #1 (Australia)
("MX_CGB1", 'EWC'), # Ten-Year Government of Canada Bond Futures, Continuous Contract #1 (Canada)
("EUREX_FGBL1", 'EWG'), # Euro-Bund (10Y) Futures, Continuous Contract #1 (Germany)
("LIFFE_R1", 'EWU'), # Long Gilt Futures, Continuous Contract #1 (U.K.)
("SGX_JB1", 'EWJ'), # SGX 10-Year Mini Japanese Government Bond Futures, Continuous Contract #1 (Japan)
("CME_TY1", 'SPY') # 10 Yr Note Futures, Continuous Contract #1 (USA)
]
# Monthly price data.
self.data = {}
self.month_period = 5
self.period = self.month_period * 12 + 1
self.SetWarmUp(self.period * 21)
for bond_future, equity_etf in self.symbols:
data = self.AddData(QuantpediaFutures, bond_future, Resolution.Daily)
data.SetFeeModel(CustomFeeModel())
data.SetLeverage(10)
# Equity data.
self.AddEquity(equity_etf, Resolution.Daily)
self.data[equity_etf] = deque(maxlen = self.period)
self.last_month = -1
self.Schedule.On(self.DateRules.MonthStart(self.symbols[0][1]), self.TimeRules.At(0, 0), self.Rebalance)
def OnData(self, data):
# Update only on new month start.
if self.Time.month == self.last_month:
return
self.last_month = self.Time.month
# Store monthly data.
for bond_future, equity_etf in self.symbols:
if equity_etf in data and data[equity_etf]:
price = data[equity_etf].Value
self.data[equity_etf].append(price)
def Rebalance(self):
# Z score calc.
weight = {}
for bond_future, equity_etf in self.symbols:
if self.Securities[bond_future].GetLastData() and self.time.date() < QuantpediaFutures.get_last_update_date()[bond_future]:
# At least 3 years of data is ready.
minimum_data_count = ((self.period-1) / self.month_period) * 3
if len(self.data[equity_etf]) >= minimum_data_count:
closes = [x for x in self.data[equity_etf]]
separete_yearly_returns = [Return(closes[x:x+13]) for x in range(0, len(closes),1)]
return_mean = np.mean(separete_yearly_returns)
return_std = np.std(separete_yearly_returns)
z_score = (separete_yearly_returns[-1] - return_mean) / return_std
if z_score > 1: z_score = 1
elif z_score < -1: z_score = -1
weight[bond_future] = -1 * z_score
# Trade execution
invested = [x.Key.Value for x in self.Portfolio if x.Value.Invested]
for symbol in invested:
if symbol not in weight:
self.Liquidate(symbol)
for symbol, weight in weight.items():
self.SetHoldings(symbol, weight)
# Custom fee model
class CustomFeeModel(FeeModel):
def GetOrderFee(self, parameters):
fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
return OrderFee(CashAmount(fee, "USD"))
# 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
def Return(values):
return (values[-1] - values[0]) / values[0]