
“该策略通过盈利加速交易纽约证券交易所、美国证券交易所和纳斯达克股票,做多最高十分位数,做空最低十分位数,并在每月持有期内的盈利公告后每日重新平衡。”
资产类别: 股票 | 地区: 美国 | 周期: 每日 | 市场: 股票 | 关键词: 加速效应
I. 策略概要
该策略针对CRSP中的纽约证券交易所、美国证券交易所和纳斯达克股票,不包括金融(SIC 6000–6999)和公用事业(SIC 4900–4949)公司。盈利加速计算为连续两个四季度期间每股收益(EPS)增长率的变化,并按股票价格缩放。股票根据盈利加速分为十分位数,做多最高十分位数,做空最低十分位数。持有期从季度t的盈利公告后两天开始,到该月的第30天结束。投资组合按价值加权,并每日重新平衡,以适应不同的盈利公告日期。
II. 策略合理性
盈利加速策略产生了异常回报,因为投资者未能充分理解当前盈利加速对未来盈利增长的影响。测试证实,正的策略回报与未来盈利增长之间存在很强的关联。与其他在2003年后重要性下降的异常现象不同,该策略继续表现稳健,在176个季度内,超额回报保持显著且稳定。该策略还满足了解释横截面回报的建议t统计阈值(>3.0),并展示了与价值加权投资组合一致的结果。这些发现突显了该策略在不同市场条件和时间段内的经济意义、持久性和实际可行性。
III. 来源论文
Earnings Acceleration and Stock Returns [点击查看论文]
- 何碩源(Shuoyuan He)与甘斯·纳拉亚纳穆尔蒂(Gans Narayanamoorthy),旧金山州立大学,杜兰大学会计与税务系
<摘要>
我们记录了盈利加速(定义为季度环比盈利增长变化)对未来超额回报具有显著的解释力。这些超额回报对先前记录的各种异常现象以及一系列风险控制措施都具有稳健性。超额回报的幅度(在一个月的窗口期内为1.8%)与账面市值比、盈利公告后漂移和总盈利能力异常相当。未来回报的可预测性似乎是由于市场错过了盈利加速对未来两三个季度盈利增长的可预测影响。最后,通过关注特定的盈利加速模式,基本盈利加速交易策略的超额回报可以进一步提高近45%。


IV. 回测表现
| 年化回报 | 23.87% |
| 波动率 | 13.81% |
| β值 | -0.096 |
| 夏普比率 | 1.44 |
| 索提诺比率 | -0.405 |
| 最大回撤 | N/A |
| 胜率 | 50% |
V. 完整的 Python 代码
from AlgorithmImports import *
from typing import Dict, List
from numpy import isnan
class EarningsAccelerationEffectinStocks(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2004, 1, 1)
self.SetCash(100000)
self.exchange_codes:List[str] = ['NYS', 'NAS', 'ASE']
self.leverage:int = 5
self.quantile:int = 10
self.min_share_price:int = 5
self.quarters_count:int = 6 # Number of quarters to calculate the earning acceleration indicator
self.fundamental_count:int = 1000
self.fundamental_sorting_key = lambda x: x.DollarVolume
self.weight:Dict[Symbol, float] = {}
self.eps_by_symbol:Dict[Symbol, RollingWindow] = {} # Contains RollingWindow objects for every stock
self.last_fundamental:List[Symbol] = []
self.symbol:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
self.selection_flag:bool = False
self.settings.daily_precise_end_time = False
self.settings.minimum_order_margin_portfolio_percentage = 0.
self.UniverseSettings.Resolution = Resolution.Daily
self.AddUniverse(self.FundamentalSelectionFunction)
self.Schedule.On(self.DateRules.MonthEnd(self.symbol), self.TimeRules.AfterMarketOpen(self.symbol), self.Selection)
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]:
'''Drop securities which have no fundamental data or have too low prices.
Select those with the highest dollar volume'''
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.MarketCap != 0 \
and not isnan(x.EarningReports.BasicEPS.ThreeMonths) and x.EarningReports.BasicEPS.ThreeMonths > 0 \
and x.SecurityReference.ExchangeId in self.exchange_codes
]
if len(selected) > self.fundamental_count:
selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
ea_by_stock:Dict[Symbol, float] = {}
for stock in selected:
symbol:Symbol = stock.Symbol
# If stock wasn't in last_fundamental we need to reintialize it to make sure data are consecutive
if symbol not in self.eps_by_symbol or symbol not in self.last_fundamental:
self.eps_by_symbol[symbol] = RollingWindow[float](self.quarters_count)
# update rolling window for every stock
self.eps_by_symbol[symbol].Add(stock.EarningReports.BasicEPS.ThreeMonths)
if self.eps_by_symbol[symbol].IsReady:
rw:List[float] = [x for x in self.eps_by_symbol[symbol]]
eps_fraction1:float = (rw[0] - rw[4]) / rw[1]
eps_fraction2:float = (rw[1] - rw[5]) / rw[2]
ea_by_stock[stock] = eps_fraction1 - eps_fraction2 # That's the earnings acceleration we want
sorted_by_ea:List[Fundamental] = [x[0] for x in sorted(ea_by_stock.items(), key = lambda item: item[1], reverse = True)]
# Create long and short quantile
quantile:int = int(len(sorted_by_ea) / self.quantile)
long:List[Fundamental] = sorted_by_ea[:quantile]
short:List[Fundamental] = sorted_by_ea[-quantile:]
# Calculate weight for each stock in portfolio
for i, portfolio in enumerate([long, short]):
mc_sum:float = sum(list(map(lambda stock: stock.MarketCap, portfolio)))
for stock in portfolio:
self.weight[stock.Symbol] = ((-1) ** i) * stock.MarketCap / mc_sum
# Change last fundamental to make sure data are consecutive
self.last_fundamental = [x.Symbol for x in fundamental]
return list(self.weight.keys())
def OnData(self, data: Slice) -> None:
if not self.selection_flag:
return
self.selection_flag = False
# Trade execution.
portfolio:List[PortfolioTarget] = [PortfolioTarget(symbol, w) for symbol, w in self.weight.items() if symbol in data and data[symbol]]
self.SetHoldings(portfolio, True)
self.weight.clear()
def Selection(self) -> None:
self.selection_flag = True
# Custom fee model
class CustomFeeModel(FeeModel):
def GetOrderFee(self, parameters):
fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
return OrderFee(CashAmount(fee, "USD"))