“该策略投资于动量最高的美国股票十分位，并对权益曲线应用24个月移动平均过滤器，仅在动量超过移动平均值时进行交易。”
资产类别：股票 | 地区：美国 | 频率：每月 | 市场：股票市场 | 关键词：时机、过滤器、动量、因子
I. 策略概述
该策略利用AMEX、NYSE和NASDAQ的股票数据，根据动量（月度收益从第t-12个月到第t-2个月，不包括上个月）对股票每月进行排序，将其分为十分位。投资组合按价值加权，重点跟踪动量最高的十分位（最高动量组合）。对该动量策略的权益曲线应用24个月移动平均过滤器，仅当前一个月的权益曲线点高于其24个月移动平均值时，投资者才对动量最高的十分位建立多头头寸。回测使用Kenneth French数据库中的十分位组合，但此方法同样适用于由上述股票构建的投资组合。
II. 策略合理性
该策略结合动量投资和市场时机的过滤器逻辑。动量策略利用过去表现较强的股票在短期内通常继续跑赢的市场特性，但容易在市场逆转时遭受回撤。通过应用24个月移动平均过滤器，仅在市场条件支持动量策略时进行交易，从而减少潜在亏损并优化风险调整后的回报。这种方法的本质是通过市场时机判断增强动量策略的鲁棒性。
III. 论文来源
Market Timing with Moving Averages [点击浏览原文]
- 作者：Glabadanidis
- 机构：阿德莱德大学商学院
<摘要>
我提供了证据表明，在均值-方差-偏度框架下，基于移动平均（MA）的交易策略在三阶随机占优意义上优于买入并持有基础资产。研究使用按市值、账面市值比、现金流价格比、收益价格比、股息价格比、短期反转、中期动量、长期反转和行业分类的价值加权十分位组合的月度收益。异常收益对Carhart（1997）四因子的敏感性较低，在扣除交易成本后，经济和统计上显著的年化阿尔法收益在10%到15%之间。
该策略的表现在不同的移动平均滞后期和分时期内都具有稳健性，同时投资者情绪、流动性风险、经济周期、牛熊市场和违约利差都无法完全解释其收益来源。此外，MA策略在随机生成的收益和自举（bootstrapped）收益中同样有效。
我还提供了MA策略在七个国际股票市场中的盈利能力证据。这种策略的收益同样适用于CRSP数据库中超过18,000只个股。MA策略的显著市场时机能力似乎是其异常收益的主要驱动因素。MA策略的收益类似于基础投资组合相对于不完美平值保护性看跌期权策略的收益。此外，将多个MA策略组合成一个按价值或等权重配置的MA策略投资组合表现更优，为证券选择和市场时机提供了一个统一的框架。
IV. 回测表现
| 年化收益率 | 21.58% |
| 波动率 | 18.69% |
| Beta | 0.453 |
| 夏普比率 | 0.94 |
| 索提诺比率 | 0.315 |
| 最大回撤 | N/A |
| 胜率 | 59% |
V. 完整python代码
from AlgorithmImports import *
import numpy as np
from numpy import isnan
class MarketTimingFilterAppliedMomentumOtherFactorStrategies(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2000, 1, 1)
self.SetCash(100000)
market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
self.SMA_period:int = 24
self.period:int = 13
self.quantile:int = 10
self.leverage:int = 5
self.fundamental_count:int = 500
self.fundamental_sorting_key = lambda x: x.DollarVolume
# Equity holdings value.
self.mimic_equity_value = self.Portfolio.TotalPortfolioValue
self.holdings_value:Dict[Symbol, List[float]] = {}
self.equity_sma = SimpleMovingAverage(self.SMA_period)
# Monthly close data.
self.data:Dict[Symbol, SymbolData] = {}
self.weight:Dict[Symbol, float] = {}
self.plot = Chart('Strategy EQ')
self.plot.AddSeries(Series('EQ', SeriesType.Line, 0))
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.SetSlippageModel(CustomSlippageModel())
security.SetFeeModel(CustomFeeModel())
security.SetLeverage(self.leverage)
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[Funamental] = [
x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' \
and not isnan(x.EarningReports.BasicAverageShares.ThreeMonths) and x.EarningReports.BasicAverageShares.ThreeMonths > 0 \
and not isnan(x.EarningReports.BasicEPS.TwelveMonths) and x.EarningReports.BasicEPS.TwelveMonths > 0 \
and not isnan(x.ValuationRatios.PERatio) and x.ValuationRatios.PERatio > 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_market_cap:Dict[Symbol, List[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*30, Resolution.Daily)
if history.empty:
self.Log(f"Not enough data for {symbol} yet.")
continue
closes: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])
if not self.data[symbol].is_ready():
continue
# Market cap calc.
market_cap:float = float(stock.EarningReports.BasicAverageShares.ThreeMonths * (stock.EarningReports.BasicEPS.TwelveMonths * stock.ValuationRatios.PERatio))
performance_market_cap[symbol] = [self.data[symbol].performance(), market_cap]
if len(performance_market_cap) <= self.quantile:
return Universe.Unchanged
# Return sorting.
sorted_by_ret:List[Tuple[Symbol, List[float]]] = sorted(performance_market_cap.items(), key = lambda x: x[1][0], reverse = True)
quantile:int = int(len(sorted_by_ret) / self.quantile)
long:List[Tuple[Symbol, List[float]]] = [x for x in sorted_by_ret[:quantile]]
# Market cap weighting.
total_market_cap:float = sum([x[1][1] for x in long])
for symbol, perf_market_cap in long:
self.weight[symbol] = perf_market_cap[1] / total_market_cap
return list(self.weight.keys())
def OnData(self, data: Slice) -> None:
if not self.selection_flag:
return
self.selection_flag = False
# Trade execution
if len(self.weight) == 0:
self.Liquidate()
return
stocks_invested:List[Symbol] = [x.Key for x in self.Portfolio if x.Value.Invested]
for symbol in stocks_invested:
if symbol not in self.weight:
self.Liquidate(symbol)
# Calculate symbol equity. - mimic trading.
for symbol, holdings in self.holdings_value.items():
curr_price:float = self.Securities[symbol].Price
holdings_price:float = holdings[0]
holdings_q:float = holdings[1]
fee:float = holdings_price * abs(holdings_q) * 0.00005
slippage:float = curr_price * float(0.0001 * np.log10(2*float(abs(holdings_q))))
last_holdings_value:float = holdings_price * holdings_q - fee - slippage
new_holdings_value:float = (curr_price * holdings_q)
trade_pl:float = (new_holdings_value - last_holdings_value)
self.mimic_equity_value += trade_pl
self.equity_sma.Update(self.Time, self.mimic_equity_value)
self.Plot("Strategy EQ", "EQ", self.mimic_equity_value)
# self.Log('Real portfolio value: {0}; Alternative portfolio value: {1}'.format(self.Portfolio.TotalPortfolioValue, self.mimic_equity_value))
self.holdings_value.clear()
for symbol, w in self.weight.items():
if symbol in data and data[symbol]:
# Store symbol equity holdings. - mimic trading.
curr_price:float = data[symbol].Value
if curr_price != 0:
q:float = (self.mimic_equity_value * w) / curr_price
self.holdings_value[symbol] = [curr_price, q]
if self.equity_sma.IsReady:
if self.mimic_equity_value > self.equity_sma.Current.Value:
self.SetHoldings(symbol, w)
else:
continue
self.weight.clear()
def Selection(self) -> None:
self.selection_flag = True
class SymbolData():
def __init__(self, period: int) -> None:
self.Closes:RollingWindow = RollingWindow[float](period)
def update(self, close: float) -> None:
self.Closes.Add(close)
def is_ready(self) -> bool:
return self.Closes.IsReady
def performance(self) -> float:
closes = [x for x in self.Closes][1:] # skip last month
return (closes[0] - closes[-1]) / closes[-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"))
# Custom slippage model.
class CustomSlippageModel:
def GetSlippageApproximation(self, asset, order):
# custom slippage math
slippage = asset.Price * float(0.0001 * np.log10(2*float(order.AbsoluteQuantity)))
return slippage
