Improved Post-Earnings Announcement Drift with NLP Analysis

1from AlgorithmImports import *
2from dateutil.relativedelta import relativedelta
3from itertools import combinations
4#endregion
5
6class ImprovedPostEarningsAnnouncementDriftwithNLPAnalysis(QCAlgorithm):
7
8def Initialize(self):
9self.SetStartDate(2012, 1, 1)
10self.SetCash(100000)
11
12self.quantile:int = 3
13self.period:int = 12
14self.week_holding_period:int = 4
15self.managed_queue:list[RebalanceQueueItem] = [] # tranched queue
16
17self.ticker_universe:list = []
18self.weight:dict[Symbol, float] = {}
19self.sentiment_by_transcript_date:dict[datetime.datetime, dict[str, float]] = {}
20self.sentiment_history:dict[str, dict[datetime.datetime, float]] = {}
21
22# management discussion sentiment data
23sentiment_file = self.Download('data.quantpedia.com/backtesting_data/index/management_discussion_sentiment.csv')
24sentiment_file_lines:list[str] = sentiment_file.split('\r\n')
25
26# parse sentiment data
27for i, line in enumerate(sentiment_file_lines):
28    # transcript_date;A;ADSK;ANF;...
29    # 2011-10-12;;;;;;;;;;;;;;;0.1625;0.4507;0.3004;...
30    if i == 0:
31        # get available tickers
32        self.ticker_universe = line.split(';')[1:]
33    else:
34        line_split:list[str] = line.split(';')
35        
36        # store sentiment data by transcript date and by ticker
37        date_str:str = line_split[0]
38        if date_str != '':
39            transcript_date:datetime.datetime = datetime.strptime(line_split[0], "%Y-%m-%d") + relativedelta(months=1)
40            self.sentiment_by_transcript_date[transcript_date] = {}
41
42            for j, sentiment in enumerate(line_split[1:]):
43                if sentiment != '':
44                    ticker:str = self.ticker_universe[j]
45                    if ticker not in self.sentiment_history:
46                        self.sentiment_history[ticker] = {}
47                    
48                    sentiment:float = float(sentiment)
49                    
50                    self.sentiment_by_transcript_date[transcript_date][ticker] = sentiment
51                    self.sentiment_history[ticker][transcript_date] = sentiment
52
53# universe setup
54self.coarse_count:int = 500
55self.selection_flag:bool = False
56self.market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
57self.UniverseSettings.Resolution = Resolution.Daily
58self.AddUniverse(self.CoarseSelectionFunction)
59self.Schedule.On(self.DateRules.WeekStart(self.market), self.TimeRules.BeforeMarketClose(self.market), self.Selection)
60
61def OnSecuritiesChanged(self, changes):
62for security in changes.AddedSecurities:
63    security.SetFeeModel(CustomFeeModel())
64    security.SetLeverage(5)
65
66def CoarseSelectionFunction(self, coarse):
67if not self.selection_flag:
68    return Universe.Unchanged
69
70# coarse = [x for x in coarse if x.Symbol.Value in self.ticker_universe]
71coarse = sorted([x for x in coarse if x.HasFundamentalData and x.Price >= 5 and x.Symbol.Value in self.ticker_universe],
72    key=lambda x: x.DollarVolume, reverse=True)[:self.coarse_count]
73
74symbols_by_ticker:list[Symbol] = {x.Symbol.Value : x.Symbol for x in coarse}
75prices_by_symbol:list[Symbol] = {x.Symbol : x.AdjustedPrice for x in coarse}
76
77surprise:dict[Symbol, float] = {}
78
79# there is some sentiment data available
80lookup_date:datetime.datetime = self.Time - timedelta(days=1)
81if lookup_date in self.sentiment_by_transcript_date:
82    for ticker, sentiment in self.sentiment_by_transcript_date[lookup_date].items():
83        if ticker in symbols_by_ticker:
84            # get previous ticker history up utill previous day
85            previous_transcript_dates:list = [x for x in self.sentiment_history[ticker].items() if x[0] < lookup_date]
86            if len(previous_transcript_dates) >= self.period:
87                trimmed_history:list = previous_transcript_dates[-self.period:]
88                avg_sentiment:float = np.mean([x[1] for x in trimmed_history])
89                latest_sentiment:float = self.sentiment_by_transcript_date[lookup_date][ticker]
90                surprise[symbols_by_ticker[ticker]] = latest_sentiment - avg_sentiment
91
92long:list[Symbol] = []
93short:list[Symbol] = []
94
95# sort by surprise
96if len(surprise) >= self.quantile:
97    sorted_by_surprise:list = sorted(surprise.items(), key=lambda x: x[1], reverse=True)
98    quantile:int = int(len(sorted_by_surprise) / self.quantile)
99    long = [x[0] for x in sorted_by_surprise[:quantile]]
100    short = [x[0] for x in sorted_by_surprise[-quantile:]]
101
102    long_w = self.Portfolio.TotalPortfolioValue / self.week_holding_period / len(long)
103    long_symbol_q = [(x, np.floor(long_w / prices_by_symbol[x])) for x in long]
104
105    short_w = self.Portfolio.TotalPortfolioValue / self.week_holding_period / len(short)
106    short_symbol_q = [(x, -np.floor(short_w / prices_by_symbol[x])) for x in short]
107    
108    self.managed_queue.append(RebalanceQueueItem(long_symbol_q + short_symbol_q))
109
110# return both - held symbols and new symbols
111symbols_to_return:set[Symbol] = set([x.Key for x in self.Portfolio if x.Value.Invested])
112for symbol in long+short:
113    symbols_to_return.add(symbol)
114
115return list(symbols_to_return)
116
117def OnData(self, data):
118if not self.selection_flag:
119    return
120self.selection_flag = False
121
122# trade execution
123remove_item = None
124
125# rebalance portfolio
126for item in self.managed_queue:
127    if item.holding_period == self.week_holding_period:
128        for symbol, quantity in item.symbol_q:
129            self.MarketOrder(symbol, -quantity)
130        remove_item = item
131        
132    elif item.holding_period == 0:
133        open_symbol_q = []
134        
135        for symbol, quantity in item.symbol_q:
136            if symbol in data and data[symbol]:
137                if quantity != 0:
138                    self.MarketOrder(symbol, quantity)
139                    open_symbol_q.append((symbol, quantity))
140                    
141        # only opened orders will be closed
142        item.symbol_q = open_symbol_q
143        
144    item.holding_period += 1
145    
146if remove_item:
147    self.managed_queue.remove(remove_item)
148
149def Selection(self):
150# weekly rebalance
151self.selection_flag = True
152
153class RebalanceQueueItem():
154def __init__(self, symbol_q):
155# symbol/quantity collections
156self.symbol_q = symbol_q
157self.holding_period = 0
158
159def get_symbols(self) -> list:
160return [x[0] for x in self.symbol_q]
161
162# custom fee model
163class CustomFeeModel():
164def GetOrderFee(self, parameters):
165fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
166return OrderFee(CashAmount(fee, "USD"))