Instititutional Ownership Effect During Earnings Announcements

1import data_tools
2from AlgorithmImports import *
3import numpy as np
4from pandas.tseries.offsets import BDay
5class InstititutionalOwnershipEffectDuringEarningsAnnouncements(QCAlgorithm):
6
7def Initialize(self) -> None:
8self.SetStartDate(2009, 1, 1) # earnings dates starts at 2010
9self.SetCash(100_000)
10self.period: int = 21
11self.lookup_period: int = 2
12self.holding_period: int = 2
13self.min_share_price: int = 5
14self.leverage: int = 5
15self.quantile: int = 5
16self.total_long_num: int = 15
17self.total_short_num: int = 15
18
19self.long: Set(Symbol) = set()
20self.short: Set(Symbol) = set()
21self.data: Dict[Symbol, data_tools.SymbolData] = {}
22self.earnings_data: Dict[datetime.date, list[str]] = {}
23
24self.first_date: Union[None, datetime.date] = None
25earnings_set: Set(str) = set()
26earnings_data: str = self.Download('data.quantpedia.com/backtesting_data/economic/earnings_dates_eps.json')
27earnings_data_json: List[dict] = json.loads(earnings_data)
28
29for obj in earnings_data_json:
30    date: datetime.date = datetime.strptime(obj['date'], "%Y-%m-%d").date()
31    if not self.first_date: self.first_date = date
32    self.earnings_data[date] = []
33    
34    for stock_data in obj['stocks']:
35        ticker: str = stock_data['ticker']
36        self.earnings_data[date].append(ticker)
37        earnings_set.add(ticker)
38
39self.tickers: List[str] = list(earnings_set)
40market: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
41# equally weighted brackets for traded symbols. - n symbols long, m symbols short, 2 days of holding
42self.trade_manager: data_tools.TradeManager = data_tools.TradeManager(
43    self, self.total_long_num, self.total_short_num, self.holding_period)
44
45self.fundamental_count: int = 3000
46self.fundamental_sorting_key = lambda x: x.MarketCap
47self.selection_flag: bool = False
48self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
49self.settings.daily_precise_end_time = False
50self.UniverseSettings.Resolution = Resolution.Daily
51self.AddUniverse(self.FundamentalSelectionFunction)
52self.Schedule.On(self.DateRules.MonthStart(market), self.TimeRules.AfterMarketOpen(market), self.Selection)
53def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
54for security in changes.AddedSecurities:
55    security.SetFeeModel(data_tools.CustomFeeModel())
56    security.SetLeverage(self.leverage)
57    
58def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
59# update the rolling window every day
60for stock in fundamental:
61    symbol: Symbol = stock.Symbol
62    if symbol in self.data:
63        self.data[symbol].update(stock.Volume)
64if not self.selection_flag:
65    return Universe.Unchanged
66self.selection_flag = False
67selected: List[Fundamental] = [
68    x for x in fundamental if x.Symbol.Value in self.tickers \
69    and x.HasFundamentalData and x.MarketCap != 0 and x.Market == 'usa' and x.Price > self.min_share_price
70]
71if len(selected) > self.fundamental_count:
72    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
73warmed_up_symbols:List[Symbol] = []
74for stock in selected:
75    symbol: Symbol = stock.Symbol
76    if symbol not in self.data:
77        self.data[symbol] = data_tools.SymbolData(symbol, self.period)
78        history: DataFrame = self.History(symbol, self.period, Resolution.Daily)
79        if history.empty:
80            continue
81        if not hasattr(history.loc[symbol], 'volume'):
82            continue
83        volumes: Series = history.loc[symbol].volume
84        for _, volume in volumes.items():
85            self.data[symbol].update(volume)
86    if self.data[symbol].is_ready():
87        warmed_up_symbols.append(symbol)
88
89if len(warmed_up_symbols) < self.quantile:
90    return Universe.Unchanged
91quantile: int = int(len(warmed_up_symbols) / self.quantile)
92lowest_market_caps: List[Symbol] = [x for x in warmed_up_symbols[-quantile:]]
93
94volumes: Dict[Symbol, float] = { x : self.data[x].sum_volumes() for x in warmed_up_symbols}
95
96quantile: int = int(len(volumes) / self.quantile)
97highest_volumes: List[Symbol] = [x[0] for x in sorted(volumes.items(), key=lambda item: item[1])][-quantile:]
98
99self.long = set(symbol for symbol in lowest_market_caps if symbol in highest_volumes)
100return list(self.long)
101def OnData(self, data: Slice) -> None:
102# liquidate opened symbols after self.holding_period days.
103self.trade_manager.TryLiquidate()
104
105# long two days before earnings annoucement
106date_to_lookup_long: datetime.date = (self.Time + BDay(self.lookup_period)).date()
107# short two days after earnings annoucement
108date_to_lookup_short: datetime.date = (self.Time - BDay(self.lookup_period)).date()
109
110if date_to_lookup_long < self.first_date:
111    self.long.clear()
112# open new trades
113symbols_to_delete: List[Symbol] = []
114if date_to_lookup_long in self.earnings_data:
115    for symbol in self.long:
116        if symbol.Value in self.earnings_data[date_to_lookup_long] and symbol in data and data[symbol]:
117            self.trade_manager.Add(symbol, True)
118            symbols_to_delete.append(symbol)
119
120# delete already traded symbols and add them to short portfolio
121for symbol in symbols_to_delete:
122    self.long.remove(symbol)
123    self.short.add(symbol)
124    
125symbols_to_delete.clear()
126if date_to_lookup_short in self.earnings_data:
127    for symbol in self.short:
128        if symbol.Value in self.earnings_data[date_to_lookup_short] and symbol in data and data[symbol]:
129            self.trade_manager.Add(symbol, False)
130            symbols_to_delete.append(symbol)
131            
132for symbol in symbols_to_delete:
133    self.short.remove(symbol)
134
135def Selection(self) -> None:
136self.selection_flag = True