“该策略基于SUE幅度交易美国股票，做多高SUE股票，做空低SUE股票，持有12个月，等权重，每月再平衡。”
资产类别: 股票 | 地区: 美国 | 周期: 每日 | 市场: 股票 | 关键词: 财报，星期五
I. 策略概要
该策略针对市值超过1亿美元、价格超过2美元且在周五发布财报的美国股票。它基于标准化意外收益（SUE）幅度构建投资组合。使用前一年的SUE分布，计算正负SUE值的中位数。SUE高于正中位数的股票被添加到多头投资组合中，而SUE低于负中位数的股票被添加到空头投资组合中。头寸等权重，从公告月份开始持有12个月，并每月再平衡。这种方法捕捉了收益意外对股票长期表现的影响。
II. 策略合理性
学术研究表明，管理层策略性地安排财报发布时间，以获得潜在利益。在公司层面，时机选择可以延迟市场的充分反应，使价格能够更缓慢地纳入新闻。在个人层面，机会主义的时机选择使管理层能够在公开公告后但在信息完全反映在价格之前买卖公司股票，利用延迟的市场反应来获取个人利益。这种理性的时机选择突显了管理层利用市场低效率并使公告与公司或自身的有利结果保持一致的策略性方法。
III. 来源论文
隐藏财报消息的最佳时机是什么时候？[点击查看论文]
- Michaely, Rubin, Vedrashko，香港大学；欧洲公司治理研究中心 (ECGI)，西蒙弗雷泽大学（SFU）- 比迪商学院；赖赫曼大学，西蒙弗雷泽大学（SFU）- 比迪商学院
<摘要>
通过结合财报公告的星期几和一天中的时间（交易时间之前、期间和之后），我们研究了管理层是否试图策略性地安排这些公告的时间。我们记录了最糟糕的财报消息是在星期五晚上发布的，并发现了强有力的证据，表明只有星期五晚上的公告才代表了管理层理性的机会主义行为。星期五晚上的公告之后，内部人士会按照财报消息的方向进行交易，并且财报后异常波动最大。管理层还试图通过在星期五晚上发布公告来减少与投资者的互动，并隐藏除财报消息之外的更多信息。我们发现，星期五晚上的公告比其他晚上的公告发布得更晚，公司举行电话会议的倾向降低，并且重大公司重组事件相对更有可能在星期五晚上的公告之后发生。
IV. 回测表现
| 年化回报 | 20.84% |
| 波动率 | N/A |
| β值 | 0 |
| 夏普比率 | N/A |
| 索提诺比率 | -0.473 |
| 最大回撤 | N/A |
| 胜率 | 44% |
V. 完整的 Python 代码
import numpy as np
from AlgorithmImports import *
from collections import deque
from dateutil.relativedelta import relativedelta
from typing import List, Dict, Tuple
from numpy import isnan
class PostEarningsAnnouncementDriftFridayEveningAnnouncers(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2000, 1, 1)
self.SetCash(100000)
self.period:int = 13
self.leverage:int = 5
self.min_share_price:int = 2
# EPS quarterly data.
self.eps_data:Dict[Symbol, List[Tuple[float]]] = {}
market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
# Surprise data count needed to count standard deviation.
self.surprise_period:int = 4
self.earnings_surprise:Dict[Symbol, List[float]] = {}
# SUE history for previous year used for statistics.
self.sue_previous_year:List[float] = []
self.sue_actual_year:List[float] = []
# Trenching.
self.holding_period:int = 12
self.managed_queue:List[RebalanceQueueItem] = []
self.market_cap_threshold: float = 100_000_000
# Last fundamental stock price
self.last_price:Dict[Symbol, float] = {}
self.month:int = 12
self.fundamental_sorting_key = lambda x: x.DollarVolume
self.fundamental_count:int = 1000
self.selection_flag:bool = False
self.UniverseSettings.Resolution = Resolution.Daily
self.AddUniverse(self.FundamentalSelectionFunction)
self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
self.Schedule.On(self.DateRules.MonthEnd(market), self.TimeRules.AfterMarketOpen(market), self.Selection)
self.settings.daily_precise_end_time = False
def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
for security in changes.AddedSecurities:
security.SetFeeModel(CustomFeeModel())
security.SetLeverage(self.leverage)
def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
if not self.selection_flag:
return Universe.Unchanged
selected:List[Fundamental] = [
x for x in fundamental if x.HasFundamentalData and x.Price > self.min_share_price and x.Market == 'usa' and not\
isnan(x.EarningReports.BasicEPS.ThreeMonths) and (x.EarningReports.BasicEPS.ThreeMonths != 0) and x.MarketCap > self.market_cap_threshold
]
if len(selected) > self.fundamental_count:
selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
for stock in selected:
self.last_price[stock.Symbol] = stock.AdjustedPrice
# Stocks with last month's earnings.
last_month_date:datetime = self.Time - timedelta(self.Time.day)
filered_fundamental:List[Symbol] = [x for x in selected if (x.EarningReports.FileDate.Value.year == last_month_date.year and x.EarningReports.FileDate.Value.month == last_month_date.month)]
sue_data:Dict[Symbol, float] = {}
for stock in filered_fundamental:
symbol:Symbol = stock.Symbol
# Store eps data.
if symbol not in self.eps_data:
self.eps_data[symbol] = deque(maxlen = self.period)
data:Tuple[float] = (stock.EarningReports.FileDate.Value.date(), stock.EarningReports.BasicEPS.ThreeMonths)
# NOTE: Handles duplicate values. QC fundamental contains duplicated stocks in some cases.
if data not in self.eps_data[symbol]:
self.eps_data[symbol].append(data)
if len(self.eps_data[symbol]) == self.eps_data[symbol].maxlen:
recent_eps_data:Tuple[float] = self.eps_data[symbol][-1]
year_range:range = range(self.Time.year - 3, self.Time.year)
last_month_date:datetime = recent_eps_data[0] + relativedelta(months = -1)
next_month_date:datetime = recent_eps_data[0] + relativedelta(months = 1)
month_range:List[int] = [last_month_date.month, recent_eps_data[0].month, next_month_date.month]
# Earnings with todays month number 4 years back.
seasonal_eps_data:List[Tuple[float]] = [x for x in self.eps_data[symbol] if x[0].month in month_range and x[0].year in year_range]
if len(seasonal_eps_data) != 3: continue
# Make sure we have a consecutive seasonal data. Same months with one year difference.
year_diff:float = np.diff([x[0].year for x in seasonal_eps_data])
if all(x == 1 for x in year_diff):
seasonal_eps:List[float] = [x[1] for x in seasonal_eps_data]
diff_values:List[float] = np.diff(seasonal_eps)
drift:float = np.average(diff_values)
# SUE calculation.
last_earnings:Tuple[float] = seasonal_eps[-1]
expected_earnings:Tuple[float] = last_earnings + drift
actual_earnings:Tuple[float] = recent_eps_data[1]
# Store sue value with earnigns date.
earnings_surprise:Tuple[float] = actual_earnings - expected_earnings
if symbol not in self.earnings_surprise:
self.earnings_surprise[symbol] = deque()
else:
# Surprise data is ready.
if len(self.earnings_surprise[symbol]) >= self.surprise_period:
earnings_surprise_std:float = np.std(self.earnings_surprise[symbol])
sue:float = earnings_surprise / earnings_surprise_std
# Store sue in this years's history of friday's earnings.
self.sue_actual_year.append(sue)
# Only stocks with last month's earnings on friday.
if stock.EarningReports.FileDate.Value.weekday() == 4:
sue_data[symbol] = sue
self.earnings_surprise[symbol].append(earnings_surprise)
long:List[Symbol] = []
short:List[Symbol] = []
# Wait until we have history data for previous year.
if len(sue_data) != 0 and len(self.sue_previous_year) != 0:
positive_sue_values:List[float] = [x for x in self.sue_previous_year if x > 0]
positive_sue_median:float = np.median(positive_sue_values)
negative_sue_values:List[float] = [x for x in self.sue_previous_year if x <= 0]
negative_sue_median:float = np.median(negative_sue_values)
long = [x[0] for x in sue_data.items() if x[1] > 0 and x[1] >= positive_sue_median]
short = [x[0] for x in sue_data.items() if x[1] <= 0 and x[1] <= negative_sue_median]
long_symbol_q:List[Tuple[Symbol, float]] = []
short_symbol_q:List[Tuple[Symbol, float]] = []
if len(long) != 0:
long_w:float = self.Portfolio.TotalPortfolioValue / self.holding_period / len(long)
long_symbol_q = [(x, np.floor(long_w / self.last_price[x])) for x in long if self.last_price[x] != 0]
if len(short) != 0:
short_w:float = self.Portfolio.TotalPortfolioValue / self.holding_period / len(short)
short_symbol_q = [(x, -np.floor(short_w / self.last_price[x])) for x in short if self.last_price[x] != 0]
self.managed_queue.append(RebalanceQueueItem(long_symbol_q, short_symbol_q))
return long + short
def OnData(self, data: Slice) -> None:
if not self.selection_flag:
return
self.selection_flag = False
# Trade execution
remove_item:Union[None, RebalanceQueueItem] = None
# Rebalance portfolio
for item in self.managed_queue:
if item.holding_period == self.holding_period:
for symbol, quantity in item.long_symbol_q + item.short_symbol_q:
if symbol in data and data[symbol]:
if self.Securities[symbol].Price != 0 and self.Securities[symbol].IsTradable:
self.MarketOrder(symbol, -quantity)
remove_item = item
elif item.holding_period == 0:
open_long_symbol_q:List[Tuple[Symbol, float]] = []
open_short_symbol_q:List[Tuple[Symbol, float]] = []
for symbol, quantity in item.long_symbol_q + item.short_symbol_q:
if symbol in data and data[symbol]:
if self.Securities[symbol].Price != 0 and self.Securities[symbol].IsTradable:
self.MarketOrder(symbol, quantity)
open_long_symbol_q.append((symbol, quantity))
# Only opened orders will be closed
item.long_symbol_q = open_long_symbol_q
item.short_symbol_q = open_short_symbol_q
item.holding_period += 1
# We need to remove closed part of portfolio after loop. Otherwise it will miss one item in self.managed_queue.
if remove_item:
self.managed_queue.remove(remove_item)
def Selection(self) -> None:
self.selection_flag = True
# Save yearly history.
if self.month == 12:
self.sue_previous_year = list(self.sue_actual_year)
self.sue_actual_year.clear()
self.month += 1
if self.month > 12:
self.month = 1
class RebalanceQueueItem():
def __init__(self, long_symbol_q:List[Tuple[Symbol, float]], short_symbol_q:List[Tuple[Symbol, float]]):
# symbol/quantity collections
self.long_symbol_q:List[Tuple[Symbol, float]] = long_symbol_q
self.short_symbol_q:List[Tuple[Symbol, float]] = short_symbol_q
self.holding_period:int = 0
# Custom fee model.
class CustomFeeModel(FeeModel):
def GetOrderFee(self, parameters):
fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
return OrderFee(CashAmount(fee, "USD"))