Dividend Risk Premium Strategy
Log in to collectAcademic paper
Strategy in a nutshell
This strategy maintains constant exposure to implied dividends by rolling positions between front- and back-month dividend futures, going long on 1-year constant maturity dividend futures while hedging equity exposure via short SX5E positions. The daily hedge ratio is calculated using regression over the prior two weeks to manage risk effectively.
Economic rationale
A dividend risk premium exists due to overestimation of single-stock dividend cuts, underestimation of diversification benefits, liquidity constraints, and systematic mispricing from large positions held by banks. These factors create a persistent opportunity in dividend futures markets.
Backtest performance
Annualised return12.9%
Volatility11.6%
Beta0.227
Sharpe ratio1.11
Maximum drawdown-28.8%
Win rate55%
Full Python code
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"))