“该策略针对纽约证券交易所（NYSE）、美国证券交易所（AMEX）和纳斯达克（NASDAQ）的股票，筛选出价格和成交量发生显著波动的股票，并选择伴随分析师修正的标的。投资者将这些股票等权配置，持有一个月，并每月重新平衡投资组合。”
资产类别：股票 | 地域：美国 | 频率：每月 | 市场：股票 | 关键词：价格、分析师修正
I. 策略概述
该策略聚焦于纽约证券交易所（NYSE）、美国证券交易所（AMEX）和纳斯达克（NASDAQ）的股票。在每月月底，策略识别出价格在单日内上涨至少+5%，且成交量超过45天平均值1.1倍的股票。随后分析这些价格波动后的分析师目标价修正情况。若分析师对目标价的修正以上调为主，则将股票归类为正向；若以下调为主，则归类为负向。仅选择价格波动后5天内发布修正的股票。投资者将筛选出的股票持有一个月，投资组合等权分配，并在每月重新平衡。此策略旨在利用分析师对重大价格与成交量变化的反应获利。
II. 策略合理性
价格的大幅波动可能由新基本面信息的出现或噪声引发。当分析师在价格大幅变动后发布修正时，这增加了价格波动是由新信息驱动的可能性。学术论文未说明为何这一效应未被市场套利机制消除，但“套利限制理论”可以作为可能的解释：这种效应并非无风险套利，且套利者的资本有限，无法使市场完全有效。
III. 论文来源
Large Price Changes and Subsequent Returns [点击浏览原文]
- Govindaraj, Livnat, Savor, Zhao
<摘要>
我们研究了大幅股票价格变动是否与短期反转或动量效应相关，具体条件是分析师在这些价格变化后立即发布目标价或收益预测修正。研究结果表明，当分析师修正在价格冲击后立即发布时，股票价格呈现动量效应，这表明初始价格变动基于新信息。而当价格变化未伴随即时分析师修正时，我们记录到短期反转，这表明初始价格冲击可能由流动性或噪声交易者引发。基于价格变动方向及分析师修正方向的交易策略在日历时间中可获得显著的异常月度收益。
IV. 回测表现
| 年化收益率 | 11.22% |
| 波动率 | 11.23% |
| Beta | 1.016 |
| 夏普比率 | N/A |
| 索提诺比率 | 0.468 |
| 最大回撤 | N/A |
| 胜率 | 57% |
V. 完整python代码
from AlgorithmImports import *
from pandas.tseries.offsets import BDay
from typing import List, Dict
import data_tools
# endregion
class LargePriceChangesCombinedWithAnalystRevisions(QCAlgorithm):
def Initialize(self) -> None:
self.SetStartDate(2010, 1, 1) # estimize dataset starts in 2011
self.SetCash(100_000)
self.years_period: int = 3
self.low_high_percentage: int = 30
self.min_values: int = 15
self.period: int = 45 # need n values for mean volumes calculation
self.volume_percentage: float = 1.1
self.return_increase: float = 0.05
self.days_for_revision: int = 5
self.leverage: int = 5
self.min_share_price: int = 5
self.exchange_codes: List[str] = ['NYS', 'NAS', 'ASE']
self.data: Dict[Symbol, SymbolData] = {}
self.weights: Dict[Symbol, float] = {}
self.already_subscribed: Dict[Symbol] = []
self.estimates: Dict[str, Dict[datetime.date, List[str]]] = {}
self.analysts_data: Dict[str, Dict[datetime.date, float]] = {}
market: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
self.fundamental_sorting_key = lambda x: x.DollarVolume
self.fundamental_count: int = 500
self.rebalance_flag: bool = False
self.selection_flag: bool = False
self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
self.UniverseSettings.Resolution = Resolution.Daily
self.AddUniverse(self.FundamentalSelectionFunction)
self.Schedule.On(self.DateRules.MonthStart(market), self.TimeRules.BeforeMarketClose(market), self.Selection)
def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
for security in changes.AddedSecurities:
security.SetFeeModel(data_tools.CustomFeeModel())
security.SetLeverage(self.leverage)
def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
curr_date: datetime.date = self.Time.date()
# daily update of prices and volumes
for equity in fundamental:
symbol: Symbol = equity.Symbol
if symbol in self.data:
self.data[symbol].update(curr_date, equity.AdjustedPrice, equity.Volume)
# monthly selection
if not self.selection_flag:
return Universe.Unchanged
self.selection_flag = False
self.rebalance_flag = True
selected: List[Fundamental] = [
x for x in fundamental if x.HasFundamentalData and x.MarketCap != 0 and x.Market == 'usa' and x.Price > self.min_share_price 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]]
selected_stocks: set = set()
for stock in selected:
symbol: Symbol = stock.Symbol
ticker: str = symbol.Value
# check if stock is already subscribed
if symbol not in self.data:
self.data[symbol] = data_tools.SymbolData(self.period)
self.AddData(EstimizeEstimate, symbol)
if not self.data[symbol].is_ready(self.min_values):
continue
if ticker not in self.estimates:
continue
large_swing_dates: List[datetime.date] = self.data[symbol].get_large_swing_dates(self.volume_percentage, self.return_increase)
# iterate through each large swing date and check if any analyst increased estimated EPS within self.days_for_revision days
for date in large_swing_dates:
for i in range(1, self.days_for_revision + 1, 1):
future_date: datetime.date = (date + BDay(i)).date()
if future_date not in self.estimates[ticker]:
continue
analyst_ids: List[str] = self.estimates[ticker][future_date]
for analyst_id in analyst_ids:
estimate_dates: List[datetime.date] = list(self.analysts_data[analyst_id].keys())
estimate_dates.reverse()
est_after_swing: float = self.analysts_data[analyst_id][future_date]
latest_date_before_swing: datetime.date = next((est_date for est_date in estimate_dates if est_date < date), None)
# check if analyst increased his/her EPS estimate value
if latest_date_before_swing != None and (est_after_swing > self.analysts_data[analyst_id][latest_date_before_swing]):
selected_stocks.add(symbol)
break
# stock were already selected, no need to check any more dates
if selected_stocks in selected_stocks:
break
# reset monthly data
for symbol, symbol_obj in self.data.items():
symbol_obj.reset_monthly_data()
long_length: int = len(selected_stocks)
for symbol in selected_stocks:
self.weights[symbol] = 1 / long_length
return list(selected_stocks)
def OnData(self, data: Slice) -> None:
estimate = data.Get(EstimizeEstimate)
for symbol, value in estimate.items():
ticker: str = symbol.Value
if ticker not in self.estimates:
self.estimates[ticker] = {}
created_at: datetime.date = value.CreatedAt.date()
if created_at not in self.estimates[ticker]:
self.estimates[ticker][created_at] = []
analyst_id: str = value.AnalystId
self.estimates[ticker][created_at].append(analyst_id)
if analyst_id not in self.analysts_data:
self.analysts_data[analyst_id] = {}
self.analysts_data[analyst_id][created_at] = value.Eps
# rebalance when selection was made
if not self.rebalance_flag:
return
self.rebalance_flag = False
# reset monthly data
for _, symbol_obj in self.data.items():
symbol_obj.reset_monthly_data()
# trade execution
portfolio: List[PortfolioTarget] = [PortfolioTarget(symbol, w) for symbol, w in self.weights.items() if symbol in data and data[symbol]]
self.SetHoldings(portfolio, True)
self.weights.clear()
def Selection(self) -> None:
self.selection_flag = True
