Macroeconomic Announcement Beta Reversal

1from AlgorithmImports import *
2import numpy as np
3from pandas.tseries.offsets import BDay
4from scipy import stats
5from typing import List, Dict
6from pandas.core.frame import DataFrame
7from pandas.core.series import Series
8#endregion
9class MacroeconomicAnnouncementBetaReversal(QCAlgorithm):
10def Initialize(self) -> None:
11self.SetStartDate(2000, 1, 1)
12self.SetCash(100000)
13
14self.market: Symbol = self.AddEquity("SPY", Resolution.Daily).Symbol
15self.period: int = 5 * 12
16self.daily_period: int = 21
17self.quantile: int = 10
18self.leverage: int = 5
19
20self.data: Dict[Symbol, SymbolData] = {}
21self.selected_symbols: List[Symbol] = []
22self.long: List[Symbol] = []
23self.short: List[Symbol] = []
24csv_string_file: str = self.Download('data.quantpedia.com/backtesting_data/economic/economic_announcements.csv')
25dates: List[str] = csv_string_file.split('\r\n')
26announcement_dates: List[datetime.date] = [datetime.strptime(x, "%Y-%m-%d") for x in dates]
27sort_dates: List[datetime.date] = [(x + BDay(1)).date() for x in announcement_dates]
28liquidation_dates: List[datetime.date] = [(x + BDay(2)).date() for x in announcement_dates]
29
30self.fundamental_count: int = 500
31self.fundamental_sorting_key = lambda x: x.DollarVolume
32self.selection_flag: bool = False
33self.rebalance_flag: bool = False
34self.settings.daily_precise_end_time = False
35self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
36self.UniverseSettings.Resolution = Resolution.Daily
37self.AddUniverse(self.FundamentalSelectionFunction)
38self.Schedule.On(self.DateRules.MonthEnd(self.market), self.TimeRules.At(0,0), self.Selection)
39self.Schedule.On(self.DateRules.On(sort_dates), self.TimeRules.At(0,0), self.Rebalance)
40self.Schedule.On(self.DateRules.On(liquidation_dates), self.TimeRules.At(0,0), self.Liquidation)
41def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
42for security in changes.AddedSecurities:
43    security.SetFeeModel(CustomFeeModel())
44    security.SetLeverage(self.leverage)
45def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
46# monthly selection
47if not self.selection_flag:
48    return Universe.Unchanged
49self.selection_flag = False
50
51# calculate monthly return
52for stock in fundamental:
53    symbol = stock.Symbol
54    
55    # check if current stock have last month price
56    if symbol in self.data and self.data[symbol].last_month_price:
57        self.data[symbol].update_monthly_return(stock.AdjustedPrice)
58
59selected: List[Fundamental] = [x for x in fundamental if x.HasFundamentalData]
60if len(selected) > self.fundamental_count:
61    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
62
63# Store monthly return for every stock selected this month.
64for stock in selected + [self.market]:
65    if stock == self.market:
66        symbol = stock
67    else:
68        symbol: Symbol = stock.Symbol
69    if symbol in self.data:
70        continue
71    
72    self.data[symbol] = SymbolData(self.period)
73    history: DataFrame = self.History([symbol], self.daily_period * self.period, Resolution.Daily)
74    
75    if history.empty:
76        self.Log(f"Not enough data for {symbol} yet")
77        continue
78    
79    closes: Series = history.loc[symbol].close
80    closes_grouped: Series = closes.groupby(pd.Grouper(freq='M')).last()
81    for close in closes_grouped:
82        self.data[symbol].update_monthly_return(close)
83
84# get stocks, which have ready monthly returns
85self.selected_symbols = [x.Symbol for x in selected if self.data[x.Symbol].is_ready() and x.Symbol != self.market]        
86
87return self.selected_symbols
88
89def OnData(self, data: Slice) -> None:
90if not self.rebalance_flag:
91    return
92self.rebalance_flag = False
93# there has to be at least one selected symbol and market returns has to be ready
94if len(self.selected_symbols) == 0 or not self.data[self.market].is_ready():
95    return
96
97market_monthly_returns: List[float] = [x for x in self.data[self.market].monthly_returns]
98
99beta: Dict[Symbol, float] = {}
100for symbol in self.selected_symbols:
101    stock_monthly_returns: List[float] = [x for x in self.data[symbol].monthly_returns]
102    
103    # Linear regression - X = market returns, Y = stock returns
104    slope, intercept, r_value, p_value, std_err = stats.linregress(market_monthly_returns, stock_monthly_returns)
105    beta[symbol] = slope
106
107# check if there are enough data for decile selection    
108if len(beta) < self.quantile:
109    self.Liquidate()
110    return
111  
112quantile: int = int(len(beta) / self.quantile) 
113sorted_by_beta: List[Symbol] = [x[0] for x in sorted(beta.items(), key=lambda item: item[1])]
114# long the lowest beta decile
115self.long: List[Symbol] = sorted_by_beta[:quantile]
116# short the highest beta decile
117self.short: List[Symbol] = sorted_by_beta[-quantile:]
118
119# Trade execution
120targets: List[PortfolioTarget] = []
121for i, portfolio in enumerate([self.long, self.short]):
122    for symbol in portfolio:
123        if symbol in data and data[symbol]:
124            targets.append(PortfolioTarget(symbol, ((-1) ** i) / len(portfolio)))
125
126self.SetHoldings(targets, True)
127def Rebalance(self) -> None:
128self.rebalance_flag = True
129def Selection(self) -> None:
130self.selection_flag = True
131
132def Liquidation(self) -> None:
133self.Liquidate()
134class SymbolData():
135def __init__(self, period: int) -> None:
136self.monthly_returns: RollingWindow = RollingWindow[float](period)
137self.last_month_price = 0
138
139def update_monthly_return(self, price: float) -> None:
140if self.last_month_price != 0:
141    monthly_return: float = (price - self.last_month_price) / self.last_month_price
142    self.monthly_returns.Add(monthly_return)
143self.last_month_price = price
144
145def is_ready(self) -> bool:
146return self.monthly_returns.IsReady
147# Custom fee model.
148class CustomFeeModel(FeeModel):
149def GetOrderFee(self, parameters: OrderFeeParameters) -> OrderFee:
150fee: float = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
151return OrderFee(CashAmount(fee, "USD"))