“通过债券收益率变化交易国家股票指数，做多变化最大的五分位，做空变化最小的五分位，使用等权重投资组合，每月在发达市场和新兴市场中重新平衡。”
资产类别: ETF、期货 | 地区: 全球 | 周期: 每月 | 市场: 股票 | 关键词: 收益率
I. 策略概要
投资范围包括发达市场和新兴市场的国家股票指数。债券收益率变化计算为过去12个月10年期政府债券收益率的负差（t-13时的收益率减去t-1时的收益率）。国家股票指数根据债券收益率变化分为五分位。该策略包括做多债券收益率变化最高的五分位，做空最低的五分位。投资组合等权重，每月重新平衡，利用债券收益率变化来指导各国股票市场头寸。
II. 策略合理性
BYC效应是由投资者行为偏差和套利限制驱动的。投资者通常由于对任意基准的非理性偏差而对债券收益率变化反应不足，导致错误定价。这种反应不足由于资本流动缓慢而持续存在。债券收益率的下降（上升）通常会导致股票价格的上涨（下跌）效应，尤其是在重要新信息发布之后。
从1900年到2019年，BYC效应在1970年代之前最为强烈，当时由于管理行业不发达，资本流动缓慢，加剧了错误定价。ETF、外国投资和全球市场等现代发展已经缓和了这种效应。在非ETF跟踪的市场中，错误定价为战略资本配置创造了机会，而ETF有助于减少套利障碍，平衡BYC效应。
III. 来源论文
Bond Yield Changes and the Cross-Section of Global Equity Returns [点击查看论文]
- Adam Zaremba、Nusret Cakici、Robert Bianchi、龙怀刚
<摘要>
我们记录了一个新的横截面异常现象，它将国际政府债券和股票市场联系起来。使用1900-2019年61个国家的独特长期数据集，我们证明了过去的债券收益率变化可以预测未来股票指数在横截面上的回报。政府债券收益率下降幅度最大（或增长幅度最小）的五分位国家每月跑赢下降幅度最小（或增长幅度最大）的五分位国家0.63%。我们的发现支持这种效应的行为根源，表明投资者对收益率变化反应不足，而缓慢流动的资本阻碍了套利者消除这种异常现象。全球投资者可以利用这种债券收益率变化效应来增强国际资产配置决策。
IV. 回测表现
| 年化回报 | 7.83% |
| 波动率 | 15.31% |
| β值 | 0.056 |
| 夏普比率 | 0.49 |
| 索提诺比率 | -0.324 |
| 最大回撤 | N/A |
| 胜率 | 48% |
V. 完整的 Python 代码
from AlgorithmImports import *
class BondYieldChangesAndTheCrossSectionOfEquityIndices(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2000, 1, 1)
self.SetCash(100000)
self.futures = [
('ASX_YT1', 'AU10YT'),
('LIFFE_FCE1', 'CA10YT'),
('EUREX_FSMI1', 'CH10YT'),
('EUREX_FSTX1', 'DE10YT'),
('LIFFE_Z1', 'GB10YT'),
('SGX_NK1', 'JP10YT'),
]
self.etfs = [
('EWA', 'AU10YT'), # iShares MSCI Australia Index ETF
('EWO', 'AS10YT'), # iShares MSCI Austria Investable Mkt Index ETF
('EWK', 'BE10YT'), # iShares MSCI Belgium Investable Market Index ETF
('EWZ', 'BR10YT'), # iShares MSCI Brazil Index ETF
('EWC', 'CA10YT'), # iShares MSCI Canada Index ETF
('FXI', 'CN10YT'), # iShares China Large-Cap ETF
('EWQ', 'FR10YT'), # iShares MSCI France Index ETF
('EWG', 'DE10YT'), # iShares MSCI Germany ETF
('EWH', 'HK10YT'), # iShares MSCI Hong Kong Index ETF
('EWI', 'IT10YT'), # iShares MSCI Italy Index ETF
('EWJ', 'JP10YT'), # iShares MSCI Japan Index ETF
('EWM', 'MY10YT'), # iShares MSCI Malaysia Index ETF
('EWW', 'MX10YT'), # iShares MSCI Mexico Inv. Mt. Idx
('EWN', 'NL10YT'), # iShares MSCI Netherlands Index ETF
('EWS', 'SG10YT'), # iShares MSCI Singapore Index ETF
('EZA', 'ZA10YT'), # iShares MSCI South Africe Index ETF
('EWY', 'KR10YT'), # iShares MSCI South Korea ETF
('EWP', 'ES10YT'), # iShares MSCI Spain Index ETF
# ('EWD', 'SE10YT'), # iShares MSCI Sweden Index ETF # No data on investpy
('EWL', 'CH10YT'), # iShares MSCI Switzerland Index ETF
('EWT', 'TW10YT'), # iShares MSCI Taiwan Index ETF
('THD', 'TH10YT'), # iShares MSCI Thailand Index ETF
('EWU', 'GB10YT'), # iShares MSCI United Kingdom Index ETF
('SPY', 'US10YT') # SPDR S&P 500 ETF
]
self.data = {}
self.bond_yields = {}
self.period = 12 * 21 # 12 months of bond yields
self.futures_flag = False
# Strategy works on equity futures
if self.futures_flag:
for equity_future, bond_yield_symbol in self.futures:
# Equity future data.
data = self.AddData(QuantpediaFutures, equity_future, Resolution.Daily)
data.SetFeeModel(CustomFeeModel())
data.SetLeverage(5)
# Bond yield data.
self.AddData(QuantpediaBondYield, bond_yield_symbol, Resolution.Daily)
# Add Symboldata to bond yield with equity future symbol
self.bond_yields[bond_yield_symbol] = SymbolData(self.period, data.Symbol)
# Otherwise strategy works on ETFs
else:
for etf_symbol, bond_yield_symbol in self.etfs:
# ETF data
data = self.AddEquity(etf_symbol, Resolution.Daily)
data.SetFeeModel(CustomFeeModel())
data.SetLeverage(5)
# Bond yield data.
self.AddData(QuantpediaBondYield, bond_yield_symbol, Resolution.Daily)
# # Add Symboldata to bond yield with ETF symbol
self.bond_yields[bond_yield_symbol] = SymbolData(self.period, data.Symbol)
self.symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
self.Schedule.On(self.DateRules.MonthStart(self.symbol), self.TimeRules.AfterMarketOpen(self.symbol), self.Rebalance)
def OnData(self, data):
# Update bond yields
for symbol, symbol_data in self.bond_yields.items():
custom_data = [symbol, symbol_data.symbol] if self.futures_flag else [symbol]
if any([self.securities[x].get_last_data() and self.time.date() > LastDateHandler.get_last_update_date()[x] for x in custom_data]):
self.liquidate()
return
if symbol in data and data[symbol]:
self.bond_yields[symbol].update(data[symbol].Value)
def Rebalance(self):
bond_yield_changes = {}
for _, symbol_data in self.bond_yields.items():
symbol = symbol_data.symbol # Get corre
# Check if bond yields data are ready and corresponding equity or future is Tradable
if symbol_data.is_ready() and self.Securities[symbol].Price != 0 and self.Securities[symbol].IsTradable:
# Construct the bond yield change as minus the difference in 10-year government bond yield to maturity over the past 12 months
bond_yield_change = -symbol_data.difference()
# Store bond yield change under it's equity or future symbol
bond_yield_changes[symbol_data.symbol] = bond_yield_change
# If there isn't enough data for quintile selection return from function
if len(bond_yield_changes) < 5:
self.Liquidate()
return
quintile = int(len(bond_yield_changes) / 5)
sorted_by_change = [x[0] for x in sorted(bond_yield_changes.items(), key=lambda item: item[1])]
# Finally, take a long position in the highest bond yield change quintile and a short position in the lowest quintile.
long = sorted_by_change[-quintile:]
short = sorted_by_change[:quintile]
# Trade execution
long_length = len(long)
short_length = len(short)
invested = [x.Key.Value for x in self.Portfolio if x.Value.Invested]
for symbol in invested:
if symbol not in long + short:
self.Liquidate(symbol)
for symbol in long:
self.SetHoldings(symbol, 1 / long_length)
for symbol in short:
self.SetHoldings(symbol, -1 / short_length)
class SymbolData():
def __init__(self, period, symbol):
self.bond_yields = RollingWindow[float](period)
self.symbol = symbol
def update(self, price):
self.bond_yields.Add(price)
def is_ready(self):
return self.bond_yields.IsReady
def difference(self):
values = [x for x in self.bond_yields]
return values[0] - values[-1]
class LastDateHandler():
_last_update_date: Dict[Symbol, datetime.date] = {}
@staticmethod
def get_last_update_date() -> Dict[Symbol, datetime.date]:
return LastDateHandler._last_update_date
# Quantpedia bond yield data.
# NOTE: IMPORTANT: Data order must be ascending (datewise)
class QuantpediaBondYield(PythonData):
def GetSource(self, config, date, isLiveMode):
return SubscriptionDataSource("data.quantpedia.com/backtesting_data/bond_yield/{0}.csv".format(config.Symbol.Value), SubscriptionTransportMedium.RemoteFile, FileFormat.Csv)
def Reader(self, config, line, date, isLiveMode):
data = QuantpediaBondYield()
data.Symbol = config.Symbol
if not line[0].isdigit(): return None
split = line.split(',')
data.Time = datetime.strptime(split[0], "%Y-%m-%d") + timedelta(days=1)
data['yield'] = float(split[1])
data.Value = float(split[1])
if config.Symbol.Value not in LastDateHandler._last_update_date:
LastDateHandler._last_update_date[config.Symbol.Value] = datetime(1,1,1).date()
if data.Time.date() > LastDateHandler._last_update_date[config.Symbol.Value]:
LastDateHandler._last_update_date[config.Symbol.Value] = data.Time.date()
return data
# Quantpedia data.
# NOTE: IMPORTANT: Data order must be ascending (datewise)
class QuantpediaFutures(PythonData):
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 not in LastDateHandler._last_update_date:
LastDateHandler._last_update_date[config.Symbol] = datetime(1,1,1).date()
if data.Time.date() > LastDateHandler._last_update_date[config.Symbol]:
LastDateHandler._last_update_date[config.Symbol] = 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"))