Trade U.S. stocks by DPP, going long on the highest DPP quintile and short on the lowest after bad news days, holding positions for 60 trading days.
ASSET CLASS: stocks | REGION: United States | FREQUENCY:
Daily | MARKET: equities | KEYWORD: News
I. STRATEGY IN A NUTSHELL
Trades U.S. stocks using the Difference Between Purchased Price (DPP) to gauge investor anchoring. After bad news days, it buys high-DPP stocks (far below purchase price) and sells low-DPP stocks, holding positions for 60 days.
II. ECONOMIC RATIONALE
Investors overreact to news relative to their purchase prices. Stocks far from their reference prices show stronger reversals, allowing the strategy to profit from temporary mispricing.
III. SOURCE PAPER
Do Reference Prices Impact How Investors Respond to News? [Click to Open PDF]
Brad Cannonz, Binghamton University; Hannes Mohrschladt, University of Muenster – Finance Center
<Abstract>
We provide evidence that purchase prices influence how investors behave towards extreme returns. Using a sample of individual investor trades and extreme return dates, we show that when a stock is trading farther from an investor’s purchase price, the investor is more likely to trade in the direction of the stock’s return. Consistent with a relative overreaction, stocks trading farthest from their average purchase price experience the most extreme returns, which are then followed by greater subsequent reversals. A cross-sectional strategy motivated by these findings earns a monthly alpha of 1.02%.
IV. BACKTEST PERFORMANCE
| Annualised Return | 15.32% |
| Volatility | 9.49% |
| Beta | 0.499 |
| Sharpe Ratio | 1.61 |
| Sortino Ratio | 0.222 |
| Maximum Drawdown | N/A |
| Win Rate | 55% |
V. FULL PYTHON CODE
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