Announcement-Adjusted Industry-Relative Reversal Factor

1from AlgorithmImports import *
2from dateutil.relativedelta import relativedelta
3from pandas.tseries.offsets import BDay
4import numpy as np
5from typing import Dict, List
6# endregion
7
8class AnnouncementAdjustedIndustryRelativeReversalFactor(QCAlgorithm):
9
10def Initialize(self):
11self.SetStartDate(2010, 1, 1)
12self.SetCash(100000)
13
14self.market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
15self.ticker_to_ignore:List[str] = ['GME']
16
17self.leverage:int = 3
18self.quantile:int = 5
19self.period:int = 31
20self.fundamental_count:int = 3000
21
22self.data:Dict[Symbol, float] = {}
23self.earnings_dates:Dict[datetime.date, List[str]] = {}
24self.long:List[Symbol] = []
25self.short:List[Symbol] = []
26
27earnings_data:str = self.Download('data.quantpedia.com/backtesting_data/economic/earnings_dates_eps.json')
28earnings_data_json:list[dict] = json.loads(earnings_data)
29
30for obj in earnings_data_json:
31    date:datetime.date = (datetime.strptime(obj['date'], '%Y-%m-%d') + BDay(1)).date()
32
33    if date not in self.earnings_dates:
34        self.earnings_dates[date] = []
35    
36    for stock_data in obj['stocks']:
37        ticker:str = stock_data['ticker']
38
39        self.earnings_dates[date].append(ticker)
40
41self.selection_flag:bool = False
42self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
43self.UniverseSettings.Resolution = Resolution.Daily
44self.AddUniverse(self.FundamentalSelectionFunction)
45self.Schedule.On(self.DateRules.MonthStart(self.market), self.TimeRules.AfterMarketOpen(self.market), self.Selection)
46
47def OnSecuritiesChanged(self, changes:SecurityChanges) -> None:
48for security in changes.AddedSecurities:
49    security.SetFeeModel(CustomFeeModel())
50    security.SetLeverage(self.leverage)
51
52def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
53# store daily prices
54for stock in fundamental:
55    symbol:Symbol = stock.Symbol
56
57    if symbol in self.data:
58        self.data[symbol].update_daily_return(self.Time, stock.AdjustedPrice)
59
60# selection on month start
61if not self.selection_flag:
62    return Universe.Unchanged
63
64selected:List[Fundamental] = [x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and x.Symbol.Value not in self.ticker_to_ignore \
65                        and x.MarketCap != 0 and not np.isnan(x.AssetClassification.MorningstarSectorCode) and x.AssetClassification.MorningstarSectorCode != 0 and \
66                        (x.SecurityReference.ExchangeId == 'NYS')]
67
68if len(selected) > self.fundamental_count:
69    selected = sorted(selected, key=lambda x: x.MarketCap, reverse=True)[:self.fundamental_count]
70
71selected:Dict[str, Fundamental] = {x.Symbol.Value: x for x in selected}
72
73# sort stocks on industry numbers and price warmup
74grouped_industries:Dict[MorningstarIndustryGroupCode, List[Symbol]] = {}
75
76for ticker, stock in selected.items():
77    symbol:Symbol = stock.Symbol
78
79    industry_sector_code:int = stock.AssetClassification.MorningstarSectorCode
80
81    if not industry_sector_code in grouped_industries:
82        grouped_industries[industry_sector_code] = []
83    grouped_industries[industry_sector_code].append(symbol)
84
85    if symbol in self.data:
86        continue
87       
88    self.data[symbol] = SymbolData()
89    history:DataFrame = self.History(symbol, self.period, Resolution.Daily)
90    if history.empty:
91        self.Log(f"Not enough data for {symbol} yet.")
92        continue
93    closes:pd.Series = history.loc[symbol].close
94    for time, close in closes.items():
95        self.data[symbol].update_daily_return(time, close)
96
97irrx:Dict[Symbol, float] = {}
98
99# check earnings annoucement days
100for date, ticker_list in self.earnings_dates.items():
101    if date >= self.Time.date() - relativedelta(months=1) and date < self.Time.date():
102        for ticker in ticker_list:
103            if ticker in selected:
104                symbol:Symbol = selected[ticker].Symbol
105
106                if self.data[symbol].is_ready() and all([self.data[x].is_ready() for x in grouped_industries[selected[ticker].AssetClassification.MorningstarSectorCode]]):
107                    symbol_announcement_returns:float = self.data[symbol].get_target_date_return(date)
108                    industry_announcement_returns:float = np.mean([self.data[x].get_target_date_return(date) for x in grouped_industries[selected[ticker].AssetClassification.MorningstarSectorCode]])
109
110                    industry_returns:float = np.mean([self.data[x].get_monthly_return() for x in grouped_industries[selected[ticker].AssetClassification.MorningstarSectorCode]])
111                    monthly_excess_return:float = self.data[symbol].get_monthly_return() - industry_returns
112
113                    irrx_ = monthly_excess_return - (symbol_announcement_returns - industry_announcement_returns)
114                    if irrx_ != sys.float_info.min:
115                        irrx[symbol] = irrx_
116
117for symbol, symbol_data in self.data.items():
118    symbol_data.reset_daily_returns()
119
120if len(irrx) >= self.quantile:
121    sorted_irrx:List[Symbol] = sorted(irrx, key=irrx.get)
122    quantile:int = len(irrx) // self.quantile
123    self.long = sorted_irrx[:quantile]
124    self.short = sorted_irrx[-quantile:]
125
126return self.long + self.short
127
128def OnData(self, data: Slice) -> None:
129# monthly rebalance
130if not self.selection_flag:
131    return
132self.selection_flag = False
133
134# order execution
135targets:List[PortfolioTarget] = []
136for i, portfolio in enumerate([self.long, self.short]):
137    for symbol in portfolio:
138        if symbol in data and data[symbol]:
139            targets.append(PortfolioTarget(symbol, ((-1) ** i) / len(portfolio)))
140
141self.SetHoldings(targets, True)
142
143self.long.clear()
144self.short.clear()
145
146def Selection(self) -> None:
147self.selection_flag = True
148
149# Custom fee model
150class CustomFeeModel(FeeModel):
151def GetOrderFee(self, parameters):
152fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
153return OrderFee(CashAmount(fee, "USD"))
154
155class SymbolData():
156def __init__(self) -> None:
157self._last_price:float|None = None
158self._daily_return:List[Tuple[datetime.date, float]] = []
159
160def update_daily_return(self, time:datetime, price:float) -> None:
161if self._last_price is not None:
162    daily_return:float = (price - self._last_price) / self._last_price
163    self._daily_return.append((time.date(), daily_return))
164
165self._last_price = price
166
167def reset_daily_returns(self) -> None:
168self._daily_return.clear()
169
170def get_monthly_return(self) -> float:
171returns:List[float] = list(map(lambda x: x[1], self._daily_return))
172return sum(returns)
173
174def get_target_date_return(self, date:datetime.date) -> float:
175#[i[0] for i in self._daily_return]:
176if date in list(map(lambda x: x[0], self._daily_return)):
177    for i in range(len(self._daily_return) - 1):
178        current_date, _ = self._daily_return[i]
179        if current_date == date:
180            return self._daily_return[i-1][1] + self._daily_return[i][1] + self._daily_return[i+1][1] 
181else:
182    return sys.float_info.min
183
184def is_ready(self) -> bool:
185return self._last_price is not None and len(self._daily_return) != 0