“该策略每月投资于最高十分位数的动量股票，采用价值加权投资组合，10%止损触发器，并进行再平衡，排除低价、小盘和某些受限制的股票。”
资产类别: 股票 | 地区: 美国 | 周期: 每日 | 市场: 股票 | 关键词: 动量效应
I. 策略概要
该策略的目标是纽约证券交易所、美国证券交易所和纳斯达克的公司，排除封闭式基金（CEFs）、房地产投资信托基金（REITs）、美国存托凭证（ADRs）、外国股票、价格低于5美元的股票以及最小规模的十分位数。每月，股票按其6个月的累计回报（跳过一个月）进行排名，并选择最高的十分位数进行投资。投资组合采用价值加权，每只股票设置10%的止损。如果触发止损，则卖出股票，资金以现金形式持有至当月剩余时间。投资组合（包括止损阈值）每月进行再平衡，利用动量的同时降低下行风险。
II. 策略合理性
学术界认为，动量盈利能力源于投资者对过去信息的过度反应，这种反应会在短时间内得到纠正。从业者通常使用预定义的止损技术来有效限制风险，并且这种方法可以通过减轻潜在损失，同时利用该策略固有的盈利能力，来增强动量的风险/回报特征。
III. 来源论文
驯服动量崩溃：一个简单的止损策略 [点击查看论文]
- 韩宇峰，周国富，朱颖姿。北卡罗来纳大学（UNC）夏洛特分校-金融。圣路易斯华盛顿大学-约翰·M·奥林商学院。清华大学-经济管理学院。
<摘要>
在本文中，我们提出了一种止损策略，以限制著名的动量策略的下行风险。通过使用1926年1月至2013年12月的数据，我们发现，在10%的止损水平下，等权重和价值加权动量策略的最大月度损失分别从-49.79%下降到-11.36%，以及从-64.97%下降到-23.28%，同时夏普比率翻了一番以上。我们还提供了一个止损交易者和非止损交易者的通用均衡模型，并表明市场价格与没有止损交易者的情况下的价格相差一个障碍期权。
IV. 回测表现
| 年化回报 | 18.72% |
| 波动率 | 20.99% |
| β值 | 0.381 |
| 夏普比率 | 0.89 |
| 索提诺比率 | 0.16 |
| 最大回撤 | N/A |
| 胜率 | 31% |
V. 完整的 Python 代码
import numpy as np
from AlgorithmImports import *
class MomentumEffectStocksCombinedStopLosses(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2000, 1, 1)
self.SetCash(100000)
self.period:int = 7
self.leverage:int = 10
self.quantile:int = 10
self.exchange_codes:List[str] = ['NYS', 'NAS', 'ASE']
market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
self.fundamental_count:int = 500
self.fundamental_sorting_key = lambda x: x.DollarVolume
# Symbol data.
self.data:Dict[Symbol, SymbolData] = {}
self.weight:Dict[Symbol, float] = {}
# Opened stop-orders.
self.opened_orders:List[OrderTicket] = []
self.last_month:int = -1
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.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:
symbol:Symbol = security.Symbol
security.SetFeeModel(CustomFeeModel())
security.SetLeverage(self.leverage)
if symbol not in self.data:
self.data[symbol] = SymbolData(self.period)
def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
if not self.selection_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)
selected:List[Fundamental] = [x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and x.MarketCap != 0 and \
x.SecurityReference.ExchangeId in self.exchange_codes and x.CompanyReference.IsREIT == 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[Fundamental, float] = {}
# Warmup price rolling windows.
for stock in fundamental:
symbol:Symbol = stock.Symbol
if symbol not in self.data:
self.data[symbol] = SymbolData(self.period)
history = self.History(symbol, self.period*30, Resolution.Daily)
if history.empty:
self.Log(f"Not enough data for {symbol} yet.")
continue
closes = history.loc[symbol].close
closes_len = 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 = time_close[0].date().month
next_date_month = closes.keys()[index + 1].month
# Found last day of month.
if date_month != next_date_month:
self.data[symbol].update(time_close[1])
if self.data[symbol].is_ready():
performance[stock] = self.data[symbol].performance()
# Performance sorting.
if len(performance) >= self.quantile:
sorted_by_return:List = sorted(performance.items(), key = lambda x: x[1], reverse = True)
quantile:int = int(len(sorted_by_return) / self.quantile)
long:List[Fundamental] = [x[0] for x in sorted_by_return[:quantile]]
# Market cap weighting.
mc_sum:float = sum([x.MarketCap for x in long])
for stock in long:
self.weight[stock.Symbol] = 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
# Trade execution.
for symbol, w in self.weight.items():
if symbol in data.Keys and data[symbol]:
curr_price:float = data[symbol].Value
if curr_price != 0:
# Unit size calc.
unit_size:float = self.CalculateOrderQuantity(symbol, w)
# Buy order.
if unit_size != 0:
self.MarketOrder(symbol, unit_size)
# SL setting.
sl_price:float = curr_price * 0.95 # 5% SL.
ticket:OrderTicket = self.StopMarketOrder(symbol, -unit_size, sl_price, 'SL')
self.opened_orders.append(ticket)
self.weight.clear()
def selection(self) -> None:
self.selection_flag = True
# Liquidate and cancel pending orders.
self.Liquidate()
for ticket_index in range(len(self.opened_orders)-1, 0, -1):
response = self.opened_orders[ticket_index].Cancel("Canceled Trade")
self.opened_orders.remove(self.opened_orders[ticket_index])
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
# 12 month momentum, one month skipped.
def performance(self) -> float:
return self._price[1] / 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"))