Lame-Duck CEOs

1from AlgorithmImports import *
2from scipy import stats
3from typing import List, Dict
4from pandas.core.frame import DataFrame
5from pandas.core.series import Series
6# endregion
7class LameDuckCEOs(QCAlgorithm):
8def Initialize(self) -> None:
9 self.SetStartDate(2000, 1, 1)
10 self.SetCash(100_000)
11 
12 # Source: https://zenodo.org/record/4543893#.YwNo1xxBw2w
13 self.ceo_departure_dates: Symbol = self.AddData(CEODepartureDates, "CEODepartureDates", Resolution.Daily).Symbol
14 self.departure_company_tickers: List[str] = []           # recent month CEO departures
15 self.departure_company_ticker_universe: set = set()      # ticker universe from the whole dataset
16 self.selected_universe: List[Symbol] = []                # currently monthly selected stock universe
17 
18 self.price_data: Dict[Symbol, RollingWindow] = {}        # daily price data
19 self.period: int = 12 * 21
20 self.leverage: int = 10
21 self.min_share_price: int = 5
22 self.market: Symbol = self.AddEquity("SPY", Resolution.Daily, leverage=self.leverage).Symbol
23 self.price_data[self.market] = RollingWindow[float](self.period)
24 self.selection_flag: bool = False
25 self.UniverseSettings.Leverage = self.leverage
26 self.UniverseSettings.Resolution = Resolution.Daily
27 self.AddUniverse(self.FundamentalSelectionFunction)
28 self.settings.daily_precise_end_time = False
29 self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
30 self.Schedule.On(self.DateRules.MonthStart(self.market), self.TimeRules.AfterMarketOpen(self.market), self.Selection)
31def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
32 for security in changes.AddedSecurities:
33     security.SetFeeModel(CustomFeeModel())
34     
35def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
36 # update the rolling window every day
37 for stock in fundamental:
38     symbol = stock.Symbol
39     # Store monthly price.
40     if symbol in self.price_data:
41         self.price_data[symbol].Add(stock.AdjustedPrice)
42 if not self.selection_flag:
43     return Universe.Unchanged
44         
45 selected: List[Symbol] = [
46     x.Symbol for x in fundamental 
47     if x.Market == 'usa' 
48     and x.Price > self.min_share_price 
49     and x.Symbol.Value in self.departure_company_tickers
50 ]
51 for symbol in selected:
52     if symbol in self.price_data:
53         continue
54     
55     self.price_data[symbol] = RollingWindow[float](self.period)
56     history: DataFrame = self.History(symbol, self.period, Resolution.Daily)
57     if history.empty:
58         self.Log(f"Not enough data for {symbol} yet.")
59         continue
60     closes: Series = history.loc[symbol].close
61     for time, close in closes.items():
62         self.price_data[symbol].Add(close)
63 
64 if self.price_data[self.market].IsReady:
65     self.selected_universe = [x for x in selected if self.price_data[x].IsReady]
66 
67 return self.selected_universe
68def OnData(self, data: Slice) -> None:
69 custom_data_last_update_date: Dict[Symbol, datetime.date] = CEODepartureDates.get_last_update_date()
70 if self.Securities[self.ceo_departure_dates].GetLastData() and self.Time.date() > custom_data_last_update_date[self.ceo_departure_dates]:
71     self.Liquidate()
72     return Universe.Unchanged
73 # store new ceo departure data
74 if data.ContainsKey(self.ceo_departure_dates):
75     # store whole ticker universe
76     if len(self.departure_company_ticker_universe) == 0:
77         self.departure_company_ticker_universe = CEODepartureDates._ticker_universe
78     departure_tickers:str = data[self.ceo_departure_dates].GetProperty('stocks')
79     for t in departure_tickers:
80         self.departure_company_tickers.append(t)
81 # monthly rebalance
82 if not self.selection_flag:
83     return
84 self.selection_flag = False
85 # select long leg
86 long: List[Symbol] = []
87 for symbol in self.selected_universe:
88     if symbol.Value in self.departure_company_tickers:
89         long.append(symbol)
90 # reset ceo departure for the recent month
91 self.departure_company_tickers.clear()
92 if len(long) == 0:
93     if self.Portfolio.Invested:
94         self.Liquidate()
95     return
96 # order execution
97 total_beta: float = 0.
98 market_prices: np.ndarray = np.array(list(self.price_data[self.market]))
99 market_perf_data: np.ndarray = market_prices[:-1] / market_prices[1:] - 1
100 targets: List[PortfolioTarget] = []
101 for symbol in long:
102     # calculate beta to market for each stock
103     stocks_prices: np.ndarray = np.array(list(self.price_data[symbol]))
104     stock_perf_data: np.ndarray = stocks_prices[:-1] / stocks_prices[1:] - 1
105     slope, intercept, r_value, p_value, std_err = stats.linregress(market_perf_data, stock_perf_data)
106     total_beta += slope
107     if data.contains_key(symbol) and data[symbol]:
108         targets.append(PortfolioTarget(symbol, 1 / len(long)))
109 self.SetHoldings(targets, True)
110 avg_beta_to_market: float = total_beta / len(long)
111 # market hedge
112 self.SetHoldings(self.market, -avg_beta_to_market)
113def Selection(self) -> None:
114 if len(self.departure_company_ticker_universe) != 0:
115     self.selection_flag = True
116# Custom fee model.
117class CustomFeeModel(FeeModel):
118def GetOrderFee(self, parameters: OrderFeeParameters) -> OrderFee:
119 fee: float = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
120 return OrderFee(CashAmount(fee, "USD"))
121# CEO departure dates.
122# SOURCE: https://zenodo.org/record/4543893#.YwNo1xxBw2w
123# NOTE: IMPORTANT: Data order must be ascending (datewise)
124class CEODepartureDates(PythonData):
125_ticker_universe:Set[str] = set()
126_last_update_date:Dict[Symbol, datetime.date] = {}
127def GetSource(self, config:SubscriptionDataConfig, date:datetime, isLiveMode:bool) -> SubscriptionDataSource:
128 return SubscriptionDataSource("data.quantpedia.com/backtesting_data/economic/ceo_departure_dates.json", SubscriptionTransportMedium.RemoteFile, FileFormat.UnfoldingCollection)
129
130@staticmethod
131def get_last_update_date() -> Dict[Symbol, datetime.date]:
132return CEODepartureDates._last_update_date
133@staticmethod
134def get_ticker_universe() -> list:
135return list(CEODepartureDates._ticker_universe)
136def Reader(self, config:SubscriptionDataConfig, line:str, date:datetime, isLiveMode:bool) -> BaseData:
137 objects:list[CEODepartureDates] = []
138 data:list[dict] = json.loads(line)
139 end_time:datetime.date|None = None
140 for index, sample in enumerate(data):
141     custom_data:CEODepartureDates = CEODepartureDates()
142     custom_data.Symbol = config.Symbol
143     
144     departure_date:datetime.date = datetime.strptime(sample['departure_date'], '%Y-%m-%d')
145     custom_data.Time = departure_date
146     custom_data.EndTime = custom_data.Time + timedelta(days=1)
147     custom_data['stocks'] = sample['stocks']
148     custom_data.Value = 1
149     end_time = custom_data.EndTime
150     # store last date of the symbol
151     if config.Symbol not in CEODepartureDates._last_update_date:
152         CEODepartureDates._last_update_date[config.Symbol] = datetime(1,1,1).date()
153     if custom_data.Time.date() > CEODepartureDates._last_update_date[config.Symbol]:
154         CEODepartureDates._last_update_date[config.Symbol] = custom_data.Time.date()
155     for ticker in sample['stocks']:
156         CEODepartureDates._ticker_universe.add(ticker)
157     objects.append(custom_data)
158 return BaseDataCollection(end_time, config.Symbol, objects)