“该策略涉及按过往表现对纳斯达克、美国证券交易所和纽约证券交易所的股票进行排序，每季度买入最高十分之一，卖出最低十分之一，股票等权重并进行重新平衡。”
资产类别: 股票 | 地区: 美国 | 周期: 季度 | 市场: 股票 | 关键词: 季节性、投资者偏好
I. 策略概要
投资范围包括纳斯达克、美国证券交易所和纽约证券交易所的所有股票。每个月，投资者根据过去21-251天的回报对股票进行排序，并将其分为十分位数。在每个季度末（三月、六月、九月和十二月），投资者买入表现最佳的十分位数，卖出表现最差的十分位数。投资组合中的股票等权重，投资组合每季度重新平衡，重点关注特定时期内基于过往回报表现最佳和最差的股票。
II. 策略合理性
该策略利用动量回报的季节性模式，该模式在日历季度末之前较高，尤其是在市场下跌之后。这受到橱窗修饰效应的驱动，机构买入赢家并卖出输家，以提高季度或年度业绩报告。这种对表现良好的股票和市场的偏好，以及对表现不佳的股票和市场的不偏好，导致赢家在季度末跑赢大盘，输家跑输大盘。该策略通过关注季度末之前近期表现强劲的股票，同时避免表现不佳的股票，从而利用这种动量。
III. 来源论文
潮水退去时的掩盖？动量季节性与投资者偏好 [点击查看论文]
- 奈杰尔·J·巴拉代尔，巴拉代尔资产管理公司。
<摘要>
我们利用动量回报的季节性模式来深入了解投资者偏好。我们发现，动量因子回报在日历季度末之前要高得多，尤其是在股市下跌之后。这种模式对于大盘股、赢家和输家、美国和国际市场，以及近年来尤其明显。已确立的年终季节性与季度模式一致，而不是与税收亏损抛售一致。市场的时间序列动量遵循相同的模式，主要是在市场下跌之后。这些模式表明，投资者在季度末偏好表现良好的股票/市场，尤其是在下跌的市场中。
IV. 回测表现
| 年化回报 | 8% |
| 波动率 | N/A |
| β值 | -0.042 |
| 夏普比率 | N/A |
| 索提诺比率 | 0.191 |
| 最大回撤 | N/A |
| 胜率 | 50% |
V. 完整的 Python 代码
from AlgorithmImports import *
from pandas.core.frame import dataframe
class MomentumSeasonalityInvestorPreferences(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2010, 1, 1)
self.SetCash(100000)
market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
self.fundamental_count:int = 500
self.fundamental_sorting_key = lambda x: x.DollarVolume
self.period:int = 12 * 21
self.quantile:int = 10
self.leverage:int = 5
self.exchange_codes:List[str] = ['NYS', 'NAS', 'ASE']
self.long:List[Symbol] = []
self.short:List[Symbol] = []
# Daily price data.
self.data:Dict[Symbol, SymbolData] = {}
self.selection_flag:bool = True
self.UniverseSettings.Resolution = Resolution.Daily
self.AddUniverse(self.FundamentalSelectionFunction)
self.settings.daily_precise_end_time = False
self.settings.minimum_order_margin_portfolio_percentage = 0.
self.schedule.on(self.date_rules.month_start(market),
self.time_rules.after_market_open(market),
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]:
# 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 and x.MarketCap != 0]
if len(selected) > self.fundamental_count:
selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
performance:Dict[Symbol, float] = {}
# 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:pd.Series = history.loc[symbol].close
for time, close in closes.items():
self.data[symbol].update(close)
if self.data[symbol].is_ready():
performance[symbol] = self.data[symbol].performance()
if len(performance) >= self.quantile:
sorted_by_performance:List = sorted(performance, key = performance.get, reverse = True)
quantile:int = int(len(sorted_by_performance) / self.quantile)
self.long = sorted_by_performance[:quantile]
self.short = sorted_by_performance[-quantile:]
return self.long + self.short
def OnData(self, data: Slice) -> None:
if not self.selection_flag:
return
self.selection_flag = False
# order 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:
if self.Time.month % 3 == 0:
self.Liquidate()
self.selection_flag = True
class SymbolData():
def __init__(self, period: int):
self._price:RollingWindow = RollingWindow[float](period)
def update(self, price: float) -> None:
self._price.Add(price)
def is_ready(self) -> bool:
return self._price.IsReady
# Yearly performance, one month skipped.
def performance(self) -> float:
closes:List[float] = list(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"))