from AlgorithmImports import *
class EarningsQualityFactor(QCAlgorithm):
def initialize(self):
self.set_start_date(self.end_date - timedelta(5*365))
self.set_cash(1_000_000)
self.settings.minimum_order_margin_portfolio_percentage = 0
self.settings.seed_initial_prices = True
self.set_warmup(timedelta(2 * 365 + 45))
self._previous_factors = {}
self._date_rule = self.date_rules.month_end('SPY')
self._long = []
self._short = []
self.universe_settings.resolution = Resolution.DAILY
# Add universe of Equities.
self.universe_settings.schedule.on(self._date_rule)
self._universe = self.add_universe(self._select)
def on_warmup_finished(self):
# Add a Scheduled event to rebalance the portfolio monthly.
time_rule = self.time_rules.at(8, 0)
self.schedule.on(self._date_rule, time_rule, self._rebalance)
# Rebalance the portfolio today too.
if self.live_mode:
self._rebalance()
else:
self.schedule.on(self.date_rules.today, time_rule, lambda: self._rebalance(True))
def _select(self, fundamentals):
if self.time.month != 6:
return Universe.UNCHANGED
# Filter non-financial stocks with valid fundamental data and accrual data.
filtered = [
f for f in fundamentals
if (f.has_fundamental_data and
f.company_reference.industry_template_code != "B" and
f.financial_statements.balance_sheet.current_assets.value and
f.financial_statements.balance_sheet.cash_and_cash_equivalents.value and
f.financial_statements.balance_sheet.current_liabilities.value and
f.financial_statements.balance_sheet.current_debt.value and
f.financial_statements.balance_sheet.income_tax_payable.value and
f.financial_statements.income_statement.depreciation_and_amortization.value)
]
factors_by_symbol = {f.symbol : AccuralData(f) for f in filtered}
if not self._previous_factors:
self._previous_factors = factors_by_symbol
return Universe.UNCHANGED
# Calculate accruals for stocks we have previous data for.
accrual_by_symbol = {
f.symbol: self._calc_accrual(self._previous_factors[f.symbol], AccuralData(f))
for f in filtered if f.symbol in self._previous_factors
}
# Filter stocks with complete factor data.
filtered = [f for f in filtered if f.symbol in accrual_by_symbol and all(
not np.isnan(val) and val != 0 for val in [
f.financial_statements.cash_flow_statement.operating_cash_flow.value,
f.earning_reports.basic_eps.value,
f.earning_reports.basic_average_shares.value,
f.operation_ratios.debt_to_assets.value,
f.operation_ratios.roe.value
])]
if not filtered:
return []
# Calculate composite score using ranks (lower is better for accrual/debt, higher for cfa/roe)
for f in filtered:
# Cash flow to assets.
f.cfa = (
f.financial_statements.cash_flow_statement.operating_cash_flow.value
/ (f.earning_reports.basic_eps.value * f.earning_reports.basic_average_shares.value)
)
# Debt to assets.
f.da = f.operation_ratios.debt_to_assets.value
# Return on equity.
f.roe = f.operation_ratios.roe.value
# Sort stocks by four factors respectively.
sorted_by_accrual = sorted(filtered, key=lambda x: accrual_by_symbol[x.symbol], reverse=True) # High score with low accrual.
sorted_by_cfa = sorted(filtered, key=lambda x: x.cfa) # High score with high CFA.
sorted_by_da = sorted(filtered, key=lambda x: x.da, reverse=True) # High score with low leverage.
sorted_by_roe = sorted(filtered, key=lambda x: x.roe) # High score with high ROE.
# Create dict to save the score for each stock.
score_dict = {}
# Assign a score to each stock according to their rank with different factors.
for i,obj in enumerate(sorted_by_accrual):
score_accrual = i
score_cfa = sorted_by_cfa.index(obj)
score_da = sorted_by_da.index(obj)
score_roe = sorted_by_roe.index(obj)
score = score_accrual + score_cfa + score_da + score_roe
score_dict[obj.symbol] = score
sorted_by_score = sorted(score_dict, key=lambda x: score_dict[x])
# Long stocks with the top score (> 10%) and short stocks with the bottom score (< 10%).
self._long = sorted_by_score[-int(0.10 * len(sorted_by_score)):]
self._short = sorted_by_score[:int(0.10 * len(sorted_by_score))]
self._previous_factors = factors_by_symbol
return self._long + self._short
def _rebalance(self, skip_checks=False):
# Rebalance portfolio by allocating equal weight to long and short positions.
if not skip_checks and not (self._long and self._short and self.time.month == 6):
return
long = [s for s in self._long if self.securities[s].price]
short = [s for s in self._short if self.securities[s].price]
targets = [PortfolioTarget(s, 0.5 / len(long)) for s in long]
targets += [PortfolioTarget(s, -0.5 / len(short)) for s in short]
self.set_holdings(targets, True)
def _calc_accrual(self, prev, curr):
# Calculates the balance sheet accruals and adds the property to the fundamental object.
delta_assets = curr.current_assets - prev.current_assets
delta_cash = curr.cash_equivalent - prev.cash_equivalent
delta_liabilities = curr.liabilities - prev.liabilities
delta_debt = curr.current_debt - prev.current_debt
delta_tax = curr.tax_payable - prev.tax_payable
dep = curr.depreciation_amortization
avg_total = (curr.total_assets + prev.total_assets) / 2
# Accounts for the size difference.
return ((delta_assets - delta_cash) - (delta_liabilities - delta_debt - delta_tax) - dep) / avg_total
class AccuralData:
def __init__(self, fundamental):
bs = fundamental.financial_statements.balance_sheet
self.total_assets = bs.total_assets.value
self.cash_equivalent = bs.cash_and_cash_equivalents.value
self.current_debt = bs.current_debt.value
self.tax_payable = bs.income_tax_payable.value
self.liabilities = bs.current_liabilities.value
self.current_assets = bs.current_assets.value
self.depreciation_amortization = fundamental.financial_statements.income_statement.depreciation_and_amortization.value
