Macroeconomic Announcement Beta Strategy

1from AlgorithmImports import *
2import numpy as np
3from pandas.tseries.offsets import BDay
4from scipy import stats
5from data_tools import CustomFeeModel, SymbolData
6from typing import List, Dict
7from pandas.core.frame import DataFrame
8from pandas.core.series import Series
9#endregion
10class MacroeconomicAnnouncementBeta(QCAlgorithm):
11def Initialize(self) -> None:
12self.SetStartDate(2000, 1, 1)
13self.SetCash(100000)
14
15self.leverage: int = 5
16self.quantile: int = 10
17self.period: int = 5 * 12
18self.daily_period: int = 21
19self.days_offset: int = 1
20self.data: Dict[Symbol, SymbolData] = {}
21self.selected_symbols: List[Symbol] = []
22self.market: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
23self.data[self.market] = SymbolData(self.period)
24self.WarmUpSymbolPrices(self.market)
25
26csv_string_file: str = self.Download('data.quantpedia.com/backtesting_data/economic/economic_announcements.csv')
27dates: List[str] = csv_string_file.split('\r\n')
28before_announcement_dates: List[datetime.date] = [(datetime.strptime(x, '%Y-%m-%d') - BDay(self.days_offset)).date() for x in dates]
29self.fundamental_count: int = 500
30self.fundamental_sorting_key = lambda x: x.DollarVolume
31self.sort_flag: bool = False
32self.selection_flag: bool = False
33self.settings.daily_precise_end_time = False
34self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
35self.UniverseSettings.Resolution = Resolution.Daily
36self.AddUniverse(self.FundamentalSelectionFunction)
37self.Schedule.On(self.DateRules.On(before_announcement_dates), self.TimeRules.AfterMarketOpen(self.market), self.DayBeforeAnnouncement)
38self.Schedule.On(self.DateRules.MonthEnd(self.market), self.TimeRules.AfterMarketOpen(self.market), self.Selection)
39def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
40for security in changes.AddedSecurities:
41    security.SetFeeModel(CustomFeeModel())
42    security.SetLeverage(self.leverage)
43def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
44# monthly selection
45if not self.selection_flag:
46    return Universe.Unchanged
47self.selection_flag = False
48
49selected: List[fundamental] = []
50self.selected_symbols.clear()
51for stock in fundamental:
52    symbol: Symbol = stock.Symbol
53    
54    if symbol in self.data and self.data[symbol].is_last_month_price_ready():
55        self.data[symbol].update_monthly_return(stock.AdjustedPrice)
56    if stock.HasFundamentalData:
57        selected.append(stock)
58
59if len(selected) > self.fundamental_count:
60    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
61
62for stock in selected:
63    symbol: Symbol = stock.Symbol
64    if symbol not in self.data:
65        self.data[symbol] = SymbolData(self.period)
66        self.WarmUpSymbolPrices(symbol)
67    if self.data[symbol].is_ready():
68        self.selected_symbols.append(symbol)
69return self.selected_symbols     
70
71def OnData(self, data: Slice) -> None:
72self.Liquidate()
73
74# sort one day before earnings annoucement
75if not self.sort_flag or len(self.selected_symbols) == 0:
76    return
77self.sort_flag = False
78
79# make sure there is n years of SPY monthly returns history
80if not self.data[self.market].is_ready():
81    return
82market_monthly_returns: List[float] = self.data[self.market].get_monthly_returns()
83
84beta: Dict[Symbol, float] = {}
85for symbol in self.selected_symbols:
86    if symbol in data and data[symbol]:
87        stock_monthly_returns: List[float] = self.data[symbol].get_monthly_returns()
88        
89        # linear regression - X = market returns, Y = stock returns
90        slope, intercept, r_value, p_value, std_err = stats.linregress(market_monthly_returns, stock_monthly_returns)
91        beta[symbol] = slope
92    
93# check if there are enough stocks for selection
94if len(beta) < self.quantile:
95    self.Liquidate()
96    return
97
98# beta sorting
99quantile: int = int(len(beta) / self.quantile)
100sorted_by_beta: List[Symbol] = [x[0] for x in sorted(beta.items(), key=lambda item: item[1])]
101long: List[Symbol] = sorted_by_beta[-quantile:]
102short: List[Symbol] = sorted_by_beta[:quantile]
103
104# trade execution
105targets: List[PortfolioTarget] = []
106for i, portfolio in enumerate([long, short]):
107    for symbol in portfolio:
108        if symbol in data and data[symbol]:
109            targets.append(PortfolioTarget(symbol, ((-1) ** i) / len(portfolio)))
110
111self.SetHoldings(targets, True)
112
113def WarmUpSymbolPrices(self, symbol: Symbol) -> None:
114history: DataFrame = self.History([symbol], self.daily_period * self.period, Resolution.Daily)
115if history.empty:
116    return
117
118closes: Series = history.loc[symbol].close
119closes_grouped: Series = closes.groupby(pd.Grouper(freq='M')).last()
120for close in closes_grouped:
121    self.data[symbol].update_monthly_return(close)
122def Selection(self) -> None:
123self.selection_flag = True
124
125def DayBeforeAnnouncement(self) -> None:
126self.sort_flag = True