“该策略交易纳斯达克100指数成分股，根据调整后的开盘至收盘价格反应，对超过60天波动率五倍的显著财报日波动，采取2%权重的40天头寸。”
资产类别: 股票 | 地区: 美国 | 周期: 每日 | 市场: 股票 | 关键词: 财报
I. 策略概要
该策略针对纳斯达克100指数成分股，识别有财报公告电话会议的股票。投资者将公告日股票调整后的开盘至收盘波动与其60天历史波动进行比较。如果股票的相对波动超过其60天已实现日波动率的五倍，则建立头寸（多头或空头）。每个头寸的权重为投资组合的2%，持有40天，利用财报公告的显著价格反应，同时通过一致的权重和持有期管理风险。
II. 策略合理性
学术研究指出了该策略成功的两个主要原因：对冲基金和股票分析师的延迟反应。由于董事会召集，拥有大量市场波动资金流的对冲基金通常反应缓慢。同样，以滞后反应而闻名的还有股票分析师，尤其是来自大型银行的分析师。在发布有影响力的出版物之后，分析师需要时间重新评估模型、投资案例并重写笔记，从而延迟了市场影响。这些延迟造成了策略所利用的低效性，因为随着时间的推移，大量资金流和修订后的分析会影响股价，而不是在初始信息发布后立即产生影响。
III. 来源论文
财报公告后漂移，一种价格信号？[点击查看论文]
- 梅西亚斯，巴黎多芬大学
<摘要>
本文研究了基于价格信号的财报后异常波动（PEAD）的稳健性，这与侧重于基本面信号的传统文献不同。研究期间为2003-2015年，针对美国四大主要指数。结果表明，尽管一些经济主体仍然具有重大的市场影响力，但他们整合信息的速度太慢。我们发现了强有力的经验证据，表明这种偏差对动量股而非蓝筹股或非动量小盘股更为突出。即使对该策略提出质疑，结论仍然很强，与这种市场低效率相关的异常回报，正信号的回报优于负信号。我们选择纳斯达克综合指数作为我们发展的基础，因为它最接近Uncia的专业领域。对于被称为动量指数的指数，我们发现系统净敞口的强烈可预测性，后者是收益信号所暗示的多头和空头头寸的结果。
IV. 回测表现
| 年化回报 | 5.64% |
| 波动率 | 4.03% |
| β值 | -0.11 |
| 夏普比率 | 1.4 |
| 索提诺比率 | -0.217 |
| 最大回撤 | -7.79% |
| 胜率 | 52% |
V. 完整的 Python 代码
from data_tools import CustomFeeModel, TradeManager, SymbolData
from AlgorithmImports import *
import numpy as np
from collections import deque
from typing import Dict, List, Set
from pandas.tseries.offsets import BDay
class PostEarningsAnnouncementDrift(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2010, 1, 1) # earnings data start in 2010
self.SetCash(100000)
self.period:int = 61
self.leverage:int = 5
self.long:List[Symbol] = []
self.short:List[Symbol] = []
self.long_count:int = 5
self.short_count:int = 5
self.holding_period:int = 40
# monthly selected universe
self.last_selection:List[Symbol] = []
self.data:Dict[Symbol, SymbolData] = {}
self.tickers:Set(str) = set()
# EPS quarterly data
self.eps:Dict[Symbol, deque] = {}
self.earnings_data:Dict[datetime.date, List[str]] = {}
earnings_data:str = self.Download('data.quantpedia.com/backtesting_data/economic/earnings_dates_eps.json')
earnings_data_json:list[dict] = json.loads(earnings_data)
for obj in earnings_data_json:
date:datetime.date = datetime.strptime(obj['date'], '%Y-%m-%d').date()
year:int = date.year
month:int = date.month
self.earnings_data[date] = []
for stock_data in obj['stocks']:
ticker:str = stock_data['ticker']
self.earnings_data[date].append(ticker)
self.tickers.add(ticker)
# equally weighted brackets for traded symbols
self.trade_manager:TradeManager = TradeManager(self, self.long_count, self.short_count, self.holding_period)
self.market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
self.fundamental_sorting_key = lambda x: x.DollarVolume
self.fundamental_count:int = 500
self.last_month:int = -1
self.UniverseSettings.Resolution = Resolution.Daily
self.AddUniverse(self.FundamentalSelectionFunction)
self.settings.daily_precise_end_time = False
self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
def OnSecuritiesChanged(self, changes:SecurityChanges) -> None:
for security in changes.AddedSecurities:
security.SetFeeModel(CustomFeeModel())
security.SetLeverage(self.leverage)
for security in changes.RemovedSecurities:
if security.Symbol in self.data:
if security.Symbol != self.market:
del self.data[security.Symbol]
def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
if self.Time.month != self.last_month:
self.last_month = self.Time.month
# in fundamental always select whole universe (stocks, which are in QP earnings data),
# because prices of each stock in this universe are updated in OnData (due to open price)
selected:List[Fundamental] = [x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and x.Symbol.Value in self.tickers]
if len(selected) > self.fundamental_count:
selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
self.last_selection = [x.Symbol for x in selected]
# warm up prices
for symbol in self.last_selection + [self.market]:
if symbol in self.data:
continue
self.data[symbol] = SymbolData(self.period)
history:dataframe = self.History(symbol, self.period+1, Resolution.Daily)
if history.empty:
continue
closes:Series = history.close
opens:Series = history.open
for (_, open_price), (_, close) in zip(opens.items(), closes.items()):
self.data[symbol].update(close, open_price)
# market prices has to be ready
if self.market not in self.data or not self.data[self.market].is_ready():
return self.last_selection
prev_business_day:datetime.date = (self.Time - BDay(1)).date()
if prev_business_day not in self.earnings_data:
return self.last_selection
# filter stocks, which had earnings on prev business day
prev_bussiness_day_earnings:List[str] = self.earnings_data[prev_business_day]
selected_fundamental:List[Symbol] = [x for x in self.last_selection if x.Value in prev_bussiness_day_earnings]
market_intraday_returns:List[float] = [x if x != 0.0 else 1.0 for x in self.data[self.market].get_intraday_returns()]
for symbol in selected_fundamental:
# symbol:Symbol = stock.Symbol
if not self.data[symbol].is_ready():
continue
stock_intraday_returns:List[float] = self.data[symbol].get_intraday_returns()
daily_moves:List[float] = [(stock_intrady_ret / market_intraday_ret) for stock_intrady_ret, market_intraday_ret \
in zip(stock_intraday_returns, market_intraday_returns)]
std:float = np.std(daily_moves)
mean:float = np.mean(daily_moves)
if daily_moves[0] > mean + 5 * std:
self.long.append(symbol)
elif daily_moves[0] < mean - 5 * std:
self.short.append(symbol)
return self.last_selection
def OnData(self, data:Slice) -> None:
# update stock data for current universe
for symbol in self.last_selection + [self.market]:
if symbol in self.data and symbol in data and data[symbol]:
close:float = data[symbol].Close
open:float = data[symbol].Open
self.data[symbol].update(close, open)
self.trade_manager.TryLiquidate()
# open new trades
for symbol in self.long:
if symbol in data and data[symbol]:
self.trade_manager.Add(symbol, True)
for symbol in self.short:
if symbol in data and data[symbol]:
self.trade_manager.Add(symbol, False)
self.long.clear()
self.short.clear()