“该策略通过CAPM残差计算特质动量，对大盘股进行排序，对排名前五分位的股票做多，对排名后五分位的股票做空。投资组合采用等权重配置，并每月进行再平衡。”
资产类别：股票 | 地区：美国 | 频率：每月 | 市场：股票市场 | 关键词：特质、动量
I. 策略概述
该策略聚焦于在纽约证券交易所（NYSE）、美国证券交易所（AMEX）和纳斯达克（NASDAQ）上市且市值高于NYSE股票第20百分位的大盘股。通过三年的数据，使用CAPM回归估算每只股票的特质回报（残差）。特质动量通过累计残差回报（11个月期间，t-12至t-2月）计算得出。根据特质动量对股票进行五分位排序。策略对最高五分位（动量最强）的股票做多，对最低五分位（动量最弱）的股票做空。投资组合采用等权重配置，并每月进行再平衡，从而利用动量异常实现潜在的超额回报。
II. 策略合理性
学术研究表明，动量策略在市场下跌后和反弹期间往往表现较差。这是因为动量策略中被做空的“输家”组合在市场下跌期间受益，但在市场反弹时遭受显著损失。这些损失与“输家”组合的期权性特征有关。通过专注于特质表现强劲的股票，该策略避开了那些主要因高市场贝塔驱动而表现优异的公司，从而减轻市场反弹的负面影响，并增强基于动量的投资的稳健性。
III. 论文来源
Eureka! A Momentum Strategy that Also Works in Japan [点击浏览原文]
- 作者：Chaves
- 机构：The Capital Group Companies
<摘要>
本文探讨了一种动量的替代定义，该定义通过市场回归的特质回报计算得出。通过剔除因市场贝塔敞口导致的回报部分，这种新的动量定义降低了动量策略的波动性，并产生了显著的四因子阿尔法。这些结果不仅适用于美国数据，还在21个国家的样本中得到了验证。最有趣的是，该研究发现这一结论同样适用于日本市场，而此前的研究未能在日本市场中发现传统动量策略的显著有效性。
IV. 回测表现
| 年化收益率 | 12.35% |
| 波动率 | 13.25% |
| Beta | -0.12 |
| 夏普比率 | 0.63 |
| 索提诺比率 | -0.033 |
| 最大回撤 | N/A |
| 胜率 | 53% |
V. 完整python代码
from scipy import stats
from AlgorithmImports import *
from typing import List, Deque, Tuple
from collections import deque
class IdiosyncraticMomentumStocks(QCAlgorithm):
def Initialize(self) -> None:
self.SetStartDate(2000, 1, 1)
self.SetCash(100000)
# Daily price data.
self.data:Dict[Symbol, RollingWindow] = {}
self.period:int = 21
self.quantile:int = 5
self.leverage:int = 5
self.exchange_codes:List[str] = ['NYS', 'NAS', 'ASE']
self.symbol:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
self.data[self.symbol] = RollingWindow[float](self.period)
self.regression_period:int = 36
self.regression_data:Dict[Symbol, Tuple] = {}
# Monthly residuals for stocks.
self.residuals_period:int = 12
self.residual:Dict[Symbol, RollingWindow] = {}
self.long:List[Symbol] = []
self.short:List[Symbol] = []
self.fundamental_count:int = 1000
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.minimum_order_margin_portfolio_percentage = 0.
self.Schedule.On(self.DateRules.MonthEnd(self.symbol), self.TimeRules.AfterMarketOpen(self.symbol), self.Selection)
def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
for security in changes.RemovedSecurities:
symbol:Symbol = security.Symbol
if symbol in self.regression_data:
del self.regression_data[symbol]
if symbol in self.residual:
del self.residual[symbol]
for security in changes.AddedSecurities:
symbol:Symbol = security.Symbol
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 daily price.
if symbol in self.data:
self.data[symbol].Add(stock.AdjustedPrice)
if not self.selection_flag:
return Universe.Unchanged
# selected = [x.Symbol for x in fundamental if x.HasFundamentalData and x.Market == 'usa']
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]]
idiosyncratic_momentum:Dict[Symbol, float] = {}
# Warmup price rolling windows.
for stock in selected:
symbol:Symbol = stock.Symbol
if symbol not in self.data:
self.data[symbol] = RollingWindow[float](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].Add(close)
# Market data is not ready.
if not self.data[self.symbol].IsReady:
continue
market_excess_return:float = self.data[self.symbol][0] / self.data[self.symbol][self.period-1] - 1
if not self.data[symbol].IsReady:
continue
stock_excess_return:float = self.data[symbol][0] / self.data[symbol][self.period-1] - 1
# store regression data
if symbol not in self.regression_data:
self.regression_data[symbol] = deque(maxlen = self.regression_period)
self.regression_data[symbol].append((market_excess_return, stock_excess_return))
# Regression.
if len(self.regression_data[symbol]) == self.regression_data[symbol].maxlen:
# Y = α + (β ∗ X)
# intercept = alpha
# slope = beta
market_excess_returns:List[float] = [x[0] for x in self.regression_data[symbol]]
stock_excess_returns:List[float] = [x[1] for x in self.regression_data[symbol]]
slope, intercept, r_value, p_value, std_err = stats.linregress(market_excess_returns, stock_excess_returns)
# Calculate every residual for recent months.
# residuals = []
# for idx, x in enumerate(market_excess_returns):
# yfit = intercept + (slope * x)
# residuals.append(yfit - stock_excess_returns[idx])
# idiosyncratic_momentum[symbol] = sum(residuals[:-2])
# Calculate only latest residual.
actual_value:float = stock_excess_returns[-1]
estimate_value:float = intercept + (slope * market_excess_returns[-1])
residual:float = actual_value - estimate_value
# store residual data
if symbol not in self.residual:
self.residual[symbol] = RollingWindow[float](self.residuals_period)
if self.residual[symbol].IsReady:
# idiosyncratic_momentum[symbol] = self.residual[symbol][1] / self.residual[symbol][self.residuals_period-1] - 1
idiosyncratic_momentum[symbol] = sum([x for x in self.residual[symbol]][1:])
self.residual[symbol].Add(residual)
if len(idiosyncratic_momentum) >= self.quantile:
sorted_by_idiosyncratic_momentum:List[Tuple[Symbol, float]] = sorted(idiosyncratic_momentum.items(), key = lambda x: x[1], reverse = True)
quintile:int = int(len(sorted_by_idiosyncratic_momentum) / 5)
self.long:List[Symbol] = [x[0] for x in sorted_by_idiosyncratic_momentum[:quintile]]
self.short:List[Symbol] = [x[0] for x in sorted_by_idiosyncratic_momentum[-quintile:]]
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
# Custom fee model
class CustomFeeModel(FeeModel):
def GetOrderFee(self, parameters):
fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
return OrderFee(CashAmount(fee, "USD"))
