Quant Buffet放轻松,别过度思虑

股息风险溢价策略

登录后收藏

学术论文

作者提取股息风险溢价 [点击查看论文]

策略概要

该策略旨在通过模拟假设的1年期股息期货来保持对隐含股息的恒定敞口。这涉及每日在近月和远月股息期货之间滚动头寸,逐步将近月期货的权重降低至零,同时将远月期货的权重增加至100%。为了捕捉股息风险溢价,投资者做多“1年期恒定到期日”股息期货,并通过做空SX5E指数来对冲股票风险敞口。每日对冲比率通过过去两周股息期货回报与SX5E回报的线性回归确定,确保有效的风险管理。

II. 策略合理性

股息风险溢价的存在归因于几个因素。投资者高估了单只股票的股息削减风险,尽管年股息削减超过60%的情况很少见(自1991年以来为4%)。他们还低估了股息指数的多元化收益。流动性风险也是一个因素,因为股息期货的流动性较差。此外,系统性错误定价是由银行交易台持有大量SX5E股息多头头寸引起的。这些因素,包括对低概率、高影响力事件的风险厌恶和流动性挑战,在股息期货市场中创造了持续的溢价。

回测表现

波动率11.6%
夏普比率1.11
最大回撤-28.8%
胜率55%

完整 Python 代码

from AlgorithmImports import *
from scipy import stats
import numpy as np
#endregion
class DividendRiskPremiumStrategy(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2010, 1, 1)
self.SetCash(100000)

self.dividend_futures = ['FEXD1', 'FEXD2', 'FEXD3']

self.fexd1 = None
for i, div_symbol in enumerate(self.dividend_futures):
    data = self.AddData(QuantpediaFutures, div_symbol, Resolution.Daily)
    data.SetLeverage(10)
    data.SetFeeModel(CustomFeeModel())
    
    # store nearby contract
    if i == 0:
        self.fexd1 = data.Symbol

# STOXX 50 future
data = self.AddData(QuantpediaFutures, "EUREX_FSTX1", Resolution.Daily)
data.SetLeverage(10)
data.SetFeeModel(CustomFeeModel())
self.SX5E = data.Symbol

# daily price
self.data = {}
self.period = 10
self.max_missing_days = 5
for symbol in [self.fexd1, self.SX5E]:
    self.data[symbol] = RollingWindow[float](self.period)

self.selection_flag = False
self.recent_month:int = -1

def OnData(self, data):
# store daily prices
if self.fexd1 in data and data[self.fexd1] and self.SX5E in data and data[self.SX5E]:
    self.data[self.fexd1].Add(data[self.fexd1].Value)
    self.data[self.SX5E].Add(data[self.SX5E].Value)
else:
    if (self.Securities[self.fexd1].GetLastData() and (self.Time.date() - self.Securities[self.fexd1].GetLastData().Time.date()).days > self.max_missing_days) or \
        (self.Securities[self.SX5E].GetLastData() and (self.Time.date() - self.Securities[self.SX5E].GetLastData().Time.date()).days > self.max_missing_days):
        self.data[self.fexd1].Reset()
        self.data[self.SX5E].Reset()
        self.Liquidate()
if self.recent_month == self.Time.month:
    return
self.recent_month = self.Time.month
if self.Time.month in [7, 1]:
    self.Liquidate()
    self.weights = [1, 0]   # divident future #2 and #3 weights
    self.selection_flag = True

if self.Portfolio.Invested or self.selection_flag:
    if self.data[self.fexd1].IsReady and self.data[self.SX5E].IsReady:
        dividend_future_prices = np.array([x for x in self.data[self.fexd1]])
        market_prices = np.array([x for x in self.data[self.SX5E]])
        
        dividend_future_returns = dividend_future_prices[:-1] / dividend_future_prices[1:] - 1
        market_returns = market_prices[:-1] / market_prices[1:] - 1
        if not all(x==0 for x in dividend_future_returns):
            slope, intercept, r_value, p_value, std_err = stats.linregress(market_returns, dividend_future_returns)
            
            # dividend futures exposure
            self.SetHoldings(self.dividend_futures[1], self.weights[0])
            self.SetHoldings(self.dividend_futures[2], self.weights[1])
            
            # adjust weights every invested day
            if self.weights[0] >= 0.008:
                self.weights[0] -= 0.008
            
            if self.weights[0] <= 0.992:
                self.weights[1] += 0.008
            
            # hedge
            self.SetHoldings(self.SX5E, -abs(slope))
            
            self.selection_flag = False
        else:
            self.Liquidate()
# 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])
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"))