How Satellite Launches Influence Stock Returns

1from AlgorithmImports import *
2from pandas.tseries.offsets import BDay
3from collections import deque
4import data_tools
5import statsmodels.api as sm
6import numpy as np
7# endregion
8
9class HowSatelliteLaunchesInfluenceStockReturns(QCAlgorithm):
10
11def Initialize(self):
12self.SetStartDate(2010, 1, 1)
13self.SetCash(100000)
14
15self.market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
16
17ff_tickers:List[str] = [
18    'fama_french_5_market_eq',
19    'fama_french_5_investment_eq',
20    'fama_french_5_profitability_eq',
21    'fama_french_5_size_eq',
22    'fama_french_5_value_eq',
23]
24self.ff_factors:List[Symbol] = [self.AddData(data_tools.FFFactorsEQ, ff_ticker, Resolution.Daily).Symbol for ff_ticker in ff_tickers]
25self.market_ff:Symbol = self.ff_factors[0]
26
27self.traded_portfolio_portion:Dict[Symbol, float] = {}
28self.data:Dict[Symbol, deque] = {}
29self.overpriced_betas:Dict[Symbol, float] = {}
30self.underpriced_betas:Dict[Symbol, float] = {}
31
32self.stocks_to_liquidate:List[data_tools.HoldingItem] = []
33self.active_coarse:list[CoarseFundamental] = []
34
35# we employ one-year data up to 10 days prior to the event (i.e., from t – 375 to t – 10)
36self.market_period:int = 375
37self.pre_launch_period:int = 2      # [-2, 0] holding period @table 8
38self.t_10:int = 10 - self.pre_launch_period
39self.holding_period:int = 3         # days [-2, -1, 0] before launch
40
41self.quantile:int = 5
42self.leverage:int = 5
43self.coarse_count:int = 500
44
45# load Satelite launch dates
46# Source: https://en.wikipedia.org/wiki/2000_in_spaceflight
47csv_string_file:str = self.Download('data.quantpedia.com/backtesting_data/calendar/spaceflight_dates.csv')
48dates:str = csv_string_file.split('\r\n')
49self.launch_dates:List[datetime.date] = [datetime.strptime(x, "%Y-%m-%d") - BDay(self.pre_launch_period) for x in dates[:-1]]
50
51self.selection_flag:bool = False
52self.rebalance_flag:bool = False
53self.UniverseSettings.Resolution = Resolution.Daily
54self.AddUniverse(self.CoarseSelectionFunction, self.FineSelectionFunction)
55self.Schedule.On(self.DateRules.MonthStart(self.market), self.TimeRules.AfterMarketOpen(self.market), self.Selection)
56
57def OnSecuritiesChanged(self, changes:SecurityChanges) -> None:
58for security in changes.AddedSecurities:
59    security.SetFeeModel(data_tools.CustomFeeModel())
60    security.SetLeverage(self.leverage)
61
62def CoarseSelectionFunction(self, coarse:List[CoarseFundamental]) -> List[Symbol]:
63# update the returns every day
64ff_factors_last_update_date:datetime.date = data_tools.FFFactorsEQ._last_update_date
65
66# FF data is still comming in
67if all([self.Securities[x].GetLastData() for x in self.ff_factors]) and self.Time.date() >= ff_factors_last_update_date:
68    self.Liquidate()
69    return Universe.Unchanged
70
71# store daily FF prices
72for symbol_ff in self.ff_factors:
73    if symbol_ff in self.data:
74        price = self.Securities[symbol_ff].Price
75        if symbol_ff == self.market_ff:
76            self.data[symbol_ff].update_daily_return(self.Time, price)
77        else:
78            self.data[symbol_ff].update_value(self.Time, price)
79
80# store daily stock prices
81for stock in coarse:
82    symbol:Symbol = stock.Symbol
83
84    if symbol in self.data:
85        self.data[symbol].update_daily_return(self.Time , stock.AdjustedPrice)
86
87if self.Time in self.launch_dates:
88    self.rebalance_flag = True
89    return self.active_coarse
90
91# selection on month start
92if not self.selection_flag:
93    return Universe.Unchanged
94self.selection_flag = False
95
96self.active_coarse:List[Symbol] = [x.Symbol
97    for x in sorted([x for x in coarse if x.HasFundamentalData and x.Market == 'usa'],
98        key = lambda x: x.DollarVolume, reverse = True)[:self.coarse_count]]
99# selected = [x.Symbol for x in coarse if x.HasFundamentalData and x.Market == 'usa']
100
101# price warmup
102for symbol in self.active_coarse:
103    if symbol in self.data:
104        continue
105    
106    self.data[symbol] = data_tools.SymbolData(self.market_period)
107    history:DataFrame = self.History(symbol, self.market_period, Resolution.Daily)
108    if history.empty:
109        self.Log(f"Not enough data for {symbol} yet.")
110        continue
111    closes:pd.Series = history.loc[symbol].close
112    for time, close in closes.iteritems():
113        self.data[symbol].update_daily_return(time, close)
114
115for symbol_ff in self.ff_factors:
116    if symbol_ff not in self.data:
117        self.data[symbol_ff] = data_tools.SymbolData(self.market_period)
118        history:DataFrame = self.History(symbol_ff, self.market_period, Resolution.Daily)
119        if not history.empty:
120            values:pd.Series = history.loc[symbol_ff].value
121            for time, value in values.iteritems():
122                if symbol_ff == self.market_ff:
123                    self.data[symbol_ff].update_daily_return(time, value)
124                else:
125                    self.data[symbol_ff].update_value(time, value)
126        else:
127            self.Log(f"Not enough data for {symbol_ff} yet.")
128
129return [x for x in self.active_coarse if self.data[x].is_ready()]
130
131def FineSelectionFunction(self, fine:List[FineFundamental]) -> List[Symbol]:
132fine = [x for x in fine if x.MarketCap != 0 and \
133        (x.SecurityReference.ExchangeId == 'NYS') or (x.SecurityReference.ExchangeId == 'NAS') or (x.SecurityReference.ExchangeId == 'ASE')]
134
135fine:Dict[Symbol, FineFundamental] = {x.Symbol: x for x in fine}
136
137if len(fine) != 0:
138    last_start:datetime.date = (self.Time.date() - timedelta(self.market_period))
139    last_end:datetime.date = (self.Time.date() - timedelta(self.t_10))
140    if not all([self.data[x].is_ready() for x in self.ff_factors]):
141        return Universe.Unchanged
142    
143    # stock returns
144    daily_returns_by_stock:Dict[Symbol, List[Tuple[datetime, float]]] = {sym : sym_data.get_daily_returns(last_start, last_end) for sym, sym_data in self.data.items() if sym_data.is_ready() and sym in fine}
145    stock_daily_returns:List = list(zip(*[[i[1] for i in x] for x in daily_returns_by_stock.values()]))
146    
147    # FF factors values
148    launch_dates:List[datetime] = pd.to_datetime(self.launch_dates)
149    ff_returns_with_dt, ff_size_with_dt, ff_investment_with_dt, ff_profitability_with_dt, ff_book_to_market_with_dt = [np.array(self.data[x].get_daily_returns(last_start, last_end)) for x in self.ff_factors]
150    ff_returns = ff_returns_with_dt[:, 1]
151    ff_size = ff_size_with_dt[:, 1]
152    ff_investment = ff_investment_with_dt[:, 1]
153    ff_profitability = ff_profitability_with_dt[:, 1]
154    ff_book_to_market = ff_book_to_market_with_dt[:, 1]
155    
156    # regression X variables
157    transformed_array = np.column_stack((
158        ff_returns,
159        ff_returns * np.where(ff_returns >= 0, 1, 0),
160        ff_returns * np.where(ff_returns < 0, 1, 0),
161        np.isin(ff_returns_with_dt[:, 0], self.launch_dates).astype(int), # satellite_w
162        ff_size,
163        ff_investment,
164        ff_profitability,
165        ff_book_to_market)) 
166
167    # apply satellite_w to particular variable columns
168    transformed_array[:, 1] *= transformed_array[:, 3]
169    transformed_array[:, 2] *= transformed_array[:, 3]
170
171    transformed_array = np.delete(transformed_array, 3, axis=1)
172    transformed_array = np.array(transformed_array, dtype=float)
173
174    # run stock regression
175    x:np.ndarray = transformed_array
176    y:np.ndarray = stock_daily_returns
177
178    if len(x) != len(y) or len(x) == 0 or len(y) == 0:
179        self.Log('Data missing')
180        return Universe.Unchanged
181
182    model = self.multiple_linear_regression(x, y)
183    beta_values_overpriced:np.ndarray = model.params[1]
184    beta_values_underpriced:np.ndarray = model.params[2]
185
186    overpriced_betas:Dict[Symbol, float] = {}
187    underpriced_betas:Dict[Symbol, float] = {}
188    
189    # fetch beta parameters for each stock
190    for i, asset in enumerate(daily_returns_by_stock):
191        asset_s:Symbol = self.Symbol(asset)
192
193        # fill data
194        if asset in fine:
195            if beta_values_overpriced[i] != 0 and beta_values_overpriced[i] is not None and \
196                beta_values_underpriced[i] != 0 and beta_values_underpriced[i] is not None:
197                overpriced_betas[fine[asset]] = beta_values_overpriced[i]
198                underpriced_betas[fine[asset]] = beta_values_underpriced[i]
199
200    # sort by beta and divide to upper decile and lower quintile
201    if len(overpriced_betas) >= self.quantile:
202        sorted_overprized_betas:List[FineFundamental] = sorted(overpriced_betas, key=overpriced_betas.get)
203        sorted_underprized_betas:List[FineFundamental] = sorted(underpriced_betas, key=underpriced_betas.get)
204        quantile:int = int(len(sorted_overprized_betas) / self.quantile)
205        long_underprized:List[FineFundamental] = sorted_underprized_betas[-quantile:]
206        short_overprized:List[FineFundamental] = sorted_overprized_betas[-quantile:]
207
208        # calculate weights based on values
209        sum_long:float = sum([x.MarketCap for x in long_underprized])
210        for stock in long_underprized:
211            self.traded_portfolio_portion[stock.Symbol] = (stock.MarketCap / sum_long) * (self.Portfolio.TotalPortfolioValue / self.holding_period)
212
213        sum_short:float = sum([x.MarketCap for x in short_overprized])
214        for stock in short_overprized:
215            self.traded_portfolio_portion[stock.Symbol] = (-stock.MarketCap / sum_short) * (self.Portfolio.TotalPortfolioValue / self.holding_period)
216
217# return list(fine.keys())
218return list(self.traded_portfolio_portion.keys())
219
220def OnData(self, data: Slice) -> None:
221items_to_remove:List[data_tools.HoldingItem] = []
222
223# execute order and hold for holding period
224for item in self.stocks_to_liquidate:
225    item._holding_period += 1
226    if item._holding_period >= self.holding_period:
227        self.MarketOrder(item._symbol, -item._quantity)
228        items_to_remove.append(item)    
229
230# remove from collection
231for item in items_to_remove:
232    self.stocks_to_liquidate.remove(item)    
233
234if not self.rebalance_flag:
235    return
236self.rebalance_flag = False
237
238# execute order
239for price_symbol, portfolio_portion in self.traded_portfolio_portion.items():
240    if price_symbol in data and data[price_symbol]:
241        final_quantity:int = portfolio_portion // data[price_symbol].Price
242        if portfolio_portion != 0:
243            self.MarketOrder(price_symbol, final_quantity)
244            self.stocks_to_liquidate.append(data_tools.HoldingItem(price_symbol, final_quantity))
245
246self.traded_portfolio_portion.clear()
247
248def Selection(self) -> None:
249self.selection_flag = True
250
251def multiple_linear_regression(self, x:np.ndarray, y:np.ndarray):
252# x:np.ndarray = np.array(x).T
253x = sm.add_constant(x, prepend=True)
254result = sm.OLS(endog=y, exog=x).fit()
255return result