“该策略涉及根据前一年的回报对美国股票进行排序。多头头寸建立在“赢家”上，空头头寸建立在“输家”上。市场崩盘后，头寸在三个月内切换为LMW。”
资产类别: 股票 | 地区: 美国 | 周期: 每月 | 市场: 股票 | 关键词: 动态动量，逆向交易
I. 策略概要
投资范围包括美国股票。在月末，股票根据前一年的回报（不包括最近一个月）被分为十分位数。最高十分位数（“赢家”）做多，最低十分位数（“输家”）做空，形成WML（赢家减去输家）策略。头寸持有一个月。如果市场经历崩盘（低于平均回报两个标准差以上），该策略在三个月内切换为LMW（输家减去赢家）。之后，如果没有进一步的市场暴跌，投资组合将恢复为WML。投资组合每月重新平衡，并按价值加权。
II. 策略合理性
动量崩盘部分是可预测的，通常发生在动量回报高、利率低或崩盘后市场反弹之后。这些崩盘往往发生在重大市场损失后的一到三个月内。崩盘与动量策略的投资组合形成过程有关，其中最近的市场表现不佳加剧了崩盘。通过在重大市场损失后纳入逆向策略，该策略旨在减少动量崩盘并将其转化为收益。这种调整通过避免潜在的崩盘并更好地适应市场条件，从而增强了动量策略。
III. 来源论文
Dynamic Momentum and Contrarian Trading [点击查看论文]
- 多布林斯卡娅，俄罗斯高等经济学院金融学院
<摘要>
高动量回报无法用风险因素解释，但它们呈负偏态，并且偶尔会遭受严重的崩盘。我探讨了动量崩盘的时机，并表明动量策略往往在局部股市暴跌后1-3个月内崩盘。接下来，我提出了一个简单的动态交易策略，该策略在平静时期与标准动量策略一致，但在市场崩盘后一个月切换到相反的逆向策略，并保持逆向头寸三个月，之后恢复到动量头寸。动态动量策略将所有重大动量崩盘转化为收益，并产生平均回报，约为标准动量回报的1.5倍。动态动量回报呈正偏态，不受风险因素影响，具有高夏普比率和阿尔法，并在全球不同的时期和地域市场中持续存在。
IV. 回测表现
| 年化回报 | 21.74% |
| 波动率 | 26.74% |
| β值 | 0.06 |
| 夏普比率 | 0.81 |
| 索提诺比率 | 0.191 |
| 最大回撤 | -39.39% |
| 胜率 | 53% |
V. 完整的 Python 代码
import numpy as np
from AlgorithmImports import *
from pandas.core.frame import dataframe
class DynamicMomentumContrarianTrading(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2000, 1, 1)
self.SetCash(100000)
self.weight:Dict[Symbol, float] = {}
# Monthly price data.
self.data:Dict[Symbol, SymbolData] = {}
self.period:int = 13
self.quantile:int = 10
self.leverage:int = 5
self.market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
# Market daily data.
daily_period:int = 21
self.data[self.market] = SymbolData(daily_period)
self.market_return_data:List[float] = []
self.min_monthly_perf_period:int = 12
self.contrarian_flag:bool = False
self.contrarian_months:int = 0
self.fundamental_count:int = 500
self.fundamental_sorting_key = lambda x: x.DollarVolume
self.selection_flag:int = False
self.UniverseSettings.Resolution = Resolution.Daily
self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
self.AddUniverse(self.FundamentalSelectionFunction)
self.Schedule.On(self.DateRules.MonthEnd(self.market), self.TimeRules.BeforeMarketClose(self.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]:
if not self.selection_flag or self.contrarian_flag:
return Universe.Unchanged
# Update the rolling window every month.
for stock in fundamental:
symbol:Symbol = stock.Symbol
# Store monthly price.
if symbol in self.data:
self.data[symbol].update(stock.AdjustedPrice)
# Market return calc.
if self.data[self.market].is_ready():
self.market_return_data.append(self.data[self.market].performance())
selected:List[Fundamental] = [x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' 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]]
# 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 * 30, Resolution.Daily)
if history.empty:
self.Log(f"Not enough data for {symbol} yet.")
continue
closes:pd.Series = history.loc[symbol].close
closes_len:int = len(closes.keys())
# Find monthly closes.
for index, time_close in enumerate(closes.items()):
# index out of bounds check.
if index + 1 < closes_len:
date_month:int = time_close[0].date().month
next_date_month:int = closes.keys()[index + 1].month
# Found last day of month.
if date_month != next_date_month:
self.data[symbol].update(time_close[1])
performance:Dict[Fundamental, float] = {x : self.data[x.Symbol].performance(1) for x in selected if x.Symbol in self.data and self.data[x.Symbol].is_ready()}
# At least one year of monthly market return is ready.
if len(self.market_return_data) >= self.min_monthly_perf_period and len(performance) >= self.quantile:
mean_ret:float = np.mean(self.market_return_data)
std_ret:float = np.std(self.market_return_data)
recent_market_ret:float = self.market_return_data[-1]
# There was a crash last month.
if recent_market_ret < mean_ret - 2*std_ret:
self.contrarian_flag = True
sorted_by_performance:List[Fundamental] = sorted(performance, key = performance.get, reverse = True)
quantile:int = int(len(sorted_by_performance) / self.quantile)
long:List[Fundamental] = []
short:List[Fundamental] = []
if self.contrarian_flag:
short = sorted_by_performance[:quantile]
long = sorted_by_performance[-quantile:]
else:
long = sorted_by_performance[:quantile]
short = sorted_by_performance[-quantile:]
# Market cap weighting.
for i, portfolio in enumerate([long, short]):
mc_sum:float = sum(map(lambda x: x.MarketCap, portfolio))
for stock in portfolio:
self.weight[stock.Symbol] = ((-1) ** i) * stock.MarketCap / mc_sum
return list(self.weight.keys())
def OnData(self, data: Slice) -> None:
if not self.selection_flag:
return
self.selection_flag = False
portfolio:List[PortfolioTarget] = [PortfolioTarget(symbol, w) for symbol, w in self.weight.items() if symbol in data and data[symbol]]
self.SetHoldings(portfolio, True)
# Trade execution.
if self.contrarian_flag:
self.contrarian_months += 1
if self.contrarian_months == 3:
self.contrarian_flag = False
self.contrarian_months = 0
self.weight.clear()
else:
self.weight.clear()
def Selection(self) -> None:
self.selection_flag = True
class SymbolData():
def __init__(self, period: int) -> None:
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
# Performance, one month skipped.
def performance(self, values_to_skip = 0) -> float:
return self._price[values_to_skip] / self._price[self._price.Count - 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"))