“该策略涉及根据收益率曲线陡峭度、债券、股票和商品回报的综合z分数，投资于特定国家的10年期债券，头寸每月重新平衡并限制在极端情况。”
资产类别: 债券、期货 | 地区: 全球 | 周期: 每月 | 市场: 债券 | 关键词: 债券
I. 策略概要
该投资范围包括来自美国、英国、德国、日本、加拿大和澳大利亚的10年期债券。头寸基于组合模型z分数开立，该分数计算为四个变量的等权重平均值：收益率曲线陡峭度、过去债券回报、过去股票回报和过去商品回报（仅使用符号）。为避免极端值，z分数上限设置为1和-1。头寸规模由z分数的绝对值决定。投资组合等权重，每月重新平衡。
II. 策略合理性
组合模型受益于使用多种简单策略，从而获得比任何单一策略更高的回报，这在学术和实践文献中都有体现。该策略的有效性通过优于其他单一策略的结果得到证实。该模型的可预测性在不同的市场条件下都保持稳健，包括衰退和扩张时期、高通胀和低通胀时期，以及股票牛市和熊市。重要的是，这种可预测性并非由于结构性债券风险，而是反映了真实的债券市场择时。该策略在市场大幅波动期间表现更好，从而增强了其在动态市场环境中的价值。
III. 来源论文
Predicting Bond Returns: 70 years of International Evidence [点击查看论文]
- Guido Baltussen, Martin Martens, Olaf Penninga，鹿特丹伊拉斯姆斯大学（EUR）；北方信托公司 – 北方信托资产管理，鹿特丹伊拉斯姆斯大学，鹿博资产管理
<摘要>
我们通过对主要债券市场70年国际数据的深入研究，考察了政府债券回报的可预测性。利用基于经济交易的测试框架，我们发现了债券回报可预测性的强有力经济和统计证据，自1950年以来夏普比率为0.87。这一发现对市场和时间段都具有稳健性，包括30年的国际债券市场样本外数据和另外九个国家的数据。此外，结果在各种经济环境中保持一致，包括利率长期上升或下降的时期，并且在扣除交易成本后仍可利用。可预测性与通胀和经济增长的可预测性相关。总的来说，政府债券溢价显示出可预测的动态，国际债券市场回报的择时为投资者提供了可利用的机会。
IV. 回测表现
| 年化回报 | 8.7% |
| 波动率 | 10% |
| β值 | -0.01 |
| 夏普比率 | 0.87 |
| 索提诺比率 | -0.506 |
| 最大回撤 | N/A |
| 胜率 | 77% |
V. 完整的 Python 代码
import data_tools
from AlgorithmImports import *
import numpy as np
from typing import List, Dict, Tuple, Deque
from collections import deque
class PredictingBondReturnswithaCombinedModel(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2000, 1, 1)
self.SetCash(100000)
# Symbols - 10Y bond futures, equity etf, 10Y bond yield, cash rate data.
# Cash rate source: https://fred.stlouisfed.org/series/IR3TIB01USM156N
self.symbols:List[Tuple[str]] = [
("ASX_XT1", 'EWA', 'AU10YT', 'IR3TIB01AUM156N'), # 10 Year Commonwealth Treasury Bond Futures, Continuous Contract #1 (Australia)
("MX_CGB1", 'EWC', 'CA10YT', 'IR3TIB01CAM156N'), # Ten-Year Government of Canada Bond Futures, Continuous Contract #1 (Canada)
("EUREX_FGBL1", 'EWG', 'DE10Y', 'IR3TIB01EZM156N'), # Euro-Bund (10Y) Futures, Continuous Contract #1 (Germany)
("LIFFE_R1", 'EWU', 'GB10Y', 'LIOR3MUKM'), # Long Gilt Futures, Continuous Contract #1 (U.K.)
("SGX_JB1", 'EWJ', 'JP10Y', 'IR3TIB01JPM156N'), # SGX 10-Year Mini Japanese Government Bond Futures, Continuous Contract #1 (Japan)
("CME_TY1", 'SPY', 'US10Y', 'IR3TIB01USM156N') # 10 Yr Note Futures, Continuous Contract #1 (USA)
]
# Daily price data.
self.data:Dict[Symbol, SymbolData] = {}
self.month_period:int = 10
self.future_period:int = 13
self.period:int = self.month_period * 12 + 1
self.SetWarmUp(self.period * 21)
self.leverage:int = 5
# Daily spread data. (10y yield minus cash rate)
self.spread:Dict[Symbol, Deque[float]] = {}
self.commodity_index:str = 'DBC'
self.AddEquity(self.commodity_index, Resolution.Daily)
self.data[self.commodity_index] = deque(maxlen = self.period)
for bond_future, equity_etf, bond_yield_symbol, cash_rate_symbol in self.symbols:
# Bond future data.
data = self.AddData(data_tools.QuantpediaFutures, bond_future, Resolution.Daily)
self.data[bond_future] = deque(maxlen = self.future_period)
data.SetFeeModel(data_tools.CustomFeeModel())
data.SetLeverage(self.leverage)
# Equity data.
self.AddEquity(equity_etf, Resolution.Daily)
self.data[equity_etf] = deque(maxlen = self.period)
# 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)
# Steepness of the yield curve.
self.spread[bond_yield_symbol] = deque(maxlen = self.period)
self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
self.last_month:int = -1
self.Schedule.On(self.DateRules.MonthStart(self.commodity_index), self.TimeRules.AfterMarketOpen(self.commodity_index), self.Rebalance)
self.settings.daily_precise_end_time = False
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, bond_yield_symbol, cash_rate_symbol in self.symbols:
if bond_future in data and data[bond_future]:
price:float = data[bond_future].Value
self.data[bond_future].append(price)
if equity_etf in data and data[equity_etf]:
price:float = data[equity_etf].Value
self.data[equity_etf].append(price)
if bond_yield_symbol in data and data[bond_yield_symbol] and cash_rate_symbol in data and data[cash_rate_symbol]:
bond_yield:float = data[bond_yield_symbol].Value
cash_rate:float = data[cash_rate_symbol].Value
steepness:float = bond_yield - cash_rate
self.spread[bond_yield_symbol].append(steepness)
# Store commodity index price.
if self.commodity_index in data and data[self.commodity_index]:
price:float = data[self.commodity_index].Value
self.data[self.commodity_index].append(price)
def Rebalance(self):
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()
# Z score calc.
weight:Dict[Symbol, float] = {}
commodity_index_z_score:None|float = None
# Past commodities returns for which only sign is used.
minimum_data_count:float = ((self.period-1) / self.month_period) * 3
if len(self.data[self.commodity_index]) >= minimum_data_count:
closes:List[float] = [x for x in self.data[self.commodity_index]]
separete_yearly_returns:List[float] = [data_tools.Return(closes[x:x+13]) for x in range(0, len(closes),1)]
return_mean:float = np.mean(separete_yearly_returns)
return_std:float = np.std(separete_yearly_returns)
commodity_index_z_score:float = (separete_yearly_returns[-1] - return_mean) / return_std
else:
return
for bond_future, equity_etf, bond_yield_symbol, cash_rate_symbol in self.symbols:
# data is still coming
if self.Securities[bond_future].GetLastData() and qp_futures_last_update_date[bond_future] <= self.Time.date() \
or self.Securities[cash_rate_symbol].GetLastData() and ir_last_update_date[cash_rate_symbol] <= self.Time.date():
continue
# Append commodity index z score right away. It's the same for every bond.
z_scores:List[float] = [commodity_index_z_score]
# Last 13 months of monthly data is ready.
if len(self.data[bond_future]) == self.data[bond_future].maxlen:
bond_future_return:float = data_tools.Return(self.data[bond_future])
bond_future_z_score:int = 1 if bond_future_return > 0 else -1
z_scores.append(bond_future_z_score)
else:
continue
data_queues = [self.data[equity_etf], self.spread[bond_yield_symbol]]
for queue in data_queues:
if len(queue) >= minimum_data_count:
closes:List[float] = [x for x in queue]
separete_yearly_returns:List[float] = [data_tools.Return(closes[x:x+13]) for x in range(0, len(closes),1)]
return_mean:float = np.mean(separete_yearly_returns)
return_std:float = np.std(separete_yearly_returns)
z_score:float = (separete_yearly_returns[-1] - return_mean) / return_std
z_scores.append(z_score)
z_scores.append(commodity_index_z_score)
if len(z_scores) == 4:
final_z_score:float = np.mean(z_scores)
if final_z_score > 1: final_z_score = 1
elif final_z_score < -1: final_z_score = -1
# weight[bond_future] = -1 * final_z_score
weight[bond_future] = final_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, w in weight.items():
self.SetHoldings(symbol, w)