股票 · 美国 · ETF、基金、股票 · 每周 · 美国 Free

行业轮动通过信用相对价值

学术论文

作者通过信用相对价值进行的股票行业轮动 [点击查看论文]

论文链接http://www.naaim.org/wp-content/uploads/2013/10/00N_Equity_Sector_Rotation-via__Credit-Relative_Value_David-Klein.pdf

策略概要

该策略的目标是九个SPDR股票行业ETF，使用美国银行/美林证券的美国高收益B指数（HYB）期权调整价差作为主要指标。每周使用26周的历史数据校准线性回归模型，以估计每个ETF的公允价值：

ETFfair=A×HYB+B

投资者计算公允价值和市场价格之间的偏差：

ETFdisconnect=(ETFfair−ETFmarket)/ETFmarket

ETF每周按偏差排名，并选择偏差最大的前六个ETF。市场价值超过公允价值的ETF被现金取代。投资组合采用等权重，并每周进行再平衡，旨在利用行业错误定价，同时通过为高估行业分配现金来管理下行风险。这种系统化方法利用公允价值和市场价格之间的偏差来获取潜在回报。

II. 策略合理性

该策略利用公司资本结构中的相对价值对行业进行排名，并确定进出点。随着信用风险上升，股权价值下降，反之亦然。通过使用可靠的信用风险代理，投资者可以评估信用市场和股票市场之间的关系，识别错误定价以及两个资产类别之间进行相对价值交易的机会。这种方法能够基于股票和信用动态的相互作用进行系统化的决策。

回测表现

波动率17.2%

夏普比率0.49

索提诺比率0.22

最大回撤-30.1%

胜率55%

完整 Python 代码

1import numpy as np
2from AlgorithmImports import *
3from scipy import stats
4from typing import List, Dict
5import data_tools
6class SectorRotationViaCreditRelativeValue(QCAlgorithm):
7
8def Initialize(self) -> None:
9self.SetStartDate(2000, 1, 1)
10self.SetCash(100000)
11
12self.symbols: List[str] = [
13    "XLK",  # Technology Select Sector SPDR Fund
14    "XLE",  # Energy Select Sector SPDR Fund
15    "XLV",  # Health Care Select Sector SPDR Fund
16    "XLF",  # Financial Select Sector SPDR Fund
17    "XLI",  # Industrials Select Sector SPDR Fund
18    "XLB",  # Materials Select Sector SPDR Fund
19    "XLY",  # Consumer Discretionary Select Sector SPDR Fund
20    "XLP",  # Consumer Staples Select Sector SPDR Fund
21    "XLU"   # Utilities Select Sector SPDR Fund
22]
23
24self.regression_period: int = 26 * 5 # Need 26 weeks data
25self.leverage: int = 5
26self.segment: int = 6
27self.regression_data: Dict[str, data_tools.SymbolData] = {}
28
29for symbol in self.symbols:
30    self.AddEquity(symbol, Resolution.Daily)
31    self.regression_data[symbol] = data_tools.SymbolData(self.regression_period)
32
33self.rf_asset: Symbol = self.AddEquity('BIL', Resolution.Daily).Symbol
34self.symbol: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
35
36self.HYB: Symbol = self.AddData(data_tools.QuantpediaDailyData, 'BAMLH0A2HYBEY', Resolution.Daily).Symbol
37self.regression_data[self.HYB.Value] = data_tools.SymbolData(self.regression_period)
38
39self.selection_flag: bool = False
40self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
41self.UniverseSettings.Resolution = Resolution.Daily
42self.Schedule.On(self.DateRules.WeekStart(self.symbol), self.TimeRules.AfterMarketOpen(self.symbol), self.Selection)
43self.settings.daily_precise_end_time = False
44def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
45for security in changes.AddedSecurities:
46    security.SetFeeModel(data_tools.CustomFeeModel())
47    security.SetLeverage(self.leverage)
48
49def OnData(self, data: Slice) -> None:
50custom_data_last_update_date: Dict[Symbol, datetime.date] = data_tools.LastDateHandler.get_last_update_date()
51ETF_market: Dict[str, float] = {}
52
53# check if data is still coming
54if self.Securities[self.HYB].GetLastData() and self.Time.date() > custom_data_last_update_date[self.HYB]:
55    self.Liquidate()
56    return
57# Each day storing data about symbols in self.symbols and HYB index
58for symbol in self.symbols: 
59    if symbol in data and data[symbol]:
60        price: float = data[symbol].Value
61        if price != 0:
62            ETF_market[symbol] = price
63            self.regression_data[symbol].update(price)
64
65if self.HYB in data and data[self.HYB]:
66    ETF_market[self.HYB.Value] = data[self.HYB].Value
67    self.regression_data[self.HYB.Value].update(data[self.HYB].Value)
68# Rebalance weekly 
69if not self.selection_flag:
70    return
71self.selection_flag = False
72
73ETF_disconnect: Dict[str, float] = {}
74# If HYB data aren't ready, we can't calculate any regression
75if self.regression_data[self.HYB.Value].is_ready():
76    X: list[float] = [x for x in self.regression_data[self.HYB.Value].RegressionData][::-1]
77    
78    for symbol in self.symbols:
79        if self.regression_data[symbol].is_ready():
80            if symbol in ETF_market and self.HYB.Value in ETF_market:
81                Y: float = [x for x in self.regression_data[symbol].RegressionData][::-1]
82                slope, intercept, r_value, p_value, std_err = stats.linregress(X, Y)
83                ETF_fair: float = slope * ETF_market[self.HYB.Value] + intercept
84                ETF_disconnect[symbol] = (ETF_fair - ETF_market[symbol]) / ETF_market[symbol]
85
86long: List[str] = []
87negative_disconnect: List[str] = []
88rf_weight: float = .0
89if len(ETF_disconnect) != 0:
90    # Sorted descending
91    sorted_by_disconnect: Dict[str, float] = {k: v for k, v in sorted(ETF_disconnect.items(), key=lambda item: item[1], reverse=True)}
92    for symbol, disc in sorted_by_disconnect.items():
93        if disc > 0:
94            long.append(symbol)
95        else:
96            negative_disconnect.append(symbol)
97    
98    long = long[:self.segment]
99        
100    total_count: int = len(long) + len(negative_disconnect)
101    long_weight: float = len(long) / total_count
102    rf_weight = len(negative_disconnect) / total_count
103        
104# Trade execution.
105invested: List[str] = [x.Key.Value for x in self.Portfolio if x.Value.Invested]
106for symbol in invested:
107    if symbol not in long:
108        self.Liquidate(symbol)
109    
110for symbol in long:
111    if symbol in data and data[symbol]:
112        self.SetHoldings(symbol, long_weight / len(long))
113    
114if rf_weight != 0:
115    if self.rf_asset in data and data[self.rf_asset]:
116        self.SetHoldings(self.rf_asset.Value, rf_weight)
117
118def Selection(self) -> None:
119self.selection_flag = True