“该策略按公允价值偏差对SPDR行业ETF进行排名，每周投资于排名前六的低估ETF，用现金替换高估行业，并对等权重投资组合进行再平衡。”
资产类别: ETF、基金、股票 | 地区: 美国 | 周期: 每周 | 市场: 股票 | 关键词: 行业轮动，信贷
I. 策略概要
该策略的目标是九个SPDR股票行业ETF，使用美国银行/美林证券的美国高收益B指数（HYB）期权调整价差作为主要指标。每周使用26周的历史数据校准线性回归模型，以估计每个ETF的公允价值：
ETFfair=A×HYB+B
投资者计算公允价值和市场价格之间的偏差：
ETFdisconnect=(ETFfair−ETFmarket)/ETFmarket
ETF每周按偏差排名，并选择偏差最大的前六个ETF。市场价值超过公允价值的ETF被现金取代。投资组合采用等权重，并每周进行再平衡，旨在利用行业错误定价，同时通过为高估行业分配现金来管理下行风险。这种系统化方法利用公允价值和市场价格之间的偏差来获取潜在回报。
II. 策略合理性
该策略利用公司资本结构中的相对价值对行业进行排名，并确定进出点。随着信用风险上升，股权价值下降，反之亦然。通过使用可靠的信用风险代理，投资者可以评估信用市场和股票市场之间的关系，识别错误定价以及两个资产类别之间进行相对价值交易的机会。这种方法能够基于股票和信用动态的相互作用进行系统化的决策。
III. 来源论文
通过信用相对价值进行的股票行业轮动 [点击查看论文]
- 克莱因，Capital Context LLC
<摘要>
主动型投资者面临着在实现优于基准的回报与管理多种风险之间取得平衡的难题。无论投资策略在长期内表现多么出色，短期亏损都会损害策略的声誉并降低投资者的热情。投资策略的圣杯是在降低波动性、降低回撤风险和获得正偏斜回报的同时，实现持续的超额表现。没有任何方法能够始终提供所有这些要素，但开发能够关注这些因素的策略是一项有价值的实践。本文概述了一种使用高流动性ETF的仅做多行业轮动策略，该策略在回测中取得了令人钦佩的结果。该策略的核心是利用公司资本结构中的相对价值来对行业进行排名，并判断何时建议进入和退出。为了实施投资策略，我们使用标准普尔精选行业SPDR ETF，因为它们具有高流动性和相对较长的历史。当适当加权时，这九个ETF可以用来复制标准普尔500指数的表现。我们的最终目标是为这些ETF的投资组合（可能与无风险资产一起）选择权重，以在降低风险的同时提供超额回报。
IV. 回测表现
| 年化回报 | 12.4% |
| 波动率 | 17.2% |
| β值 | 0.513 |
| 夏普比率 | 0.49 |
| 索提诺比率 | 0.22 |
| 最大回撤 | -30.1% |
| 胜率 | 55% |
V. 完整的 Python 代码
import numpy as np
from AlgorithmImports import *
from scipy import stats
from typing import List, Dict
import data_tools
class SectorRotationViaCreditRelativeValue(QCAlgorithm):
def Initialize(self) -> None:
self.SetStartDate(2000, 1, 1)
self.SetCash(100000)
self.symbols: List[str] = [
"XLK", # Technology Select Sector SPDR Fund
"XLE", # Energy Select Sector SPDR Fund
"XLV", # Health Care Select Sector SPDR Fund
"XLF", # Financial Select Sector SPDR Fund
"XLI", # Industrials Select Sector SPDR Fund
"XLB", # Materials Select Sector SPDR Fund
"XLY", # Consumer Discretionary Select Sector SPDR Fund
"XLP", # Consumer Staples Select Sector SPDR Fund
"XLU" # Utilities Select Sector SPDR Fund
]
self.regression_period: int = 26 * 5 # Need 26 weeks data
self.leverage: int = 5
self.segment: int = 6
self.regression_data: Dict[str, data_tools.SymbolData] = {}
for symbol in self.symbols:
self.AddEquity(symbol, Resolution.Daily)
self.regression_data[symbol] = data_tools.SymbolData(self.regression_period)
self.rf_asset: Symbol = self.AddEquity('BIL', Resolution.Daily).Symbol
self.symbol: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
self.HYB: Symbol = self.AddData(data_tools.QuantpediaDailyData, 'BAMLH0A2HYBEY', Resolution.Daily).Symbol
self.regression_data[self.HYB.Value] = data_tools.SymbolData(self.regression_period)
self.selection_flag: bool = False
self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
self.UniverseSettings.Resolution = Resolution.Daily
self.Schedule.On(self.DateRules.WeekStart(self.symbol), self.TimeRules.AfterMarketOpen(self.symbol), self.Selection)
self.settings.daily_precise_end_time = False
def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
for security in changes.AddedSecurities:
security.SetFeeModel(data_tools.CustomFeeModel())
security.SetLeverage(self.leverage)
def OnData(self, data: Slice) -> None:
custom_data_last_update_date: Dict[Symbol, datetime.date] = data_tools.LastDateHandler.get_last_update_date()
ETF_market: Dict[str, float] = {}
# check if data is still coming
if self.Securities[self.HYB].GetLastData() and self.Time.date() > custom_data_last_update_date[self.HYB]:
self.Liquidate()
return
# Each day storing data about symbols in self.symbols and HYB index
for symbol in self.symbols:
if symbol in data and data[symbol]:
price: float = data[symbol].Value
if price != 0:
ETF_market[symbol] = price
self.regression_data[symbol].update(price)
if self.HYB in data and data[self.HYB]:
ETF_market[self.HYB.Value] = data[self.HYB].Value
self.regression_data[self.HYB.Value].update(data[self.HYB].Value)
# Rebalance weekly
if not self.selection_flag:
return
self.selection_flag = False
ETF_disconnect: Dict[str, float] = {}
# If HYB data aren't ready, we can't calculate any regression
if self.regression_data[self.HYB.Value].is_ready():
X: list[float] = [x for x in self.regression_data[self.HYB.Value].RegressionData][::-1]
for symbol in self.symbols:
if self.regression_data[symbol].is_ready():
if symbol in ETF_market and self.HYB.Value in ETF_market:
Y: float = [x for x in self.regression_data[symbol].RegressionData][::-1]
slope, intercept, r_value, p_value, std_err = stats.linregress(X, Y)
ETF_fair: float = slope * ETF_market[self.HYB.Value] + intercept
ETF_disconnect[symbol] = (ETF_fair - ETF_market[symbol]) / ETF_market[symbol]
long: List[str] = []
negative_disconnect: List[str] = []
rf_weight: float = .0
if len(ETF_disconnect) != 0:
# Sorted descending
sorted_by_disconnect: Dict[str, float] = {k: v for k, v in sorted(ETF_disconnect.items(), key=lambda item: item[1], reverse=True)}
for symbol, disc in sorted_by_disconnect.items():
if disc > 0:
long.append(symbol)
else:
negative_disconnect.append(symbol)
long = long[:self.segment]
total_count: int = len(long) + len(negative_disconnect)
long_weight: float = len(long) / total_count
rf_weight = len(negative_disconnect) / total_count
# Trade execution.
invested: List[str] = [x.Key.Value for x in self.Portfolio if x.Value.Invested]
for symbol in invested:
if symbol not in long:
self.Liquidate(symbol)
for symbol in long:
if symbol in data and data[symbol]:
self.SetHoldings(symbol, long_weight / len(long))
if rf_weight != 0:
if self.rf_asset in data and data[self.rf_asset]:
self.SetHoldings(self.rf_asset.Value, rf_weight)
def Selection(self) -> None:
self.selection_flag = True