动量效应与高应计项目

“该策略通过对高应计项目赢家和输家进行多空头寸交易美国股票，采用价值加权，持有六个月，间隔一个月，并每月重新平衡以优化回报。”

资产类别: 股票 | 地区: 美国 | 周期: 每月 | 市场: 股票 | 关键词: 动量、应计

I. 策略概要

该策略交易CRSP数据库中的美国股票，不包括金融公司、外国公司、封闭式基金、房地产投资信托基金和美国存托凭证。股票根据前一财年的应计项目分为三组（低、中、高）。每个应计项目组根据六个月累计回报（过去的输家到赢家）进一步分为若干分位数。在高应计项目组中，该策略对过去的赢家做多，对过去的过去的输家做空，持仓六个月，形成期和持仓期之间有一个月的间隔。投资组合采用价值加权，每月重新平衡以优化绩效。

II. 策略合理性

管理者可以利用应计项目来传递私人信息，或者操纵收益以误导关注短期业绩的投资者。被高估的公司往往会进一步夸大应计项目，以延长高估期以获取个人利益。动量盈利能力在高应计公司中显著，但在低应计和中等应计公司中则微不足道。这种效应是稳健的，并且无法用市场状况、Fama-French因子、交易量、信用评级或动量等因素来解释。提出了两种假设——收益高估和操纵。测试表明，可自由支配的应计项目驱动了动量利润，支持了收益操纵假说。基于应计项目的动量利润主要来源于高应计亏损股，这由与操纵和高估相关的向下收益驱动。

III. 来源论文

Accruals and Momentum [点击查看论文]

• 顾明和吴仰儒。厦门大学经济学院。罗格斯大学纽瓦克分校商学院金融与经济系

<摘要>

我们建立了动量与应计项目之间稳健的联系。动量盈利能力主要集中在高应计项目公司。此前记录的动量横截面特征并未涵盖应计项目对动量的影响。高应计项目的亏损股在随后几年中经历了行业调整后销售增长的显著下降以及最大数量的减少收入的特殊项目。高应计项目公司的大部分动量利润可归因于高可自由支配应计项目组。我们的发现表明，由于收益高估和收益操纵的共同作用，高应计项目亏损股的向下收益在很大程度上驱动了基于应计项目的动量利润。

IV. 回测表现

年化回报	11.22%
波动率	15.03%
β值	-0.045
夏普比率	0.75
索提诺比率	0.107
最大回撤	N/A
胜率	50%

V. 完整的 Python 代码

from AlgorithmImports import *
from numpy import isnan
from functools import reduce
#endregion
class AccuralsData():
    def __init__(self, 
                current_assets: float, 
                cash_and_cash_equivalents: float, 
                current_liabilities: float, 
                current_debt: float, 
                income_tax_payable: float, 
                depreciation_and_amortization: float, 
                total_assets: float
                ):
        self.current_assets = current_assets
        self.cash_and_cash_equivalents = cash_and_cash_equivalents
        self.current_liabilities = current_liabilities
        self.current_debt = current_debt
        self.income_tax_payable = income_tax_payable
        self.depreciation_and_amortization = depreciation_and_amortization
        self.total_assets = total_assets
class MomentumAndHighAccruals(QCAlgorithm):
    def Initialize(self):
        self.SetStartDate(2000, 1, 1)
        self.SetCash(100_000)
        
        self.data: Dict[Symbol, SymbolData] = {}
        self.accural_data = {} # Latest accurals data
        self.managed_queue = []
        
        self.period: int = 6 * 21
        self.holding_period: int = 6
        self.accrual_quantile: int = 3
        self.momentum_quantile: int = 5
        self.min_share_price: float = 5.
        self.leverage: int = 5
        self.fundamental_count: int = 3_000
        self.fundamental_sorting_key = lambda x: x.MarketCap
        self.exchange_codes: List[str] = ['NYS', 'NAS', 'ASE']
        self.financial_statement_names: List[str] = [
            'MarketCap',
            'FinancialStatements.BalanceSheet.CurrentAssets.TwelveMonths',
            'FinancialStatements.BalanceSheet.CashAndCashEquivalents.TwelveMonths',
            'FinancialStatements.BalanceSheet.CurrentLiabilities.TwelveMonths',
            'FinancialStatements.BalanceSheet.CurrentDebt.TwelveMonths',
            'FinancialStatements.BalanceSheet.IncomeTaxPayable.TwelveMonths',
            'FinancialStatements.IncomeStatement.DepreciationAndAmortization.TwelveMonths',
        ]
        market: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
        
        self.selection_flag: bool = True
        self.UniverseSettings.Resolution = Resolution.Daily
        self.AddUniverse(self.FundamentalSelectionFunction)
        self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
        self.Schedule.On(self.DateRules.MonthStart(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.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].update(stock.AdjustedPrice)
        if not self.selection_flag:
            return Universe.Unchanged
        selected: List[Fundamental] = [
            x for x in fundamental if (x.CompanyReference.IsREIT != 1)
            and x.Price > self.min_share_price
            and (x.AssetClassification.MorningstarSectorCode != MorningstarSectorCode.FinancialServices)
            and all((not isnan(self.rgetattr(x, statement_name)) and self.rgetattr(x, statement_name) != 0) for statement_name in self.financial_statement_names)
            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]]
        bs_acc: Dict[Fundamental, float] = {}
        momentum: Dict[Fundamental, float] = {}
        current_accurals_data: Dict[Symbol, AccuralsData] = {}
        # 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, 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].update(close)
            
            if self.data[symbol].is_ready():
                # accural calculation
                current_accurals_data[symbol] = AccuralsData(
                    stock.FinancialStatements.BalanceSheet.CurrentAssets.TwelveMonths, stock.FinancialStatements.BalanceSheet.CashAndCashEquivalents.TwelveMonths,
                    stock.FinancialStatements.BalanceSheet.CurrentLiabilities.TwelveMonths, stock.FinancialStatements.BalanceSheet.CurrentDebt.TwelveMonths, stock.FinancialStatements.BalanceSheet.IncomeTaxPayable.TwelveMonths,
                    stock.FinancialStatements.IncomeStatement.DepreciationAndAmortization.TwelveMonths, stock.FinancialStatements.BalanceSheet.TotalAssets.TwelveMonths
                )
            
                if symbol in self.accural_data:
                    bs_acc[stock] = self.CalculateAccurals(current_accurals_data[symbol], self.accural_data[symbol])
                    momentum[stock] = self.data[symbol].performance()
        
        # clear old accruals and set new ones 
        self.accural_data = current_accurals_data
        winners: List[Fundamental] = []
        losers: List[Fundamental] = []
        if len(momentum) >= self.momentum_quantile * self.accrual_quantile:
            # accural sorting
            sorted_by_acc: List[Tuple] = sorted(bs_acc.items(), key = lambda x: x[1], reverse = True)
            quantile: int = int(len(sorted_by_acc) / self.accrual_quantile)
            
            first_group: Fundamental = [x[0] for x in sorted_by_acc[:quantile]]
            second_group: Fundamental = [x[0] for x in sorted_by_acc[quantile:-quantile:]]
            third_group: Fundamental = [x[0] for x in sorted_by_acc[-quantile:]]
            
            # momentum sorting
            first_sorted_by_mom: Fundamental = sorted(first_group, key = lambda x: momentum[x], reverse = True)
            second_sorted_by_mom: Fundamental = sorted(second_group, key = lambda x: momentum[x], reverse = True)
            third_sorted_by_mom: Fundamental = sorted(third_group, key = lambda x: momentum[x], reverse = True)
            
            # selecting winners and losers
            first_quintile: int = int(len(first_sorted_by_mom) / self.momentum_quantile)
            second_quintile: int = int(len(second_sorted_by_mom) / self.momentum_quantile)
            third_quintile: int = int(len(third_sorted_by_mom) / self.momentum_quantile)
            
            winners = first_sorted_by_mom[:first_quintile] + second_sorted_by_mom[:second_quintile] + third_sorted_by_mom[:third_quintile]
            losers = first_sorted_by_mom[-first_quintile:] + second_sorted_by_mom[-second_quintile:] + third_sorted_by_mom[-third_quintile:]
    
            symbol_q: List[Tuple] = []
            if len(winners) != 0:
                winners_market_cap: float = sum([x.MarketCap for x in winners])
                long_w: float = self.Portfolio.TotalPortfolioValue / self.holding_period
                for stock in winners:
                    symbol_q.append((stock.Symbol, np.floor((long_w * (stock.MarketCap / winners_market_cap)) / self.data[stock.Symbol].last_price)))
            
            if len(losers) != 0:
                losers_market_cap: float = sum([x.MarketCap for x in losers])
                short_w: float = self.Portfolio.TotalPortfolioValue / self.holding_period
                for stock in losers:
                    symbol_q.append((stock.Symbol, -np.floor((short_w * (stock.MarketCap / losers_market_cap)) / self.data[stock.Symbol].last_price)))
            
            self.managed_queue.append(RebalanceQueueItem(symbol_q))
            
        return list(map(lambda x: x.Symbol, winners + losers))
    def OnData(self, slice: Slice) -> None:
        if not self.selection_flag:
            return
        self.selection_flag = False
        # trade execution
        remove_item = None
        
        # rebalance portfolio
        for item in self.managed_queue:
            if item.holding_period == self.holding_period + 1:
                for symbol, quantity in item.symbol_q:
                    self.MarketOrder(symbol, -quantity)
                            
                remove_item = item
                
            elif item.holding_period == 1:
                opened_symbol_q: List[Tuple[Symbol, float]] = []
                
                for symbol, quantity in item.symbol_q:
                    if slice.ContainsKey(symbol) and slice[symbol] is not None and self.Securities[symbol].IsTradable:
                        self.MarketOrder(symbol, quantity)
                        opened_symbol_q.append((symbol, quantity))
                            
                # only opened orders will be closed        
                item.symbol_q = opened_symbol_q
                
            item.holding_period += 1
            
        if remove_item:
            self.managed_queue.remove(remove_item)
        
    def Selection(self) -> None:
        self.selection_flag = True
    
    def CalculateAccurals(
        self, 
        current_accural_data: Dict[Symbol, AccuralsData], 
        prev_accural_data: Dict[Symbol, AccuralsData]
        ) -> float:
        delta_assets: float = current_accural_data.current_assets - prev_accural_data.current_assets
        delta_cash: float = current_accural_data.cash_and_cash_equivalents - prev_accural_data.cash_and_cash_equivalents
        delta_liabilities: float = current_accural_data.current_liabilities - prev_accural_data.current_liabilities
        delta_debt: float = current_accural_data.current_debt - prev_accural_data.current_debt
        delta_tax: float = current_accural_data.income_tax_payable - prev_accural_data.income_tax_payable
        dep: float = current_accural_data.depreciation_and_amortization
        avg_total: float = (current_accural_data.total_assets + prev_accural_data.total_assets) / 2
        
        bs_acc: float = ((delta_assets - delta_cash) - (delta_liabilities - delta_debt-delta_tax) - dep) / avg_total
        return bs_acc
    def rgetattr(self, obj, attr, *args):
        def _getattr(obj, attr):
            return getattr(obj, attr, *args)
        return reduce(_getattr, [obj] + attr.split('.'))
        
class RebalanceQueueItem():
    def __init__(self, symbol_q: Tuple[Symbol, float]) -> None:
        # symbol/quantity collections
        self.symbol_q: Tuple[Symbol, float] = symbol_q  
        self.holding_period: int = 0
        
class SymbolData():
    def __init__(self, period: int):
        self.closes: RollingWindow = RollingWindow[float](period)
        self.last_price: float|None = None
    
    def update(self, close: float) -> None:
        self.closes.Add(close)
        self.last_price = close
        
    def is_ready(self) -> bool:
        return self.closes.IsReady
        
    def performance(self) -> float:
        return self.closes[0] / self.closes[self.closes.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"))