# https://quantpedia.com/strategies/earnings-quality-factor/ # # The investment universe consists of all non-financial stocks from NYSE, Amex and Nasdaq. Big stocks are defined as the largest stocks # that make up 90% of the total market cap within the region, while small stocks make up the remaining 10% of the market cap. Investor defines # breakpoints by the 30th and 70th percentiles of the multiple “Earnings Quality” ratios between large caps and small caps. # The first “Earnings Quality” ratio is defined by cash flow relative to reported earnings. The high-quality earnings firms are characterized # by high cash flows (relative to reported earnings) while the low-quality firms are characterized by high reported earnings (relative to cash flow). # The second factor is based on return on equity (ROE) to exploit the well-documented “profitability anomaly” by going long high-ROE firms # (top 30%) and short low-ROE firms (bottom 30%). The third ratio – CF/A (cash flow to assets) factor goes long firms with high cash flow to total assets. # The fourth ratio – D/A (debt to assets) factor goes long firms with low leverage and short firms with high leverage. # The investor builds a scored composite quality metric by computing the percentile score of each stock on each of the four quality metrics # (where “good” quality has a high score, so ideally a stock has low accruals, low leverage, high ROE, and high cash flow) and then add up # the percentiles to get a score for each stock from 0 to 400. He then forms the composite factor by going long the top 30% of small-cap # stocks and also large-cap stocks and short the bottom 30% of the small-cap stocks and also large-cap stocks and cap-weighting individual # stocks within the portfolios. The final factor portfolio is formed at the end of each June and is rebalanced yearly. # # QC implementation changes: # - Universe consists of top 3000 US non-financial stocks by market cap from NYSE, AMEX and NASDAQ. class EarningsQualityFactor(QCAlgorithm): def Initialize(self): self.SetStartDate(2000, 1, 1) self.SetCash(100000) self.coarse_count = 3000 self.symbol = self.AddEquity("SPY", Resolution.Daily).Symbol self.accruals_data = {} self.long = [] self.short = [] self.data = {} self.selection_flag = True self.UniverseSettings.Resolution = Resolution.Daily self.AddUniverse(self.CoarseSelectionFunction, self.FineSelectionFunction) self.Schedule.On( self.DateRules.MonthEnd(self.symbol), self.TimeRules.AfterMarketOpen(self.symbol), self.Selection, ) def OnSecuritiesChanged(self, changes): for security in changes.AddedSecurities: security.SetLeverage(10) security.SetFeeModel(CustomFeeModel(self)) def CoarseSelectionFunction(self, coarse): if not self.selection_flag: return Universe.Unchanged selected = [ x.Symbol for x in coarse if x.HasFundamentalData and x.Market == "usa" ] return selected def FineSelectionFunction(self, fine): fine = [ x for x in fine if x.MarketCap != 0 and x.CompanyReference.IndustryTemplateCode != "B" and ( (x.SecurityReference.ExchangeId == "NYS") or (x.SecurityReference.ExchangeId == "NAS") or (x.SecurityReference.ExchangeId == "ASE") ) and x.FinancialStatements.BalanceSheet.CurrentAssets.Value != 0 and x.FinancialStatements.BalanceSheet.CashAndCashEquivalents.Value != 0 and x.FinancialStatements.BalanceSheet.CurrentLiabilities.Value != 0 and x.FinancialStatements.BalanceSheet.CurrentDebt.Value != 0 and x.FinancialStatements.IncomeStatement.DepreciationAndAmortization.Value != 0 and x.FinancialStatements.BalanceSheet.GrossPPE.Value != 0 and x.FinancialStatements.IncomeStatement.TotalRevenueAsReported.Value != 0 and x.FinancialStatements.CashFlowStatement.OperatingCashFlow.Value != 0 and x.EarningReports.BasicEPS.Value != 0 and x.EarningReports.BasicAverageShares.Value != 0 and x.OperationRatios.DebttoAssets.Value != 0 and x.OperationRatios.ROE.Value != 0 ] if len(fine) > self.coarse_count: sorted_by_market_cap = sorted(fine, key=lambda x: x.MarketCap, reverse=True) top_by_market_cap = [x for x in sorted_by_market_cap[: self.coarse_count]] else: top_by_market_cap = fine for stock in top_by_market_cap: symbol = stock.Symbol if symbol not in self.accruals_data: # Data for previous year. self.accruals_data[symbol] = None # Accrual calc. current_accruals_data = AcrrualsData( stock.FinancialStatements.BalanceSheet.CurrentAssets.Value, stock.FinancialStatements.BalanceSheet.CashAndCashEquivalents.Value, stock.FinancialStatements.BalanceSheet.CurrentLiabilities.Value, stock.FinancialStatements.BalanceSheet.CurrentDebt.Value, stock.FinancialStatements.BalanceSheet.IncomeTaxPayable.Value, stock.FinancialStatements.IncomeStatement.DepreciationAndAmortization.Value, stock.FinancialStatements.BalanceSheet.TotalAssets.Value, stock.FinancialStatements.IncomeStatement.TotalRevenueAsReported.Value, ) # There is not previous accruals data. if not self.accruals_data[symbol]: self.accruals_data[symbol] = current_accruals_data continue current_accruals = self.CalculateAccruals( current_accruals_data, self.accruals_data[symbol] ) # cash flow to assets CFA = ( stock.FinancialStatements.CashFlowStatement.OperatingCashFlow.Value / ( stock.EarningReports.BasicEPS.Value * stock.EarningReports.BasicAverageShares.Value ) ) # debt to assets DA = stock.OperationRatios.DebttoAssets.Value # return on equity ROE = stock.OperationRatios.ROE.Value if symbol not in self.data: self.data[symbol] = None self.data[symbol] = StockData(current_accruals, CFA, DA, ROE) self.accruals_data[symbol] = current_accruals_data # Remove not updated symbols. updated_symbols = [x.Symbol for x in top_by_market_cap] not_updated = [x for x in self.data if x not in updated_symbols] for symbol in not_updated: del self.data[symbol] del self.accruals_data[symbol] return [x[0] for x in self.data.items()] def OnData(self, data): if not self.selection_flag: return self.selection_flag = False # Sort stocks by four factors respectively. sorted_by_accruals = sorted( self.data.items(), key=lambda x: x[1].Accruals, reverse=True ) # high score with low accrual sorted_by_CFA = sorted( self.data.items(), key=lambda x: x[1].CFA ) # high score with high CFA sorted_by_DA = sorted( self.data.items(), key=lambda x: x[1].DA, reverse=True ) # high score with low leverage sorted_by_ROE = sorted( self.data.items(), key=lambda x: x[1].ROE ) # high score with high ROE score = {} # Assign a score to each stock according to their rank with different factors. for i, obj in enumerate(sorted_by_accruals): score_accruals = i score_CFA = sorted_by_CFA.index(obj) score_DA = sorted_by_DA.index(obj) score_ROE = sorted_by_ROE.index(obj) score[obj[0]] = score_accruals + score_CFA + score_DA + score_ROE sorted_by_score = sorted(score.items(), key=lambda x: x[1], reverse=True) tercile = int(len(sorted_by_score) / 3) long = [x[0] for x in sorted_by_score[:tercile]] short = [x[0] for x in sorted_by_score[-tercile:]] # Trade execution. # NOTE: Skip year 2007 due to data error. if self.Time.year == 2007: self.Liquidate() return stocks_invested = [x.Key for x in self.Portfolio if x.Value.Invested] for symbol in stocks_invested: if symbol not in long + short: self.Liquidate(symbol) for symbol in long: if ( self.Securities[symbol].Price != 0 and self.Securities[symbol].IsTradable ): # Prevent error message. self.SetHoldings(symbol, 1 / len(long)) for symbol in short: if ( self.Securities[symbol].Price != 0 and self.Securities[symbol].IsTradable ): # Prevent error message. self.SetHoldings(symbol, -1 / len(short)) # Source: https://papers.ssrn.com/sol3/papers.cfm?abstract_id=3188172 def CalculateAccruals(self, current_accrual_data, prev_accrual_data): delta_assets = ( current_accrual_data.CurrentAssets - prev_accrual_data.CurrentAssets ) delta_cash = ( current_accrual_data.CashAndCashEquivalents - prev_accrual_data.CashAndCashEquivalents ) delta_liabilities = ( current_accrual_data.CurrentLiabilities - prev_accrual_data.CurrentLiabilities ) delta_debt = current_accrual_data.CurrentDebt - prev_accrual_data.CurrentDebt dep = current_accrual_data.DepreciationAndAmortization total_assets_prev_year = prev_accrual_data.TotalAssets acc = ( delta_assets - delta_liabilities - delta_cash + delta_debt - dep ) / total_assets_prev_year return acc def Selection(self): if self.Time.month == 7: self.selection_flag = True class AcrrualsData: def __init__( self, current_assets, cash_and_cash_equivalents, current_liabilities, current_debt, income_tax_payable, depreciation_and_amortization, total_assets, sales, ): self.CurrentAssets = current_assets self.CashAndCashEquivalents = cash_and_cash_equivalents self.CurrentLiabilities = current_liabilities self.CurrentDebt = current_debt self.IncomeTaxPayable = income_tax_payable self.DepreciationAndAmortization = depreciation_and_amortization self.TotalAssets = total_assets self.Sales = sales class StockData: def __init__(self, accruals, cfa, da, roe): self.Accruals = accruals self.CFA = cfa self.DA = da self.ROE = roe def MultipleLinearRegression(x, y): x = np.array(x).T x = sm.add_constant(x) result = sm.OLS(endog=y, exog=x).fit() return result # Custom fee model class CustomFeeModel(FeeModel): def GetOrderFee(self, parameters): fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005 return OrderFee(CashAmount(fee, "USD"))