Impact of Macro News on PEAD Strategy

1from AlgorithmImports import *
2import numpy as np
3from collections import deque
4from typing import List, Dict, Deque
5from numpy import isnan
6#endregion
7class ImpactofMacroNewsonPEADStrategy(QCAlgorithm):
8def Initialize(self) -> None:
9self.SetStartDate(2000, 1, 1)
10self.SetCash(100000)
11self.exchange_codes: List[str] = ['NYS', 'NAS', 'ASE']	
12self.period: int = 13
13self.quantile: int = 5
14self.leverage: int = 5
15self.threshold: int = 3
16self.min_share_price: int = 5
17# EPS quarterly data.
18self.eps_data: Dict[Symbol, Deque[List[float]]] = {} 
19self.long: List[Symbol] = []
20self.short: List[Symbol] = []
21
22symbol: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
23# Import macro dates.
24csv_string_file: str = self.Download('data.quantpedia.com/backtesting_data/economic/economic_announcements.csv')
25dates: List[str] = csv_string_file.split('\r\n')
26self.macro_dates: List[datetime.date] = [datetime.strptime(x, "%Y-%m-%d").date() for x in dates]
27self.fundamental_count: int = 3000
28self.fundamental_sorting_key = lambda x: x.MarketCap
29self.selection_flag: int = False
30self.settings.daily_precise_end_time = False
31self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
32self.UniverseSettings.Resolution = Resolution.Daily
33self.AddUniverse(self.FundamentalSelectionFunction)
34self.Schedule.On(self.DateRules.MonthStart(symbol), self.TimeRules.AfterMarketOpen(symbol), self.Selection)
35def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
36for security in changes.AddedSecurities:
37    security.SetFeeModel(CustomFeeModel())
38    security.SetLeverage(self.leverage)
39
40def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
41if not self.selection_flag:
42    return Universe.Unchanged
43
44selected: List[Fundamental] = [
45    x for x in fundamental 
46    if x.HasFundamentalData 
47    and x.Market == 'usa' 
48    and x.Price > self.min_share_price 
49    and x.SecurityReference.ExchangeId in self.exchange_codes
50    and not isnan(x.EarningReports.BasicEPS.ThreeMonths) and (x.EarningReports.BasicEPS.ThreeMonths != 0)
51]
52
53if len(selected) > self.fundamental_count:
54    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
55
56# Stocks with last month's earnings.
57last_month_date: datetime.date = self.Time - timedelta(self.Time.day)
58filtered_fundamental: List[Fundamental] = [x for x in selected if (x.EarningReports.FileDate.ThreeMonths.year == last_month_date.year and x.EarningReports.FileDate.ThreeMonths.month == last_month_date.month)]
59
60# earnings surprises data for stocks
61earnings_surprises: Dict[Symbol, List[float, datetime.date]] = {}
62
63for stock in filtered_fundamental:
64    symbol: Symbol = stock.Symbol
65    
66    # Store eps data.
67    if symbol not in self.eps_data:
68        self.eps_data[symbol] = deque(maxlen = self.period)
69    self.eps_data[symbol].append([stock.EarningReports.FileDate.ThreeMonths.date(), stock.EarningReports.BasicEPS.ThreeMonths])
70    if len(self.eps_data[symbol]) == self.eps_data[symbol].maxlen:
71        year_range: range = range(self.Time.year - 3, self.Time.year)
72        month_range: List[datetime.date] = [last_month_date.month - 1, last_month_date.month, last_month_date.month + 1]
73        
74        # Earnings 3 years back.
75        seasonal_eps_data: List[List[datetime.date, float]] = [
76            x for x in self.eps_data[symbol] 
77            if x[0].month in month_range
78            and x[0].year in year_range]
79        if len(seasonal_eps_data) != self.threshold:
80            continue
81        recent_eps_data: List[datetime.date, float] = self.eps_data[symbol][-1]
82        
83        # Make sure we have a consecutive seasonal data. Same months with one year difference.
84        year_diff: np.ndarray = np.diff([x[0].year for x in seasonal_eps_data])
85        if all(x == 1 for x in year_diff):
86            seasonal_eps: List[float] = [x[1] for x in seasonal_eps_data]
87            diff_values: np.ndarray = np.diff(seasonal_eps)
88            drift: float = np.average(diff_values)
89            
90            # earnings surprise calculation
91            last_earnings: float = seasonal_eps[-1]
92            expected_earnings: float = last_earnings + drift
93            actual_earnings: float = recent_eps_data[1]
94            # Store sue value with earnigns date
95            earnings_surprise: float = actual_earnings - expected_earnings
96            earnings_surprises[symbol] = [earnings_surprise, stock.EarningReports.FileDate.ThreeMonths.date()]
97
98# wait until earnings suprises are ready           
99if len(earnings_surprises) < self.quantile:
100    return Universe.Unchanged
101if self.Time.date() > self.macro_dates[-1]:
102    return Universe.Unchanged
103# sort by earnings suprises.
104quantile: int = int(len(earnings_surprises) / self.quantile) 
105sorted_by_earnings_surprise: List[Symbol] = [x[0] for x in sorted(earnings_surprises.items(), key=lambda item: item[1][0])]
106
107# select top quintile and bottom quintile based on earnings suprise sort
108top_quintile: List[Symbol] = sorted_by_earnings_surprise[-quantile:]
109bottom_quintile: List[Symbol] = sorted_by_earnings_surprise[:quantile]
110
111# long stocks, which are in top quintile by earnings suprise sort and have non-macro date
112self.long = [x for x in top_quintile if earnings_surprises[x][1] not in self.macro_dates]
113# short stocks, which are in bottom quintile by earnings suprise sort and have non-macro date
114self.short = [x for x in bottom_quintile if earnings_surprises[x][1] not in self.macro_dates]
115
116return self.long + self.short
117def OnData(self, data: Slice) -> None:
118if not self.selection_flag:
119    return
120self.selection_flag = False
121    
122# Trade execution.
123targets: List[PortfolioTarget] = []
124for i, portfolio in enumerate([self.long, self.short]):
125    for symbol in portfolio:
126        if symbol in data and data[symbol]:
127            targets.append(PortfolioTarget(symbol, ((-1) ** i) / len(portfolio)))
128
129self.SetHoldings(targets, True)
130self.long.clear()
131self.short.clear()
132
133def Selection(self) -> None:
134self.selection_flag = True
135# Custom fee model.
136class CustomFeeModel(FeeModel):
137def GetOrderFee(self, parameters: OrderFeeParameters) -> OrderFee:
138fee: float = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
139return OrderFee(CashAmount(fee, "USD"))