“该投资范围包括MSCI发达市场标准指数中的大中盘股票，不包括金融和房地产公司。”
资产类别: 股票 | 地区: 全球 | 周期: 每季度 | 市场: 股票 | 关键词: 质量
I. 策略概要
该投资范围包括MSCI发达市场标准指数中的大中盘股票，不包括金融和房地产公司。对于其余公司，首先按以下方式计算质量分数：
对于四个维度中的每一个，都有两个指标：
- 盈利能力——毛利润与资产比率（GPA）、投资资本现金流回报率（CFROIC）；
- 盈利质量——斯隆应计项目（WCAcc）、现金流应计项目（AccCF）；
- 安全性——长期债务与股本比率（LTDE）、营运资本与资产比率（WCA）；
- 投资——资产增长1年（AG1Y）、资本支出与销售额比率（Capex）。
首先，计算每个指标的值（参见研究论文第3节中的公式），并将其转换为百分位数。当然，最高质量的公司应该拥有最高的百分位数。通过将两个指标的百分位数平均为该维度的新单一百分位数，将两个指标组合成该维度的值。使用逆正态分布函数将其转换为z分数，并将所有4个z分数平均以获得一个质量分数。其次，根据质量分数的值创建按五分位数排序的价值加权投资组合。做多最高五分位数，做空最低五分位数。该策略每季度重新平衡。
II. 策略合理性
作者认为，质量溢价的来源是风险因子和行为偏差的结合。由于其性质，质量是多维的，难以准确捕捉。作者提供了一个复杂的模型，由4个有充分记录的质量衡量指标组成，每个指标都由2个精心挑选的度量标准描述。所有这些指标之前都经过了学术研究，证实了它们预测股票表现的能力。然而，与此同时，该策略产生的质量溢价并非由这些组成部分中的任何一个单独驱动，而是由它们的组合驱动。同时，通过为每个支柱选择两个而不是一个度量标准，显著降低了错误分类股票的可能性。选择大中盘股并仅使用近年来的样本确保了结果的时效性，对愿意在实践中实施此策略的投资者有用。
III. 来源论文
Revisiting Quality Investing [Click to Open PDF]
- Frederic Lepetit、Amina Cherief、Yannick Ly、Takaya Sekine。东方汇理研究所、东方汇理研究所、东方汇理资产管理公司、东方汇理研究所
<摘要>
在因子投资领域，质量无疑是共识最弱的股票因子。本研究探讨了定义它的最佳方式。为了捕捉学术界所描绘的因子多方面现实，我们通过定义四个独立的支柱：盈利能力、盈利质量、安全性和投资，来处理质量因子。为了更好地满足机构投资者的需求，我们分析了由此产生的因子，重点关注过去十八年以及全球发达市场流动性股票（大盘股和中盘股）的投资范围。
在多空框架下，我们的质量因子提供了统计上显著的阿尔法，这无法通过对传统股票因子（市场、价值、规模和动量）的载荷来解释。大多数地区和维度都对这个阿尔法做出了积极贡献，但欧元区和安全维度是明显的例外。在仅做多框架下，我们的质量因子在整个分析期间每年跑赢其基准2.8%，信息比率为0.81。此外，自2008年全球金融危机（GFC）以来，超额表现一直非常稳定。这四个维度之间相关性较弱，因此是互补的。我们表明，在市场动荡时期（全球金融危机、新冠疫情），安全性尤为重要，因此该维度本身就是质量因子的一部分。在欧元区方面，行业中性投资组合构建似乎更合适。
我们还引入了一种新的投资组合构建方法，通过实施基于K-means算法的聚类方法，根据与基本面和市场特征相关的特征对公司进行分组。这种方法可以捕捉基本面和其他股票特征之间的动态变化。这个完全可实施的过程带来了更好的质量因子表现，而不会影响相关的风险衡量或投资组合的质量敞口（根据无约束质量因子衡量）。
IV. 回测表现
| 年化回报 | 5% |
| 波动率 | 6.4% |
| β值 | 0.035 |
| 夏普比率 | 0.78 |
| 索提诺比率 | -0.022 |
| 最大回撤 | N/A |
| 胜率 | 50% |
V. 完整的 Python 代码
from AlgorithmImports import *
from scipy import stats
from typing import List, Dict
from numpy import isnan
#endregion
class RobustQualityInStocks(QCAlgorithm):
def Initialize(self) -> None:
self.SetStartDate(2000, 1, 1)
self.SetCash(100_000)
self.period: int = 4 # need n of quarterly data
self.leverage: int = 10
self.quantile: int = 5
self.min_share_price: int = 5
self.data: Dict[Symbol, SymbolData] = {}
self.weights: Dict[Symbol, float] = {}
self.last_selection: List[Symbol] = []
market: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
self.months_counter: int = 0
self.selection_flag: bool = False
self.UniverseSettings.Leverage = self.leverage
self.UniverseSettings.Resolution = Resolution.Daily
self.AddUniverse(self.FundamentalSelectionFunction)
self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
self.Schedule.On(self.DateRules.MonthStart(market), self.TimeRules.BeforeMarketClose(market, 0), self.Selection)
self.settings.daily_precise_end_time = False
def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
for security in changes.AddedSecurities:
security.SetFeeModel(CustomFeeModel())
def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
# quarterly rebalance
if not self.selection_flag:
return Universe.Unchanged
selected = [
x for x in fundamental
if x.HasFundamentalData
and x.Market == 'usa'
and x.Price > self.min_share_price
and x.MarketCap != 0
and not isnan(x.FinancialStatements.IncomeStatement.GrossProfit.ThreeMonths) and x.FinancialStatements.IncomeStatement.GrossProfit.ThreeMonths != 0
and not isnan(x.FinancialStatements.BalanceSheet.TotalAssets.ThreeMonths) and x.FinancialStatements.BalanceSheet.TotalAssets.ThreeMonths != 0
and not isnan(x.FinancialStatements.BalanceSheet.WorkingCapital.ThreeMonths) and x.FinancialStatements.BalanceSheet.WorkingCapital.ThreeMonths != 0
and not isnan(x.FinancialStatements.CashFlowStatement.CapExReported.ThreeMonths) and x.FinancialStatements.CashFlowStatement.CapExReported.ThreeMonths != 0
and not isnan(x.FinancialStatements.IncomeStatement.TotalRevenue.ThreeMonths) and x.FinancialStatements.IncomeStatement.TotalRevenue.ThreeMonths != 0
and not isnan(x.OperationRatios.ROIC.OneYear) and x.OperationRatios.ROIC.OneYear != 0
and not isnan(x.OperationRatios.LongTermDebtEquityRatio.OneYear) and x.OperationRatios.LongTermDebtEquityRatio.OneYear != 0
and not isnan(x.OperationRatios.TotalAssetsGrowth.OneYear) and x.OperationRatios.TotalAssetsGrowth.OneYear != 0
# and not isnan(x.FinancialStatements.CashFlowStatement.ChangeInOtherCurrentAssets.ThreeMonths) and x.FinancialStatements.CashFlowStatement.ChangeInOtherCurrentAssets.ThreeMonths != 0
# and not isnan(x.FinancialStatements.CashFlowStatement.ChangeInInventory.ThreeMonths) and x.FinancialStatements.CashFlowStatement.ChangeInInventory.ThreeMonths != 0
# and not isnan(x.FinancialStatements.CashFlowStatement.ChangeInReceivables.ThreeMonths) and x.FinancialStatements.CashFlowStatement.ChangeInReceivables.ThreeMonths != 0
# and not isnan(x.OperationRatios.TotalLiabilitiesGrowth.OneYear) and x.OperationRatios.TotalLiabilitiesGrowth.OneYear != 0
# and not isnan(x.FinancialStatements.IncomeStatement.Depreciation.TwelveMonths) and x.FinancialStatements.IncomeStatement.Depreciation.TwelveMonths != 0
and not isnan(x.FinancialStatements.IncomeStatement.NetIncome.TwelveMonths) and x.FinancialStatements.IncomeStatement.NetIncome.TwelveMonths != 0
and not isnan(x.FinancialStatements.CashFlowStatement.OperatingCashFlow.TwelveMonths) and x.FinancialStatements.CashFlowStatement.OperatingCashFlow.TwelveMonths != 0
and not isnan(x.FinancialStatements.CashFlowStatement.InvestingCashFlow.TwelveMonths) and x.FinancialStatements.CashFlowStatement.InvestingCashFlow.TwelveMonths != 0
and not isnan(x.FinancialStatements.BalanceSheet.Cash.ThreeMonths) and x.FinancialStatements.BalanceSheet.Cash.ThreeMonths != 0
and not isnan(x.FinancialStatements.BalanceSheet.TotalLiabilitiesAsReported.ThreeMonths) and x.FinancialStatements.BalanceSheet.TotalLiabilitiesAsReported.ThreeMonths != 0
and not isnan(x.FinancialStatements.BalanceSheet.CurrentDebt.ThreeMonths) and x.FinancialStatements.BalanceSheet.CurrentDebt.ThreeMonths != 0
and not isnan(x.FinancialStatements.BalanceSheet.LongTermDebt.ThreeMonths) and x.FinancialStatements.BalanceSheet.LongTermDebt.ThreeMonths != 0
]
market_cap: Dict[Symbol, float] = {} # storing stocks market capitalization keyed by stocks symbols
# storing metrics values keyed by stock symbols
GPA: Dict[Symbol, float] = {} # GrossProfit / TotalAssets
CFROIC: Dict[Symbol, float] = {} # ROIC.TwelveMonths
LTDE: Dict[Symbol, float] = {} # LongTermDebtEquityRatio.TwelveMonths
AG1Y: Dict[Symbol, float] = {} # TotalAssetsGrowth.TwelveMonths
WCA: Dict[Symbol, float] = {} # WorkingCapital / TotalAssets
Capex: Dict[Symbol, float] = {} # CapExReported / TotalRevenue
# WCAcc: Dict[Symbol, float] = {} # (ChangeInOtherCurrentAssets + ChangeInInventor + ChangeInReceivables) / mean TA - TotalLiabilitiesGrowth / mean TA - Depreciation / mean TA
AccCF: Dict[Symbol, float] = {} #
for stock in selected:
symbol: Symbol = stock.Symbol
total_assets: float = stock.FinancialStatements.BalanceSheet.TotalAssets.ThreeMonths # TA
gross_profit: float = stock.FinancialStatements.IncomeStatement.GrossProfit.ThreeMonths
working_capital: float = stock.FinancialStatements.BalanceSheet.WorkingCapital.ThreeMonths
cap_ex_reported: float = stock.FinancialStatements.CashFlowStatement.CapExReported.ThreeMonths
total_revenue: float = stock.FinancialStatements.IncomeStatement.TotalRevenue.ThreeMonths
# get all needed values for NOA calculation
cash: float = stock.FinancialStatements.BalanceSheet.Cash.ThreeMonths # CHEi in formula for NOA
total_liabilities_as_reported: float = stock.FinancialStatements.BalanceSheet.TotalLiabilitiesAsReported.ThreeMonths # LT in formula for NOA
current_debt: float = stock.FinancialStatements.BalanceSheet.CurrentDebt.ThreeMonths # DLC in formula for NOA
long_term_debt: float = stock.FinancialStatements.BalanceSheet.LongTermDebt.ThreeMonths # DLTT in formula for NOA
# make sure data are consecutive
if symbol not in self.last_selection:
self.data[symbol] = SymbolData(self.period)
# calculate NOA for this quarter
# NOA = mean(TA - CHEi) - mean(LT-DLC-DLTT)
mean_TA_CHEi: float = np.mean([total_assets, cash])
mean_LT_DLC_DLTT: float = np.mean([total_liabilities_as_reported, current_debt, long_term_debt])
noa_value: float = mean_TA_CHEi - mean_LT_DLC_DLTT
# update NOA values
self.data[symbol].update_NOA(noa_value)
# update quarterly values
self.data[symbol].update_total_assets(total_assets)
gpa_value: float = gross_profit / total_assets
self.data[symbol].update_GPA(gpa_value)
wca_value: float = working_capital / total_assets
self.data[symbol].update_WCA(wca_value)
capex_value: float = cap_ex_reported / total_revenue
self.data[symbol].update_Capex(capex_value)
# check if all quarterly data are ready for metrics calculations
if not self.data[symbol].are_quarterly_data_ready():
continue
# data for each metric are ready
# calculate and store metrics
GPA[symbol] = self.data[symbol].calculate_GPA()
CFROIC[symbol] = stock.OperationRatios.ROIC.OneYear
LTDE[symbol] = stock.OperationRatios.LongTermDebtEquityRatio.OneYear
AG1Y[symbol] = stock.OperationRatios.TotalAssetsGrowth.OneYear
WCA[symbol] = self.data[symbol].calculate_WCA()
Capex[symbol] = self.data[symbol].calculate_Capex()
# # get all needed values for WCAcc metric
# change_in_other_current_assets = stock.FinancialStatements.CashFlowStatement.ChangeInOtherCurrentAssets.ThreeMonths
# change_in_inventory = stock.FinancialStatements.CashFlowStatement.ChangeInInventory.ThreeMonths
# change_in_receivables = stock.FinancialStatements.CashFlowStatement.ChangeInReceivables.ThreeMonths
# total_liabilities = stock.OperationRatios.TotalLiabilitiesGrowth.OneYear / 4
# depreciation = stock.FinancialStatements.IncomeStatement.Depreciation.TwelveMonths
# mean_total_assets = self.data[symbol].mean_total_assets()
# # calculate parts of WCAcc formula
# first_formula_part = (change_in_other_current_assets + change_in_inventory + change_in_receivables) / mean_total_assets
# second_formula_part = total_liabilities / mean_total_assets
# third_formula_part = depreciation / mean_total_assets
# # calculate WCAcc value
# wcacc_value = first_formula_part - second_formula_part - third_formula_part
# # store WCAcc metric value keyed by stock's symbol
# WCAcc[symbol] = wcacc_value
# get all needed values for AccCF
net_income: float = stock.FinancialStatements.IncomeStatement.NetIncome.TwelveMonths
operating_cashflow: float = stock.FinancialStatements.CashFlowStatement.OperatingCashFlow.TwelveMonths
investing_cashflow: float = stock.FinancialStatements.CashFlowStatement.InvestingCashFlow.TwelveMonths
# get NOA values for specific quarters
current_quarter_NOA, fourth_quarter_NOA = self.data[symbol].get_NOA_values()
# calculate NOA formula part for AccCF formula
NOA_formula_part: float = (current_quarter_NOA + fourth_quarter_NOA) / 2
# calculate AccCF value
acccf_value: float = (net_income - (operating_cashflow + investing_cashflow)) / NOA_formula_part
# store AccCF value keyed by stock's symbol
AccCF[symbol] = acccf_value
# store stock's market capitalization
market_cap[symbol] = stock.MarketCap
# change last_selection for consecutive data
self.last_selection = [x.Symbol for x in selected]
# make sure, there are enough data for quintile selection
if len(GPA) < self.quantile:
return Universe.Unchanged
# calculate percentil value for each metric and create dictinories keyed by stock symbols
GPA_percentiles: Dict[Symbol, float] = self.CalculatePercentiles(GPA)
CFROIC_percentiles: Dict[Symbol, float] = self.CalculatePercentiles(CFROIC)
LTDE_percentiles: Dict[Symbol, float] = self.CalculatePercentiles(LTDE)
AG1Y_percentiles: Dict[Symbol, float] = self.CalculatePercentiles(AG1Y)
WCA_percentiles: Dict[Symbol, float] = self.CalculatePercentiles(WCA)
Capex_percentiles: Dict[Symbol, float] = self.CalculatePercentiles(Capex)
# WCAcc_percentiles: Dict[Symbol, float] = self.CalculatePercentiles(WCAcc)
AccCF_percentiles: Dict[Symbol, float] = self.CalculatePercentiles(AccCF)
# calcualte dimension values
# dimension_value = np.mean(metric1_percentile + metric2_percentile)
# dimension value = mean of stock's percentile from first and second metric in current division
# for each dimension create dictionary with stocks dimension values keyed by stocks symbols
profitability: Dict[Symbol, float] = self.CalculateDimensionValues(GPA_percentiles, CFROIC_percentiles)
earnings_quality: Dict[Symbol, float] = self.CalculateDimensionValues(AccCF_percentiles, {})
safety: Dict[Symbol, float] = self.CalculateDimensionValues(LTDE_percentiles, WCA_percentiles)
investment: Dict[Symbol, float] = self.CalculateDimensionValues(AG1Y_percentiles, Capex_percentiles)
# calculate z-scores of stocks for each dimension
# for each dimension create dictionary with stocks z-scores keyed by stocks symbols
profitability_z_scores: Dict[Symbol, float] = self.CalculateDimensionZScore(profitability)
earnings_quality_z_scores: Dict[Symbol, float] = self.CalculateDimensionZScore(earnings_quality)
safety_z_scores: Dict[Symbol, float] = self.CalculateDimensionZScore(safety)
investment_z_scores: Dict[Symbol, float] = self.CalculateDimensionZScore(investment)
# create dictionary of quality factor values keyed by stocks symbols based on dimensions z-scores
quality_factor: Dict[Symbol, float] = self.CalculateQualityFactor(
profitability_z_scores,
earnings_quality_z_scores,
safety_z_scores,
investment_z_scores
)
# sort stocks by their quality factor
quantile: int = int(len(quality_factor) / self.quantile)
sorted_by_quality_factor: List[Symbol] = [x[0] for x in sorted(quality_factor.items(), key=lambda item: item[1])]
# long highest quintile according to quality factor
long: List[Symbol] = sorted_by_quality_factor[-quantile:]
# short lowest quintile according to quality factor
short: List[Symbol] = sorted_by_quality_factor[:quantile]
# calculate weights
for i, portfolio in enumerate([long, short]):
mc_sum: float = sum(list(map(lambda symbol: market_cap[symbol], portfolio)))
for symbol in portfolio:
self.weights[symbol] = ((-1)**i) * market_cap[symbol] / mc_sum
return list(self.weights.keys())
def OnData(self, slice: Slice) -> None:
if not self.selection_flag:
return
self.selection_flag = False
# trade exectuion
portfolio: List[PortfolioTarget] = [PortfolioTarget(symbol, w) for symbol, w in self.weights.items() if slice.contains_key(symbol) and slice[symbol]]
self.SetHoldings(portfolio, True)
self.weights.clear()
def CalculatePercentiles(self, metric_dict: Dict[Symbol, float]) -> Dict[Symbol, float]:
''' create and return dictionary with percentile values of metric keyed by stock symbol '''
# return dict of stocks percentiles values keyed by their symbols
metric_percentiles: Dict[Symbol, float] = { x[0] : stats.percentileofscore(list(metric_dict.values()), x[1]) for x in metric_dict.items() }
return metric_percentiles
def CalculateDimensionValues(self,
metric1_percentiles: Dict[Symbol, float],
metric2_percentiles: Dict[Symbol, float]) -> Dict[Symbol, float]:
''' create and return dictionary of dimension values keyed by stocks symbols '''
dimension_values: Dict[Symbol, float] = {} # storing dimension values keyed by stocks symbols
# calculate and store stock dimension values
for stock_symbol, stock_percentile_1 in metric1_percentiles.items():
if len(metric2_percentiles) > 0:
# get stock's percentile from second metric in current dimension
stock_percentile_2: float = metric2_percentiles[stock_symbol]
# calculate dimension value
# dimension value = mean of stock's percentile from first and second metric in current division
dimension_value: float = np.mean([stock_percentile_1, stock_percentile_2])
# store stock's dimension value keyed by stock's symbol
dimension_values[stock_symbol] = dimension_value
else:
# store stock's dimension value keyed by stock's symbol
dimension_values[stock_symbol] = stock_percentile_1
# return dictionary of dimension values keyed by stocks symbols
return dimension_values
def CalculateDimensionZScore(self, dimension_dict: Dict[Symbol, float]) -> Dict[Symbol, float]:
''' create and return dictionary of z-score values for each stock in dimension keyed by their symbols '''
dimension_values: List[float] = [value for _, value in dimension_dict.items()]
dimension_mean: float = np.mean(dimension_values)
dimension_std: float = np.std(dimension_values)
dimension_z_score: Dict[Symbol, float] = {} # storing stocks z-score values keyed by their symbols
# calculate and store z-score for each stock in dimension
for stock_symbol, value in dimension_dict.items():
z_score: float = (value - dimension_mean) / dimension_std
# store stock's z-score keyed by stock's symbol
dimension_z_score[stock_symbol] = z_score
# return dictionary with stocks z-scores keyed by their symbols
return dimension_z_score
def CalculateQualityFactor(self,
dimension_dict1: Dict[Symbol, float],
dimension_dict2: Dict[Symbol, float],
dimension_dict3: Dict[Symbol, float],
dimension_dict4: Dict[Symbol, float]) -> Dict[Symbol, float]:
''' create and return dictionary of quality factor values keyed by stocks symbols '''
quality_factor: Dict[Symbol, float] = {} # storing quality factor values keyed by stocks symbols
for stock_symbol, z_score1 in dimension_dict1.items():
z_score2: float = dimension_dict2[stock_symbol]
z_score3: float = dimension_dict3[stock_symbol]
z_score4: float = dimension_dict4[stock_symbol]
# calculate stock's quality factor based on dimensions z-scores
quality_factor_value: float = np.mean([z_score1, z_score2, z_score3, z_score4])
# store stock's quality factore keyed by stock's symbols
quality_factor[stock_symbol] = quality_factor_value
# return dictionary of quality factor values keyed by stocks symbols
return quality_factor
def Selection(self) -> None:
# quarterly rebalance
if self.months_counter % 3 == 0:
self.selection_flag = True
self.months_counter += 1
class SymbolData():
def __init__(self, period: int) -> None:
self.GPA: RollingWindow = RollingWindow[float](period)
self.WCA: RollingWindow = RollingWindow[float](period)
self.Capex: RollingWindow = RollingWindow[float](period)
self.total_assets: RollingWindow = RollingWindow[float](period)
self.NOA: RollingWindow = RollingWindow[float](period)
def update_NOA(self, noa_value: float) -> None:
self.NOA.Add(noa_value)
def update_total_assets(self, total_assets_value: float) -> None:
self.total_assets.Add(total_assets_value)
def update_GPA(self, gpa_value: float) -> None:
self.GPA.Add(gpa_value)
def update_WCA(self, wca_value: float) -> None:
self.WCA.Add(wca_value)
def update_Capex(self, capex_value: float) -> None:
self.Capex.Add(capex_value)
def are_quarterly_data_ready(self) -> bool:
return self.GPA.IsReady and self.WCA.IsReady and self.Capex.IsReady and self.total_assets.IsReady and self.NOA.IsReady
def mean_total_assets(self) -> float:
total_assets_values = [x for x in self.total_assets]
return np.mean(total_assets_values)
def get_NOA_values(self) -> float:
noa_values = [x for x in self.NOA]
# return NOA values in quartet t and quarter t-4
return noa_values[0], noa_values[-1]
def calculate_GPA(self) -> float:
gpa_values = [x for x in self.GPA]
return sum(gpa_values)
def calculate_WCA(self) -> float:
wca_values = [x for x in self.WCA]
return sum(wca_values)
def calculate_Capex(self) -> float:
capex_values = [x for x in self.Capex]
return sum(capex_values)
# Custom fee model.
class CustomFeeModel(FeeModel):
def GetOrderFee(self, parameters: OrderFeeParameters) -> OrderFee:
fee: float = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
return OrderFee(CashAmount(fee, "USD"))