信用违约掉期（CDS）期限结构预测股票收益

“研究表明，大宗商品投资与风险偏好密切相关，从而促进了套利交易的表现。货币波动率和TED利差能够有效预测套利交易的表现，反映市场压力和全球流动性变化的影响。”

资产类别：股票 | 地区：美国 | 频率：每月 | 市场：股票市场 | 关键词：期限结构

I. 策略概述

研究表明，企业信用违约掉期（CDS）利差的期限结构（五年期CDS利差减去一年期利差）可以用于预测股票收益。具体而言，CDS期限结构斜率较低（较平）的股票在未来六个月内平均跑赢CDS期限结构斜率较高（较陡）的股票，每月超额收益超过1%。此策略通过识别CDS期限结构斜率较低的股票构建多头头寸，同时对CDS斜率较高的股票构建空头头寸，投资组合按等权重配置，并按月进行再平衡。

II. 策略合理性

学术研究指出，陡峭的CDS期限结构可能表明投资者预期企业信用质量恶化以及未来CDS利差扩大。这一预期通常因信息扩散的滞后性而逐步反映在股价中。CDS期限结构斜率较低的股票通常与更高的市场信心和更低的信用风险相关，这使得这些股票在未来表现优于期限结构斜率较高的股票。通过利用这一关联，该策略捕捉市场中由信用预期变化驱动的定价异常。

III. 论文来源

Term Structure of Credit Default Swap Spreads and Cross-Section of Stock Returns [点击浏览原文]

• 作者：Han, Zhou
• 机构：多伦多大学罗特曼管理学院（University of Toronto, Rotman School of Management），旧金山州立大学

<摘要>

企业信用违约掉期（CDS）利差期限结构的斜率（五年期减去一年期的利差）能够显著预测未来的股票收益。CDS斜率较低的股票在未来六个月内平均跑赢CDS斜率较高的股票，每月超额收益超过1%。这一结果无法通过标准风险因子、股票特征、违约风险测量或CDS利差变化解释。研究进一步发现，CDS期限结构斜率与市场定价之间存在一种独立于传统资产定价模型的强相关性，这表明CDS期限结构在预测股票收益方面具有重要作用。

IV. 回测表现

年化收益率	27.57%
波动率	23.84%
Beta	0.166
夏普比率	0.99
索提诺比率	-0.096
最大回撤	N/A
胜率	50%

V. 完整python代码

from AlgorithmImports import *
import data_tools
from typing import List, Dict
#endregion
class CombinedStockandCDSMomentum(QCAlgorithm):
    def Initialize(self) -> None:
        self.SetStartDate(2003, 1, 1)
        self.SetCash(100_000)
        self.UniverseSettings.Leverage = 5
        self.UniverseSettings.Resolution = Resolution.Daily
        self.AddUniverse(self.FundamentalSelectionFunction)
        self.Settings.MinimumOrderMarginPortfolioPercentage = 0.0
        self.settings.daily_precise_end_time = False
        
        self.quantile: int = 10
        self.selection_flag: bool = False
        self.tickers: List[str] = []
        self.long_symbols: List[Symbol] = []
        self.short_symbols: List[Symbol] = []
        self.cds_1y: Symbol = self.AddData(data_tools.EquityCDS1Y, 'CDS1Y', Resolution.Daily).Symbol
        self.cds_5y: Symbol = self.AddData(data_tools.EquityCDS5Y, 'CDS5Y', Resolution.Daily).Symbol
        market: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
        self.Schedule.On(self.DateRules.MonthStart(market), 
                        self.TimeRules.AfterMarketOpen(market), 
                        self.Selection)
    def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
        for security in changes.AddedSecurities:
            security.SetFeeModel(data_tools.CustomFeeModel())
    def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
        slope: Dict[Symbol, int] = {}
        
        # calculate slope of the CDS term structure
        if self.Securities.ContainsKey(self.cds_1y) and self.Securities.ContainsKey(self.cds_5y):
            cds_1y_data = self.Securities[self.cds_1y].GetLastData()
            cds_5y_data = self.Securities[self.cds_5y].GetLastData()
            
            if cds_1y_data and cds_5y_data:
                # data has not been initialized yet
                if len(self.tickers) == 0:
                    self.tickers = list([x.upper() for x in cds_1y_data.GetStorageDictionary().Keys])
                
                if self.selection_flag:
                    for f in fundamental:
                        symbol: Symbol = f.Symbol
                        ticker: str = symbol.Value
        
                        # calculate slope
                        if ticker in self.tickers:
                            cds_1y: int = cds_1y_data[ticker]
                            cds_5y: int = cds_5y_data[ticker]
                            slope[symbol] = cds_5y - cds_1y
        
        if not self.selection_flag:
            return Universe.Unchanged
        
        last_update_date_1y: datetime.date = data_tools.EquityCDS1Y.get_last_update_date()
        last_update_date_5y: datetime.date = data_tools.EquityCDS5Y.get_last_update_date()
        if (self.Securities[self.cds_1y].GetLastData() and last_update_date_1y < self.Time.date() or
            self.Securities[self.cds_5y].GetLastData() and last_update_date_5y < self.Time.date()):
            return []
        if len(slope) >= self.quantile:
            sorted_by_slope: List[Symbol] = sorted(slope, key=slope.get, reverse=True)
            quantile: int = int(len(sorted_by_slope) / self.quantile)
            self.long_symbols = sorted_by_slope[-quantile:]
            self.short_symbols = sorted_by_slope[:quantile]
        return self.long_symbols + self.short_symbols
    
    def OnData(self, slice: Slice) -> None:
        if not self.selection_flag: 
            return
        self.selection_flag = False
        
        # trade execution
        targets: List[PortfolioTarget] = []
        for i, portfolio in enumerate([self.long_symbols, self.short_symbols]):
            for symbol in portfolio:
                if slice.ContainsKey(symbol) and slice[symbol] is not None:
                    targets.append(PortfolioTarget(symbol, ((-1) ** i) / len(portfolio)))
        self.SetHoldings(targets, True)
        self.long_symbols.clear()
        self.short_symbols.clear()
    def Selection(self) -> None:
        self.selection_flag = True