“该策略交易纽约证券交易所/美国证券交易所股票，结合动量和应计项目排序，做多低应计项目的赢家，做空高应计项目的输家，投资组合重叠、等权重且每月重新平衡。”
资产类别: 股票 | 地区: 美国 | 周期: 每月 | 市场: 股票 | 关键词: 应计项目、动量
I. 策略概要
该策略交易价格高于5美元的纽约证券交易所/美国证券交易所普通股。股票被分为低买卖差价和高买卖差价投资组合，重点关注低买卖差价股票。股票根据过去6个月的回报（动量）和应计项目进行双重排序，分为五分位数。该策略在应计项目最低的动量赢家投资组合中建立多头头寸，在应计项目最高的动量输家投资组合中建立空头头寸。投资组合持有六个月，形成期和持有期之间有一个月的间隔，从而创建重叠投资组合。投资组合等权重，每月重新平衡，利用动量和应计项目的相互作用来获取回报。
II. 策略合理性
该论文将众所周知的动量异常和应计项目异常结合起来，形成一种增强的动量策略。研究结果与应计项目异常的盈利固着解释相符，表明投资者忽视了应计项目与现金流相比的较低持久性，从而加剧了错误定价并增加了动量收益。这支持了动量源于投资者未能准确处理信息准确的观点。
增强型策略在各种市场状况、投资者情绪水平和子周期中始终优于传统动量策略。其表现对一月效应、时间变化和交易成本具有鲁棒性。重要的是，应计项目的增量效应不能完全由现有资产定价模型或常见风险因素解释。
III. 来源论文
Persistence of Earnings Components and Price Momentum [点击查看论文]
- 徐芳明、曾铖、郑力懿，德克萨斯大学里奥格兰德河谷分校（原德克萨斯大学泛美分校）、布里斯托大学、香港理工大学、布里斯托大学
<摘要>
这项研究调查了自由现金流以及（横截面和时间序列）价格动量在预测未来股票回报方面的作用。在控制了另一个因素之后，过去的收益和自由现金流都能正向预测未来的股票回报，这表明现金流和动量都包含关于未来股票回报的有价值和独特的见解。购买过去高自由现金流的赢家和卖空过去低自由现金流的输家的策略，明显优于传统的动量交易策略。增强的业绩对投资者情绪、时间变化或交易成本不敏感。进一步的分析表明，增量的现金流效应主要归因于对股权/债务持有者的净分配。总的来说，我们的发现阐明了公司基本面在技术交易策略中的作用。
IV. 回测表现
| 年化回报 | 10.43% |
| 波动率 | 10.82% |
| β值 | 0.029 |
| 夏普比率 | 0.59 |
| 索提诺比率 | -0.098 |
| 最大回撤 | N/A |
| 胜率 | 52% |
V. 完整的 Python 代码
from numpy import floor, isnan
from AlgorithmImports import *
from typing import List, Dict
import data_tools
class AccrualsEffectPriceMomentum(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2000, 1, 1)
self.SetCash(100000)
self.exchange_codes:List[str] = ['NYS', 'ASE']
self.long:List[Symbol] = []
self.short:List[Symbol] = []
self.quantile:int = 5
self.leverage:int = 5
self.min_share_price:int = 5
self.months:int = 0
self.period:int = 6 * 21
self.holding_period:int = 6
self.data:Dict[Symbol, data_tools.SymbolData] = {}
self.managed_queue:List[data_tools.RebalanceQueueItem] = []
# Latest accurals data
self.accural_data:Dict[Symbol, data_tools.AccuralsData] = {}
self.symbol:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
self.fundamental_count:int = 1000
self.fundamental_sorting_key = lambda x: x.MarketCap
self.selection_flag:bool = False
self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
self.UniverseSettings.Resolution = Resolution.Daily
self.AddUniverse(self.FundamentalSelectionFunction)
self.Schedule.On(self.DateRules.MonthStart(self.symbol), self.TimeRules.AfterMarketOpen(self.symbol), self.Selection)
self.settings.daily_precise_end_time = False
def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
for security in changes.AddedSecurities:
security.SetFeeModel(data_tools.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 monthly 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.HasFundamentalData and x.Price > self.min_share_price and x.Market == 'usa' and x.MarketCap != 0
and not isnan(x.FinancialStatements.BalanceSheet.CurrentAssets.ThreeMonths) and (x.FinancialStatements.BalanceSheet.CurrentAssets.ThreeMonths > 0) \
and not isnan(x.FinancialStatements.BalanceSheet.CashAndCashEquivalents.ThreeMonths) and (x.FinancialStatements.BalanceSheet.CashAndCashEquivalents.ThreeMonths) > 0 \
and not isnan(x.FinancialStatements.BalanceSheet.CurrentLiabilities.ThreeMonths) and (x.FinancialStatements.BalanceSheet.CurrentLiabilities.ThreeMonths) > 0 \
and not isnan(x.FinancialStatements.BalanceSheet.CurrentDebt.ThreeMonths) and (x.FinancialStatements.BalanceSheet.CurrentDebt.ThreeMonths) > 0 \
and not isnan(x.FinancialStatements.BalanceSheet.IncomeTaxPayable.ThreeMonths) and (x.FinancialStatements.BalanceSheet.IncomeTaxPayable.ThreeMonths) > 0 \
and not isnan(x.FinancialStatements.IncomeStatement.DepreciationAndAmortization.ThreeMonths) and (x.FinancialStatements.IncomeStatement.DepreciationAndAmortization.ThreeMonths) > 0 \
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]]
# Warmup price rolling windows.
for stock in selected:
symbol:Symbol = stock.Symbol
if symbol in self.data:
continue
self.data[symbol] = data_tools.SymbolData(self.period)
history = self.History(symbol, self.period, Resolution.Daily)
if history.empty:
self.Log(f"Not enough data for {symbol} yet")
continue
closes = history.loc[symbol].close
for time, close in closes.items():
self.data[symbol].update(close)
bs_acc:Dict[Symbol, float] = {}
momentum:Dict[Symbol, float] = {}
current_accurals_data:Dict[Symbol, data_tools.AccuralsData] = {}
for stock in selected:
symbol = stock.Symbol
if not self.data[symbol].is_ready():
continue
momentum[symbol] = self.data[symbol].performance()
# Accural calc
current_accurals_data[symbol] = data_tools.AccuralsData(stock.FinancialStatements.BalanceSheet.CurrentAssets.ThreeMonths, stock.FinancialStatements.BalanceSheet.CashAndCashEquivalents.ThreeMonths,
stock.FinancialStatements.BalanceSheet.CurrentLiabilities.ThreeMonths, stock.FinancialStatements.BalanceSheet.CurrentDebt.ThreeMonths, stock.FinancialStatements.BalanceSheet.IncomeTaxPayable.ThreeMonths,
stock.FinancialStatements.IncomeStatement.DepreciationAndAmortization.ThreeMonths, stock.FinancialStatements.BalanceSheet.TotalAssets.ThreeMonths)
if symbol in self.accural_data:
bs_acc[symbol] = self.CalculateAccurals(current_accurals_data[symbol], self.accural_data[symbol])
# Clear old accruals and set new ones
self.accural_data.clear()
for symbol in current_accurals_data:
self.accural_data[symbol] = current_accurals_data[symbol]
long:List[Symbol] = []
short:List[Symbol] = []
if len(momentum) != 0 and len(bs_acc) != 0:
# Momentum sorting
sorted_by_mom:List[Tuple[Symbol, float]] = sorted(momentum.items(), key = lambda x: x[1], reverse = True)
quintile:int = int(len(sorted_by_mom) / self.quantile)
top_by_mom:List[Symbol] = [x[0] for x in sorted_by_mom[:quintile]]
low_by_mom:List[Symbol] = [x[0] for x in sorted_by_mom[-quintile:]]
# Accural sorting
sorted_by_acc:List[Tuple[Symbol, float]] = sorted(bs_acc.items(), key = lambda x: x[1], reverse = True)
quintile:int = int(len(sorted_by_acc) / self.quantile)
top_by_acc:List[Symbol] = [x[0] for x in sorted_by_acc[:quintile]]
low_by_acc:List[Symbol] = [x[0] for x in sorted_by_acc[-quintile:]]
long = [x for x in top_by_mom if x in low_by_acc]
short = [x for x in low_by_mom if x in top_by_acc]
if len(long) != 0:
long_w:float = self.Portfolio.TotalPortfolioValue / self.holding_period / len(long)
# symbol/quantity collection
long_symbol_q:List[Tuple[Symbol, int]] = [(x, floor(long_w / self.data[x].LastPrice)) for x in long]
else:
long_symbol_q = []
if len(short) != 0:
short_w:float = self.Portfolio.TotalPortfolioValue / self.holding_period / len(short)
# symbol/quantity collection
short_symbol_q:List[Tuple[Symbol, int]] = [(x, floor(short_w / self.data[x].LastPrice)) for x in short]
else:
short_symbol_q = []
self.managed_queue.append(data_tools.RebalanceQueueItem(long_symbol_q, short_symbol_q))
return long + short
def OnData(self, data: Slice) -> None:
if not self.selection_flag:
return
self.selection_flag = False
remove_item = None
# Rebalance portfolio
for item in self.managed_queue:
if item.holding_period == self.holding_period + 1: # All portfolios are held for six months (month 2 to 7)
# Selling long on Liquidate
for symbol, quantity in item.long_symbol_q:
self.MarketOrder(symbol, -quantity)
# Buying short on Liquidate
for symbol, quantity in item.short_symbol_q:
self.MarketOrder(symbol, quantity)
remove_item = item
# Trade execution
if item.holding_period == 1: # All portfolios are held for six months (month 2 to 7)
open_long_symbol_q:List[Tuple[Symbol, int]] = []
open_short_symbol_q:List[Tuple[Symbol, int]] = []
for symbol, quantity in item.long_symbol_q:
if self.Securities[symbol].Price != 0 and self.Securities[symbol].IsTradable:
self.MarketOrder(symbol, quantity)
open_long_symbol_q.append((symbol, quantity))
for symbol, quantity in item.short_symbol_q:
if self.Securities[symbol].Price != 0 and self.Securities[symbol].IsTradable:
self.MarketOrder(symbol, -quantity)
open_short_symbol_q.append((symbol, quantity))
# Only opened orders will be closed
item.long_symbol_q = open_long_symbol_q
item.short_symbol_q = open_short_symbol_q
item.holding_period += 1
# We need to remove closed part of portfolio after loop. Otherwise it will miss one item in self.managed_queue.
if remove_item:
self.managed_queue.remove(remove_item)
def Selection(self) -> None:
self.selection_flag = True
def CalculateAccurals(self, current_accural_data, prev_accural_data):
delta_assets:float = current_accural_data.CurrentAssets - prev_accural_data.CurrentAssets
delta_cash:float = current_accural_data.CashAndCashEquivalents - prev_accural_data.CashAndCashEquivalents
delta_liabilities:float = current_accural_data.CurrentLiabilities - prev_accural_data.CurrentLiabilities
delta_debt:float = current_accural_data.CurrentDebt - prev_accural_data.CurrentDebt
delta_tax:float = current_accural_data.IncomeTaxPayable - prev_accural_data.IncomeTaxPayable
dep:float = current_accural_data.DepreciationAndAmortization
avg_total:float = (current_accural_data.TotalAssets + prev_accural_data.TotalAssets) / 2
bs_acc:float = ((delta_assets - delta_cash) - (delta_liabilities - delta_debt-delta_tax) - dep) / avg_total
return bs_acc