Cold IPOs Effect

1from AlgorithmImports import *
2from typing import List, Dict, Union, Tuple
3from dateutil.relativedelta import relativedelta
4from dataclasses import dataclass
5import numpy as np
6import datetime
7#endregion
8class ColdIPOsEffect(QCAlgorithm):
9def Initialize(self) -> None:
10self.SetStartDate(2010, 1, 1)
11self.SetCash(100_000)
12tickers_to_ignore: List[str] = ['EVOK', 'SGNL', 'VRDN', 'NRBO', 'GEMP', 'CCCR']
13self.UniverseSettings.Leverage = 10
14self.UniverseSettings.Resolution = Resolution.Daily
15self.AddUniverse(self.FundamentalSelectionFunction)
16self.Settings.MinimumOrderMarginPortfolioPercentage = 0.0
17self.settings.daily_precise_end_time = False
18
19self.holding_period: int = 6 # Months
20self.min_ipo_price: int = 5
21self.selection_flag: bool = False
22self.last_update_date: datetime.date = datetime.date(1700, 1, 1)
23self.traded_percentage: float = 0.3
24
25self.etf_symbols: List[Symbol] = []
26# Tuple (stock_ticker: str, is_cold_IPO: bool) in a list keyed by date
27self.ipo_dates: Dict[datetime.date, List[Tuple[str, bool]]] = {}        
28self.price_data: Dict[Symbol, float] = {}
29self.rebalancing_queue: List[RebalanceQueueItem] = []
30self.sector_etfs: Dict[int, Union[str, Symbol]] = {
31    104: 'VNQ',  # Vanguard Real Estate Index Fund
32    311: 'XLK',  # Technology Select Sector SPDR Fund
33    309: 'XLE',  # Energy Select Sector SPDR Fund
34    206: 'XLV',  # Health Care Select Sector SPDR Fund
35    103: 'XLF',  # Financial Select Sector SPDR Fund
36    310: 'XLI',  # Industrials Select Sector SPDR Fund
37    101: 'XLB',  # Materials Select Sector SPDR Fund
38    102: 'XLY',  # Consumer Discretionary Select Sector SPDR Fund
39    105: 'XLP',  # Consumer Staples Select Sector SPDR Fund
40    207: 'XLU'   # Utilities Select Sector SPDR Fund    
41}
42
43# Subscribe sector ETFs
44for sector_num, ticker in self.sector_etfs.items():
45    security = self.AddEquity(ticker, Resolution.Daily)
46    security.SetFeeModel(CustomFeeModel())
47    
48    # Change sector etf's ticker to sector etf's symbols
49    self.sector_etfs[sector_num] = security.Symbol
50    self.etf_symbols.append(security.Symbol)
51
52csv_string: str = self.Download('data.quantpedia.com/backtesting_data/equity/cold_ipos_formatted.csv')
53lines: List[str] = csv_string.split('\r\n') 
54# Skip csv header
55lines = lines[1:]
56
57for line in lines:
58    if line == '': continue
59    
60    splitted_line: List[str] = line.split(';')
61    
62    # csv header: date;ticker;offer_price;opening_price
63    date: datetime.date = datetime.datetime.strptime(splitted_line[0], "%d.%m.%Y").date()
64    ticker: str = splitted_line[1]
65    if ticker in tickers_to_ignore:
66        continue
67    offer_price: float = float(splitted_line[2])
68    opening_price: float = float(splitted_line[3])
69    if offer_price < self.min_ipo_price or opening_price < self.min_ipo_price:
70        continue
71    
72    if date not in self.ipo_dates:
73        self.ipo_dates[date] = []
74    if date > self.last_update_date:
75        self.last_update_date = date
76        
77    # Check if stock has cold IPO (offering price is greater than opening price)
78    if offer_price > opening_price:
79        self.ipo_dates[date].append((ticker, True))
80    else:
81        self.ipo_dates[date].append((ticker, False))
82
83market: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
84self.Schedule.On(self.DateRules.MonthStart(market), 
85                self.TimeRules.BeforeMarketClose(market), 
86                self.Selection)
87def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
88for security in changes.AddedSecurities:
89    security.SetFeeModel(CustomFeeModel())
90def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
91# Rebalance monthly
92if not self.selection_flag:
93    return Universe.Unchanged
94
95current_date: datetime.date = self.Time.date()
96prev_month_date: datetime.date = current_date - relativedelta(months=1)
97
98cold_IPO_tickers: List[str] = []
99
100for date in self.ipo_dates:
101    if date >= prev_month_date and date < current_date:
102        # Select stocks, which had cold IPO
103        for ticker, is_cold_IPO in self.ipo_dates[date]:
104            if is_cold_IPO:
105                cold_IPO_tickers.append(ticker)
106        
107# Store prices of stocks, which had cold IPO for weight calculations
108for f in fundamental:
109    symbol: Symbol = f.Symbol
110    ticker: str = symbol.Value
111    
112    if ticker in cold_IPO_tickers or symbol in self.etf_symbols:
113        self.price_data[symbol] = f.Price
114# Select only stocks, which have MorningsartSectorCode and exclude sector ETFs
115filtered: List[Fundamental] = [
116    f for f in fundamental if f.HasFundamentalData
117    and f.Symbol in self.price_data
118    and f.Symbol not in self.etf_symbols
119    and not f.CompanyReference.IsREIT
120    and not f.SecurityReference.IsDepositaryReceipt
121    and not np.isnan(f.AssetClassification.MorningstarSectorCode)
122]
123
124# Storing symbols of cold IPO stocks, which have sector value
125cold_IPOs_stocks_symbols: List[Symbol] = []
126# Storing total count of cold IPOs stocks keyed by sector number
127sectors_total_cold_IPOs: Dict[int, int] = {}  
128
129for f in filtered:
130    sector: int = f.AssetClassification.MorningstarSectorCode
131    
132    # Check if there is etf for stock's sector and sector etf has price
133    if sector not in self.sector_etfs or self.sector_etfs[sector] not in self.price_data:
134        continue
135    
136    # Initialize sector's total count of cold IPOs stocks
137    if sector not in sectors_total_cold_IPOs:
138        sectors_total_cold_IPOs[sector] = 0
139    
140    # Increase total count of cold IPOs stocks for specific sector
141    sectors_total_cold_IPOs[sector] += 1
142    
143    # Store symbol of cold IPO stock to list
144    cold_IPOs_stocks_symbols.append(f.Symbol)
145    
146long_symbol_q: List[Tuple[Symbol, float]] = []
147short_symbol_q: List[Tuple[Symbol, float]] = []
148# Calculate weights for stocks, which were selected
149if len(cold_IPOs_stocks_symbols) > 0:
150    portfolio_portion: float = (self.Portfolio.TotalPortfolioValue * self.traded_percentage) / self.holding_period / len(cold_IPOs_stocks_symbols)
151    long_symbol_q: List[Tuple[Symbol, float]] = [
152        (symbol, np.floor(portfolio_portion / self.price_data[symbol])) for symbol in cold_IPOs_stocks_symbols
153    ]
154    
155    total_cold_IPOs_count: int = sum(list(sectors_total_cold_IPOs.values()))
156    
157    for sector_num, total_sector_IPOs_count in sectors_total_cold_IPOs.items():
158        sector_symbol: Symbol = self.sector_etfs[sector_num]
159        price: float = self.price_data[sector_symbol]
160        
161        # Calculate sector weight:
162        # Divide weight by sector price, then multiply it by total count of cold IPO 
163        # stocks in this sector. This makes sure, that sectors are equally weigted based 
164        # on number of stocks, which had cold IPO in specific sector.
165        sector_weight: float = -np.floor((portfolio_portion / price) * (total_sector_IPOs_count / total_cold_IPOs_count))
166        
167        short_symbol_q.append((sector_symbol, sector_weight))
168    
169self.rebalancing_queue.append(RebalanceQueueItem(long_symbol_q + short_symbol_q))
170
171self.price_data.clear()
172
173return cold_IPOs_stocks_symbols + self.etf_symbols
174
175def OnData(self, slice: Slice) -> None:
176if not self.selection_flag:
177    return
178self.selection_flag = False
179if self.Time.date() > self.last_update_date + relativedelta(months=self.holding_period):
180    self.Liquidate()
181
182# Rebalance portfolio
183for item in self.rebalancing_queue:
184    if item.holding_period == self.holding_period:
185        for symbol, quantity in item.opened_symbol_quantity:
186            self.MarketOrder(symbol, -quantity)
187    
188    # Trade execution    
189    if item.holding_period == 0:
190        opened_symbol_quantity: List[Tuple[Symbol, float]] = []
191        
192        for symbol, quantity in item.opened_symbol_quantity:
193            if slice.contains_key(symbol) and slice[symbol] is not None:
194                self.MarketOrder(symbol, quantity)
195                opened_symbol_quantity.append((symbol, quantity))
196                    
197        # only opened orders will be closed        
198        item.opened_symbol_quantity = opened_symbol_quantity
199        
200    item.holding_period += 1
201    
202# Remove closed part of portfolio after loop
203self.rebalancing_queue = [
204    item for item in self.rebalancing_queue if item.holding_period <= self.holding_period
205]
206
207def Selection(self) -> None:
208self.selection_flag = True
209@dataclass
210class RebalanceQueueItem:
211opened_symbol_quantity: List[Tuple[Symbol, float]]
212holding_period: int = 0
213
214class CustomFeeModel(FeeModel):
215def GetOrderFee(self, parameters: OrderFeeParameters) -> OrderFee:
216fee: float = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
217return OrderFee(CashAmount(fee, "USD"))