Cash Operating Profitability Predicts Earnings Announcement Returns

1from AlgorithmImports import *
2import numpy as np
3from collections import deque
4from typing import Dict, List
5import data_tools
6from pandas.tseries.offsets import BDay
7from dateutil.relativedelta import relativedelta
8from numpy import isnan
9from functools import reduce
10# endregion
11
12class CashOperatingProfitabilityPredictsEarningsAnnouncementReturns(QCAlgorithm):
13
14def Initialize(self):
15self.SetStartDate(2008, 1, 1)
16self.SetCash(100000)
17
18self.quantile:int = 10
19self.leverage:int = 5
20self.long_num:int = 5
21self.short_num:int = 5
22self.holding_period:int = 5
23self.DR_period:int = 2
24
25self.financial_statement_names:List[str] = [
26    'FinancialStatements.IncomeStatement.TotalRevenue.TwelveMonths',
27    'FinancialStatements.IncomeStatement.CostOfRevenue.TwelveMonths',
28    'FinancialStatements.IncomeStatement.GeneralAndAdministrativeExpense.TwelveMonths',
29    'FinancialStatements.IncomeStatement.ResearchAndDevelopment.TwelveMonths',
30    'FinancialStatements.CashFlowStatement.ChangeInReceivables.TwelveMonths',
31    'FinancialStatements.CashFlowStatement.ChangeInInventory.TwelveMonths',
32    'FinancialStatements.CashFlowStatement.ChangeInPrepaidAssets.TwelveMonths',
33    'FinancialStatements.BalanceSheet.CurrentDeferredRevenue.TwelveMonths',
34    'FinancialStatements.CashFlowStatement.ChangeInAccountPayable.TwelveMonths',
35    'FinancialStatements.CashFlowStatement.ChangeInAccruedExpense.TwelveMonths',
36]
37
38self.long:List[Symbol] = []
39self.short:List[Symbol] = []
40self.DR_data:Dict[Symbol, float] = {}  
41
42self.earnings_data:Dict[datetime.date, List[str]] = {} 
43self.tickers:Set(str) = set()
44
45self.first_date:Union[datetime.date, None] = None
46
47earnings_data:str = self.Download('data.quantpedia.com/backtesting_data/economic/earnings_dates_eps.json')
48earnings_data_json:list[dict] = json.loads(earnings_data)
49
50for obj in earnings_data_json:
51    date:datetime.date = datetime.strptime(obj['date'], '%Y-%m-%d').date()
52
53    self.earnings_data[date] = []
54
55    if not self.first_date: self.first_date = date
56    
57    for stock_data in obj['stocks']:
58        ticker:str = stock_data['ticker']
59
60        self.tickers.add(ticker)
61        self.earnings_data[date].append(ticker)
62
63# 5 equally weighted brackets for traded symbols. - 5 symbols long, 5 symbols short, 5 days of holding
64self.trade_manager:TradeManager = data_tools.TradeManager(self, self.long_num, self.short_num, self.holding_period)
65
66self.symbol:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
67
68self.selection_month:int = 4
69self.selection_flag:bool = False
70self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
71self.UniverseSettings.Resolution = Resolution.Daily
72self.AddUniverse(self.FundamentalSelectionFunction)
73self.Schedule.On(self.DateRules.MonthStart(self.symbol), self.TimeRules.AfterMarketOpen(self.symbol), self.Selection)
74
75def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
76for security in changes.AddedSecurities:
77    security.SetFeeModel(data_tools.CustomFeeModel())
78    security.SetLeverage(self.leverage)
79
80def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]: 
81# yearly selection
82if self.selection_flag:
83    self.selection_flag = False
84
85    COP_stocks:Dict[Fundamental, float] = {}
86    selected:List[Fundamental] = [x for x in fundamental if x.Symbol.Value in self.tickers and \
87                all((not isnan(self.rgetattr(x, statement_name)) and self.rgetattr(x, statement_name) != 0) for statement_name in self.financial_statement_names)]
88
89    for stock in selected:
90        symbol:Symbol = stock.Symbol
91
92        if symbol not in self.DR_data:
93            self.DR_data[symbol] = data_tools.DRChangeManager(self.DR_period)
94        self.DR_data[symbol].update_data(stock.FinancialStatements.BalanceSheet.CurrentDeferredRevenue.TwelveMonths)
95        
96        if not self.DR_data[symbol].is_ready():
97            continue
98        
99        DR_change = self.DR_data[symbol].get_DR_change()
100        sales, COGS, SG_A, R_D, REC_change, INV_change, XPP_change, DR_change, AP_change, XACC_change = [self.rgetattr(stock, statement_name) for statement_name in self.financial_statement_names]
101
102        # calculate COP value on stocks
103        if stock not in COP_stocks:
104            COP_stocks[symbol] = sales - COGS - (SG_A - R_D) - REC_change - INV_change - XPP_change + DR_change + AP_change + XACC_change            
105
106    # sort and divide to quantiles
107    if len(COP_stocks) >= self.quantile:
108        sorted_COP = sorted(COP_stocks, key=COP_stocks.get, reverse=True)
109        quantile:int = int(len(sorted_COP) / self.quantile)
110        self.long = sorted_COP[:quantile]
111        self.short = sorted_COP[-quantile:]       
112
113return self.long + self.short
114
115def OnData(self, data: Slice) -> None:
116# liquidate opened symbols after five days
117self.trade_manager.TryLiquidate()
118
119date_to_lookup:datetime.date = (self.Time + BDay(2)).date()
120
121# if there is no earnings data yet
122if date_to_lookup < self.first_date:
123    # clear long set
124    self.long.clear()
125    self.short.clear()
126
127# open new trades
128symbols_to_delete = []
129if date_to_lookup in self.earnings_data:
130    for symbol in self.long:
131        # Next day is earnings day for the symbol.
132        if symbol.Value in self.earnings_data[date_to_lookup] and symbol in data and data[symbol]:
133            self.trade_manager.Add(symbol, True)
134
135    for symbol in self.short:
136        # Next day is earnings day for the symbol.
137        if symbol.Value in self.earnings_data[date_to_lookup] and symbol in data and data[symbol]:
138            self.trade_manager.Add(symbol, False)
139
140def Selection(self) -> None:
141if self.Time.month == self.selection_month:
142    self.selection_flag = True
143    
144    self.long.clear()
145    self.short.clear()
146
147# https://gist.github.com/wonderbeyond/d293e7a2af1de4873f2d757edd580288
148def rgetattr(self, obj, attr, *args):
149def _getattr(obj, attr):
150    return getattr(obj, attr, *args)
151return reduce(_getattr, [obj] + attr.split('.'))