# 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.AddUniverse(self.FundamentalFunction)
        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
        
        # Trade execution.
        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"))