“该策略涉及根据t-12到t-2的累计回报对股票进行排序,买入绝对赢家(前10%),卖出绝对输家(后10%),每月对价值加权投资组合进行重新平衡。”

I. 策略概要

投资范围包括来自纳斯达克、美国证券交易所和纽约证券交易所的股票,不包括价格低于1美元的股票。在每个月初,股票根据其从t-12到t-2的累计回报被分为十个价值加权投资组合。累计回报断点是使用自1927年以来的历史非重叠11个月回报确定的。累计回报最高和最低的股票分别被归类为绝对赢家和绝对输家。该策略涉及买入绝对赢家和卖出绝对输家,投资组合每月重新平衡。

II. 策略合理性

投资者通常关注股票的绝对表现,由于对过去信息的过度反应,导致了类似动量的模式。绝对强度动量策略避免了相对强度动量中出现的“回声”效应,正如Novy-Marx(2012)所展示的那样。

III. 来源论文

Absolute strength: Exploring momentum in stock returns [点击查看论文]

<摘要>

我们记录了一种新的股票回报模式,我们称之为绝对强度动量。在最近一段时间内价值显著增长的股票(绝对强度赢家)继续上涨,而价值显著下降的股票(绝对强度输家)在不久的将来继续下跌。绝对强度赢家和输家投资组合的断点是通过跨时间和跨股票的已实现累计回报的历史分布递归确定的。历史分布产生了稳定的断点,对于赢家(输家)投资组合,这些断点始终为正(负)。因此,赢家是那些经历了显著上升趋势的股票,而输家是那些经历了显著下降趋势的股票,没有动量的股票的累计回报与零没有显著差异。绝对强度动量产生了巨大且显著的风险调整后回报,优于Jegadeesh和Titman(1993)的相对强度动量策略和其他突出的动量策略,并且其盈利能力在样本期、国际市场、资产类别和持有期内保持一致。

IV. 回测表现

年化回报22.28%
波动率29.2%
β值-0.199
夏普比率0.76
索提诺比率-0.026
最大回撤N/A
胜率52%

V. 完整的 Python 代码

from AlgorithmImports import *
from scipy import stats
from pandas.core.frame import dataframe
class AbsoluteMomentumEffectStocks(QCAlgorithm):
    def Initialize(self):
        self.SetStartDate(2000, 1, 1)
        self.SetCash(100000)
        market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
        
        self.period:int = 13
        self.quantile:int = 5
        self.leverage:int = 5
        self.exchange_codes:List[str] = ['NYS', 'NAS', 'ASE']
        self.required_yearly_return_period:int = 10 # Minimum of years to calculate distribution from.
        
        self.data:Dict[Symbol, SymbolData] = {} # Monthly price data.
        self.weight:Dict[Symbol, float] = {}
        
        self.fundamental_count:int = 500
        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.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]:
        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)
    
                # Add yearly performance.
                if self.data[symbol].is_ready():
                    self.data[symbol].add_yearly_return(self.data[symbol].performance())
        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]]
        long:List[Fundamental] = []
        short:List[Fundamental] = []
        # Warmup price rolling windows.
        for stock in selected:
            symbol:Symbol = stock.Symbol
            if symbol not in self.data:
                self.data[symbol] = SymbolData(self.period, self.required_yearly_return_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])
            
            if self.data[symbol].yearly_returns_ready():
                # Calculate distribution.
                yearly_returns:List[float] = [x for x in self.data[symbol]._yearly_returns]
                prev_yearly_returns:List[float] = yearly_returns[:-1]
                yearly_ret:float = yearly_returns[-1]
                
                percentile:float = stats.percentileofscore(prev_yearly_returns, yearly_ret) / 100
                if percentile >= 0.9:
                    long.append(stock)
                elif percentile <= 0.1:
                    short.append(stock)
        
        # 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
        # Trade execution.
        portfolio:List[PortfolioTarget] = [
            PortfolioTarget(symbol, w) for symbol, w in self.weight.items() if symbol in data and data[symbol]
        ]
        self.SetHoldings(portfolio, True)
        self.weight.clear()
    def selection(self) -> None:
        self.selection_flag = True
        
class SymbolData():
    def __init__(self, period: int, required_yearly_return_period: int):
        self._prices:RollingWindow = RollingWindow[float](period)
        self._yearly_returns:List[float] = []
        self._required_yearly_return_period:int = required_yearly_return_period
    
    def update(self, price: float) -> None:
        self._prices.Add(price)
    
    def add_yearly_return(self, value: float) -> None:
        self._yearly_returns.append(value)
        
    def is_ready(self) -> bool:
        return self._prices.IsReady
    
    def yearly_returns_ready(self) -> bool:
        return len(self._yearly_returns) >= self._required_yearly_return_period
        
    # Yearly performance, one month skipped.
    def performance(self) -> float:
        return (self._prices[1] / self._prices[self._prices.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"))

发表评论

了解 Quant Buffet 的更多信息

立即订阅以继续阅读并访问完整档案。

继续阅读