“通过DPP交易美国股票，在坏消息日后做多最高DPP五分位，做空最低DPP五分位，持有头寸60个交易日。”
资产类别: 股票 | 地区: 美国 | 周期: 每日 | 市场: 股票 | 关键词: 价格、坏消息
I. 策略概要
投资范围包括CRSP数据库中的美国股票（纽约证券交易所、美国证券交易所、纳斯达克）。关键变量是购买价格差异（DPP），每月使用An（2016）的方法计算。DPP由两部分组成：（1）基于股票以特定价格购买但尚未交易的概率（使用换手率）的权重，以及（2）当前价格与n个周期前购买价格之间的绝对差异，除以当前价格（详细信息见第15页的公式3）。
新闻日通过极端回报来识别，定义为异常回报的绝对值超过2（使用Fama-French三因子模型）。负极端回报表示坏消息日。
每月，股票被分为DPP五分位。在坏消息日之后，做多来自最高五分位（最高DPP）的股票，做空来自最低五分位（最低DPP）的股票，持有60个交易日。
II. 策略合理性
该策略的功能源于参考价格和投资者偏见。作者衡量股票交易价格与购买价格之间的差异，以评估对新闻的过度反应或反应不足。与接近购买价格的股票相比，偏离购买价格最远的股票在好消息日（坏消息日）表现出显著更高（更低）的异常回报。这些股票在新闻发布后也经历更大的回报反转。在一个季度（61天）内，偏离购买价格最远和最近的股票之间的回报差为4.13%。该策略利用这种横截面反应，每月获得0.93%的阿尔法，证明了其在捕捉投资者行为方面的有效性。
III. 来源论文
Do Reference Prices Impact How Investors Respond to News? [点击查看论文]
- Brad Cannonz、Hannes Mohrschladt。宾厄姆顿大学。明斯特大学金融中心
<摘要>
我们提供的证据表明，购买价格会影响投资者对极端回报的行为。通过个体投资者交易和极端回报日期的样本，我们发现当股票交易价格偏离投资者购买价格越远时，投资者越有可能按照股票回报的方向进行交易。与相对过度反应一致，交易价格偏离其平均购买价格最远的股票经历最极端的收益，随后是更大的后续反转。受这些发现启发而采用的横截面策略每月获得1.02%的阿尔法。
IV. 回测表现
| 年化回报 | 15.32% |
| 波动率 | 9.49% |
| β值 | 0.499 |
| 夏普比率 | 0.161 |
| 索提诺比率 | 0.222 |
| 最大回撤 | N/A |
| 胜率 | 55% |
V. 完整的 Python 代码
import numpy as np
from AlgorithmImports import *
import statsmodels.api as sm
import data_tools
from pandas.core.frame import dataframe
class ReferencePricesAndBadNews(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2000, 1, 1)
self.SetCash(100000)
self.symbol:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
self.data:Dict[Symbol, data_tools.SymbolData] = {} # storing SymbolData objects under stocks symbols
self.top_quintile:List[Symbol] = [] # storing stocks from top quintile by DPP
self.bottom_quintile:List[Symbol] = [] # storing stocks from bottom quintile by DPP
self.currently_not_traded:List[Symbol] = [] # storing symbols of currently not traded stocks
self.currently_traded:List[Symbol] = [] # storing symbols of currently traded stocks
self.managed_symbols:List[ManagedSymbol] = []
self.short_period:int = 5 # need n of daily prices and volumes
self.turnover_period:int = 250 # need n of weekly turnovers for DPP calculation
self.long_period:int = 21 * 12 # need n of daily prices for market and stocks
self.holding_period:int = 60 # stocks are held for n days
self.traded_symbols:int = 50 # max value of currently traded stocks
self.bad_news_ret_threshold:float = -0.02
self.exchange_codes:List[str] = ['NYS', 'NAS', 'ASE']
self.quantile:int = 5
self.leverage:int = 5
self.market_prices:RollingWindow = RollingWindow[float](self.long_period) # storing daily market prices for regression
self.fundamental_count:int = 500
self.fundamental_sorting_key = lambda x: x.DollarVolume
self.selection_flag:bool = False
self.UniverseSettings.Resolution = Resolution.Daily
self.AddUniverse(self.FundamentalSelectionFunction)
self.settings.daily_precise_end_time = False
self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
self.Schedule.On(self.DateRules.MonthStart(self.symbol), self.TimeRules.AfterMarketOpen(self.symbol), 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]:
# update rolling windows each day
for stock in fundamental:
symbol:Symbol = stock.Symbol
if symbol in self.data:
# update stock price and volume
self.data[symbol].update(stock.AdjustedPrice, stock.Volume)
# get stock's EarningReports.BasicAverageShares.OneMonth in fine and update turnover
if self.data[symbol].short_period_ready():
self.data[symbol].update_turnover(stock.EarningReports.BasicAverageShares.ThreeMonths)
self.data[symbol].update_week_performance()
if symbol == self.symbol:
# update market prices
self.market_prices.Add(stock.AdjustedPrice)
if not self.selection_flag:
return Universe.Unchanged
selected:List[Fundamental] = [x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and x.Symbol != self.symbol and \
x.SecurityReference.ExchangeId in self.exchange_codes and not np.isnan(x.EarningReports.BasicAverageShares.ThreeMonths) and x.EarningReports.BasicAverageShares.ThreeMonths != 0]
if len(selected) > self.fundamental_count:
selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
DPP:Dict[Symbol, float] = {}
# warmup price rolling windows.
for stock in selected:
symbol:Symbol = stock.Symbol
if symbol not in self.data:
self.data[symbol] = data_tools.SymbolData(self.short_period, self.long_period, self.turnover_period)
# creating history for self.short_period,
# because it can't perform turnover calculation without EarningReports.BasicAverageShares.OneMonth
history:dataframe = self.History(symbol, self.short_period, Resolution.Daily)
if history.empty:
self.Log(f"Not enough data for {symbol} yet")
continue
closes:pd.Series = history.loc[symbol].close
volumes:pd.Series = history.loc[symbol].volume
for (_, close), (_, volume) in zip(closes.items(), volumes.items()):
if close != 0:
self.data[symbol].update(close, volume)
# get stock's EarningReports.BasicAverageShares.OneMonth in fine and update turnover
if self.data[symbol].short_period_ready():
# update stock's turnover and calculate week performance, if stock's data are ready
self.data[symbol].update_turnover(stock.EarningReports.BasicAverageShares.ThreeMonths)
self.data[symbol].update_week_performance()
# check if stock's turnovers are ready
if self.data[symbol].turnovers_ready() and self.selection_flag:
DPP[symbol] = self.data[symbol].DPP_calculation()
# change universe on monthly selection
if self.selection_flag:
# keep monthly selection
self.selection_flag = False
if len(DPP) < self.quantile:
return Universe.Unchanged
quantile:int = int(len(DPP) / self.quantile)
sorted_by_DPP:List[Symbol] = [x[0] for x in sorted(DPP.items(), key=lambda item: item[1])]
self.currently_traded = [x.symbol for x in self.managed_symbols]
# first will be stocks symbols, which aren't currently traded
self.top_quintile = sorted_by_DPP[-quantile:]
self.bottom_quintile = sorted_by_DPP[:quantile]
# get symbols of stocks, which aren't currently traded
self.currently_not_traded = [x for x in self.top_quintile + self.bottom_quintile if x not in self.currently_traded]
return self.top_quintile + self.bottom_quintile
else:
# keep old universe
return Universe.Unchanged
def OnData(self, data: Slice) -> None:
# storing stock, which were hold for too long
remove_managed_symbols:List[ManagedSymbol] = []
# update holding period for each held stock and check if any of them is held for too long
for managed_symbol in self.managed_symbols:
# increase holding period
managed_symbol.holding_period += 1
# check if stock is held for too long
if managed_symbol.holding_period == self.holding_period:
remove_managed_symbols.append(managed_symbol)
# liquidate stocks, which were held too long and remove them from self.managed_symbols dictionary
for managed_symbol in remove_managed_symbols:
if self.Portfolio[managed_symbol.symbol].Invested:
self.MarketOrder(managed_symbol.symbol, -managed_symbol.quantity)
self.managed_symbols.remove(managed_symbol)
# market prices for regression aren't ready
if not self.market_prices.IsReady:
return
market_prices:List[float] = list(self.market_prices)
remove_from_curr_not_traded:List[Symbol] = []
# check bad news days
# firstly try to trade stocks, which aren't currently traded,
# then try to trade stocks, which are currently traded, but they had bad news days
for symbol in self.currently_not_traded + self.currently_traded:
# stock doesn't have data for regression or stock wasn't selected in monthly selection
if not self.data[symbol].long_period_ready() and not self.data[symbol].short_period_ready() and (symbol in self.top_quintile or symbol in self.bottom_quintile):
continue
stock_prices:List[float] = [x for x in self.data[symbol].long_closes]
# make regression
regression_model = self.MultipleLinearRegression(market_prices, stock_prices)
# get beta
beta:float = regression_model.params[-1]
# get daily performance
stock_recent_perf:float = self.data[symbol].performance(2)
market_recent_perf:float = market_prices[0] / market_prices[1] - 1 if market_prices[1] != 0 else 0
implied_capm_return:float = market_recent_perf*beta
# stock has bad news day and there is a space in portfolio for trading this stock
if stock_recent_perf - implied_capm_return <= self.bad_news_ret_threshold and len(self.managed_symbols) < self.traded_symbols:
# calculate traded quantity for stock
weight:float = self.Portfolio.TotalPortfolioValue / self.traded_symbols
quantity:float = np.floor(weight / self.data[symbol].last_price())
# change quantity to negative, because stock is in short quintile
if symbol in self.bottom_quintile:
quantity = quantity * -1
# trade execution
if self.Securities[symbol].Price != 0 and self.Securities[symbol].IsTradable:
self.MarketOrder(symbol, quantity)
# add stock's symbol to self.managed_symbols dictionary
self.managed_symbols.append(data_tools.ManagedSymbol(symbol, quantity))
# stock is already traded, so it needs to be removed from self.currently_not_traded list
if symbol in self.currently_not_traded:
remove_from_curr_not_traded.append(symbol)
for symbol in remove_from_curr_not_traded:
# remove from currently not traded list
self.currently_not_traded.remove(symbol)
# add to currently traded list
self.currently_traded.append(symbol)
def MultipleLinearRegression(self, x, y):
x = np.array(x).T
x = sm.add_constant(x)
result = sm.OLS(endog=y, exog=x).fit()
return result
def Selection(self) -> None:
self.selection_flag = True