Overall Statistics Total Trades 3962 Average Win 2.17% Average Loss 0.16% Compounding Annual Return -2.325% Drawdown 66.600% Expectancy 6.280 Net Profit -34.676% Sharpe Ratio -0.089 Sortino Ratio -0.079 Probabilistic Sharpe Ratio 0.000% Loss Rate 51% Win Rate 49% Profit-Loss Ratio 13.82 Alpha -0.025 Beta 0.124 Annual Standard Deviation 0.192 Annual Variance 0.037 Information Ratio -0.33 Tracking Error 0.238 Treynor Ratio -0.138 Total Fees \$196.77 Estimated Strategy Capacity \$14000.00 Lowest Capacity Asset BTBD XTFB9J43NOTH Portfolio Turnover 0.96%
```# https://quantpedia.com/strategies/accrual-anomaly/
#
# The investment universe consists of all stocks on NYSE, AMEX, and NASDAQ. Balance sheet based accruals (the non-cash component of
# earnings) are calculated as: BS_ACC = ( ∆CA – ∆Cash) – ( ∆CL – ∆STD – ∆ITP) – Dep
# Where:
# ∆CA = annual change in current assets
# ∆Cash = change in cash and cash equivalents
# ∆CL = change in current liabilities
# ∆STD = change in debt included in current liabilities
# ∆ITP = change in income taxes payable
# Dep = annual depreciation and amortization expense
# Stocks are then sorted into deciles and investor goes long stocks with the lowest accruals and short stocks with the highest accruals.
# The portfolio is rebalanced yearly during May (after all companies publish their earnings).
#
# QC implementation changes:
#   - Investment universe consists of 3000 largest stocks traded on NYSE, AMEX and NASDAQ.

from AlgorithmImports import *
from typing import List, Dict
import numpy as np

class AccrualAnomaly(QCAlgorithm):

def Initialize(self) -> None:
self.SetStartDate(2006, 1, 1)
self.SetCash(100_000)

self.UniverseSettings.Leverage = 5
self.UniverseSettings.Resolution = Resolution.Daily
self.Settings.MinimumOrderMarginPortfolioPercentage = 0.0

# Strategy Parameters
self.exchange_codes: List[str] = ['NYS', 'NAS', 'ASE']
self.quantile: int = 10
self.rebalancing_month: int = 5

self.fundamental_count: int = 3000
self.fundamental_sorting_key = lambda x: x.MarketCap

# Latest accruals data.
self.accrual_data: Disct[Symbol, AccrualsData] = {}
self.long_symbols: List[Symbol] = []
self.short_symbols: List[Symbol] = []
self.selection_flag: bool = False

self.exchange: Symbol = self.AddEquity("SPY", Resolution.Daily).Symbol
self.Schedule.On(self.DateRules.MonthStart(self.exchange),
self.TimeRules.AfterMarketOpen(self.exchange),
self.Selection)

def FundamentalFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
if not self.selection_flag:
return Universe.Unchanged

filtered: List[Fundamental] = [f for f in fundamental if f.HasFundamentalData
and f.SecurityReference.ExchangeId in self.exchange_codes
and not np.isnan(f.FinancialStatements.BalanceSheet.CurrentAssets.TwelveMonths)
and not np.isnan(f.FinancialStatements.BalanceSheet.CashAndCashEquivalents.TwelveMonths)
and not np.isnan(f.FinancialStatements.BalanceSheet.CurrentLiabilities.TwelveMonths)
and not np.isnan(f.FinancialStatements.BalanceSheet.CurrentDebt.TwelveMonths)
and not np.isnan(f.FinancialStatements.BalanceSheet.IncomeTaxPayable.TwelveMonths)
and not np.isnan(f.FinancialStatements.IncomeStatement.DepreciationAndAmortization.TwelveMonths)]

if len(filtered) > self.fundamental_count:
filtered = [x for x in sorted(filtered, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]

accruals: Dict[Symbol, float] = {}
for security in filtered:
symbol: Symbol = security.Symbol

if symbol not in self.accrual_data:
self.accrual_data[symbol] = None

# Accrual calc.
current_accruals_data: AccrualsData = AccrualsData(
security.FinancialStatements.BalanceSheet.CurrentAssets.TwelveMonths,
security.FinancialStatements.BalanceSheet.CashAndCashEquivalents.TwelveMonths,
security.FinancialStatements.BalanceSheet.CurrentLiabilities.TwelveMonths,
security.FinancialStatements.BalanceSheet.CurrentDebt.TwelveMonths,
security.FinancialStatements.BalanceSheet.IncomeTaxPayable.TwelveMonths,
security.FinancialStatements.IncomeStatement.DepreciationAndAmortization.TwelveMonths,
security.FinancialStatements.BalanceSheet.TotalAssets.TwelveMonths)

# There is not previous accrual data.
if not self.accrual_data[symbol]:
self.accrual_data[symbol] = current_accruals_data
continue

# Accruals and market cap calc.
accruals[symbol] = self.CalculateAccruals(current_accruals_data, self.accrual_data[symbol])

# Update accruals data.
self.accrual_data[symbol] = current_accruals_data

# Accruals sorting.
if len(accruals) >= self.quantile:
sorted_by_accruals: Dict[Symbol, float] = sorted(accruals.items(),
key = lambda x: x[1],
reverse = True)
decile: int = int(len(sorted_by_accruals) / self.quantile)
self.long_symbols = [x[0] for x in sorted_by_accruals[-decile:]]
self.short_symbols = [x[0] for x in sorted_by_accruals[:decile]]

return self.long_symbols + self.short_symbols

def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
for security in changes.AddedSecurities:
security.SetFeeModel(CustomFeeModel())

for security in changes.RemovedSecurities:
if security.Symbol in self.accrual_data:
del self.accrual_data[security.Symbol]

def OnData(self, slice: Slice) -> None:
if not self.selection_flag:
return
self.selection_flag = False

targets: List[PortfolioTarget] = []
for i, portfolio in enumerate([self.long_symbols, self.short_symbols]):
for symbol in portfolio:
if slice.ContainsKey(symbol) and slice[symbol] is not None:
targets.append(PortfolioTarget(symbol, ((-1) ** i) / len(portfolio)))

self.SetHoldings(targets, True)
self.long_symbols.clear()
self.short_symbols.clear()

def Selection(self) -> None:
if self.Time.month == self.rebalancing_month:
self.selection_flag = True

def CalculateAccruals(self, current_accrual_data, prev_accrual_data) -> float:
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
delta_tax = current_accrual_data.IncomeTaxPayable - prev_accrual_data.IncomeTaxPayable
dep = current_accrual_data.DepreciationAndAmortization
avg_total = (current_accrual_data.TotalAssets + prev_accrual_data.TotalAssets) / 2

return ((delta_assets - delta_cash) - (delta_liabilities - delta_debt - delta_tax) - dep) / avg_total

class AccrualsData():
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) -> None:
self.CurrentAssets: float = current_assets
self.CashAndCashEquivalents: float = cash_and_cash_equivalents
self.CurrentLiabilities: float = current_liabilities
self.CurrentDebt: float = current_debt
self.IncomeTaxPayable: float = income_tax_payable
self.DepreciationAndAmortization: float = depreciation_and_amortization
self.TotalAssets: float = total_assets

# 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"))```