“该策略涉及大盘股，按1个月回报和PTH比率对其进行排序。它创建了25个投资组合，做多低PTH过去输家，做空低PTH过去赢家，每月重新平衡。”
资产类别: 股票 | 地区: 美国 | 周期: 每月 | 市场: 股票 | 关键词: 52周，短期反转
I. 策略概要
投资范围包括纽约证券交易所（NYSE）、美国证券交易所（AMEX）和纳斯达克（NASDAQ）的股票，不包括在t月末价格低于5美元的股票。使用纽约证券交易所的百分位数，将股票分为微盘股、小盘股和大盘股公司。重点关注大盘股，PTH比率计算为当前价格相对于过去52周最高价格的比率。股票根据1个月回报和PTH排名被分为五分位投资组合。这两个投资组合的交集创建了25种组合。该策略涉及做多低PTH过去输家和做空低PTH过去赢家，等权重投资组合每月重新平衡。
II. 策略合理性
52周最高股票价格作为Andrei和Cujean模型中会议强度的代理，影响动量和反转效应。达到52周高点的股票吸引媒体关注，增加投资者兴趣和信息交流。该论文表明，由于投资者会议频繁，动量在高PTH股票中占主导地位，而低PTH股票则经历短期反转。该策略显示PTH值与短期反转盈利能力之间存在强烈的负相关关系。结果在不同的回报类型、公司规模和子样本期间保持稳健。分别对PTH和反转排名进行计时，可以防止赢家和输家的投资组合不平衡。
III. 来源论文
Information Percolation, the 52-Week High, and Short-Term Reversal in Stock Returns [点击查看论文]
- 朱兆波、孙立成、克里斯·T·斯蒂弗斯、张凯，深圳大学；奥德西亚商学院；奥尔德多米尼翁大学；路易斯维尔大学
<摘要>
我们发现，价格锚定在理解1个月（1M）股票回报的短期反转中，与众所周知的流动性提供渠道相结合，发挥着作用。具体而言，我们确定，对于（a）相对于其52周高点价格较低的股票（George和Hwang）和（b）资本利得悬置较低的股票（Grinblatt和Han），1M反转策略表现更好。此外，我们发现过去赢家和过去输家之间的反转行为存在显著的非对称性，这取决于股票相对于价格参考点的价格。这些反转非对称性与假设的价格锚定偏差相符。
IV. 回测表现
| 年化回报 | 19.6% |
| 波动率 | 30.61% |
| β值 | 0.121 |
| 夏普比率 | 0.64 |
| 索提诺比率 | 0.396 |
| 最大回撤 | N/A |
| 胜率 | 52% |
V. 完整的 Python 代码
from AlgorithmImports import *
class ReversalCombinedwithVolatility(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2000, 1, 1)
self.SetCash(100000)
self.SetSecurityInitializer(lambda x: x.SetMarketPrice(self.GetLastKnownPrice(x)))
self.leverage:int = 10
self.quantile:int = 5
self.period:int = 52 * 5
self.exchange_codes:List[str] = ['NYS', 'NAS', 'ASE']
market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
self.long:List[Symbol] = []
self.short:List[Symbol] = []
self.data:Dict[Symbol, SymbolData] = {}
self.fundamental_count:int = 500
self.fundamental_sorting_key = lambda x: x.DollarVolume
self.selection_flag:bool = False
self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
self.UniverseSettings.Resolution = Resolution.Daily
self.AddUniverse(self.FundamentalSelectionFunction)
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]:
# Update the rolling window every day.
for stock in fundamental:
symbol:Symbol = stock.Symbol
# Store monthly price.
if symbol in self.data:
self.data[symbol].update(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.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_ready:List[Fundamental] = []
# Warmup price rolling windows.
for stock in selected:
symbol:Symbol = stock.Symbol
if symbol not in self.data:
self.data[symbol] = SymbolData(self.period)
history:dataframe = self.History(symbol, self.period, Resolution.Daily)
if history.empty:
self.Log(f"Not enough data for {symbol} yet.")
continue
closes:Series = history.loc[symbol].close
for time, close in closes.items():
self.data[symbol].update(close)
if self.data[symbol].is_ready():
selected_ready.append(stock)
pth_performance:Dict[Symbol, Tuple[float]] = {x.Symbol : (self.data[x.Symbol].pth(), self.data[x.Symbol].performance()) for x in selected_ready}
sorted_by_pth:List[Tuple[Symbol, float]] = sorted(pth_performance.items(), key = lambda x: x[1][0], reverse = True)
sorted_by_pth:List[Symbol] = [x[0] for x in sorted_by_pth]
sorted_by_ret:List[Tuple[Symbol, float]] = sorted(pth_performance.items(), key = lambda x: x[1][1], reverse = True)
sorted_by_ret:List[Symbol] = [x[0] for x in sorted_by_ret]
quintile:int = int(len(sorted_by_ret) / self.quantile)
low_pth:List[Symbol] = sorted_by_pth[-quintile:]
top_ret:List[Symbol] = sorted_by_ret[:quintile]
low_ret:List[Symbol] = sorted_by_ret[-quintile:]
self.long = [x for x in low_pth if x in low_ret]
self.short = [x for x in low_pth if x in top_ret]
return self.long + self.short
def OnData(self, data: Slice) -> None:
if not self.selection_flag:
return
self.selection_flag = False
# Trade execution.
targets:List[PortfolioTarget] = []
for i, portfolio in enumerate([self.long, self.short]):
for symbol in portfolio:
if symbol in data and data[symbol]:
targets.append(PortfolioTarget(symbol, ((-1) ** i) / len(portfolio)))
self.SetHoldings(targets, True)
self.long.clear()
self.short.clear()
def Selection(self) -> None:
self.selection_flag = True
class SymbolData():
def __init__(self, period:int):
self._price:RollingWindow = RollingWindow[float](period)
def update(self, value:float) -> None:
self._price.Add(value)
def is_ready(self) -> bool:
return self._price.IsReady
def pth(self) -> float:
high_proxy = [x for x in self._price]
symbol_price = high_proxy[0]
return symbol_price / max(high_proxy[21:])
def performance(self) -> float:
closes = [x for x in self._price][:21]
return (closes[0] / closes[-1] - 1)
# Custom fee model.
class CustomFeeModel(FeeModel):
def GetOrderFee(self, parameters):
fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
return OrderFee(CashAmount(fee, "USD"))