Spin-off Anomaly

1from AlgorithmImports import *
2import numpy as np
3from typing import List, Dict
4#endregion
5class SpinOffAnomaly(QCAlgorithm):
6def Initialize(self) -> None:
7 self.SetStartDate(2009, 1, 1) # spin off dates start at 2009
8 self.SetCash(100_000)
9 
10 self.data: Dict[Symbol, SymbolData] = {}    # storing stocks sector
11 self.tickers: List[str] = []                # storing tickers, which have spin offs
12 self.selected_symbols: List[Symbol] = []    # storing stocks for trading
13 self.managed_symbols: List[ManagedSymbol] = []
14 
15 self.holding_period: int = 6 * 21   # holding stocks and shorting ETF for n days
16 self.max_traded_stocks: int = 25    # maximum number of trading max_traded_stocks
17 self.leverage: int = 5
18 
19 self.etfs: Dict[int, str] = {
20     104: 'VNQ',  # Vanguard Real Estate Index Fund
21     311: 'XLK',  # Technology Select Sector SPDR Fund
22     309: 'XLE',  # Energy Select Sector SPDR Fund
23     206: 'XLV',  # Health Care Select Sector SPDR Fund
24     103: 'XLF',  # Financial Select Sector SPDR Fund
25     310: 'XLI',  # Industrials Select Sector SPDR Fund
26     101: 'XLB',  # Materials Select Sector SPDR Fund
27     205: 'XLY',  # Consumer Discretionary Select Sector SPDR Fund
28     102: 'XLP',  # Consumer Staples Select Sector SPDR Fund
29     207: 'XLU',  # Utilities Select Sector SPDR Fund    
30     308: 'XLC'   # Communications Services
31 }
32 
33 self.symbol: Symbol = self.AddEquity("SPY", Resolution.Daily).Symbol
34 
35 self.spin_offs: Symbol = self.AddData(QuantpediaSpinOffs, 'SPIN_OFFS', Resolution.Daily).Symbol
36 
37 for _, ticker in self.etfs.items():
38     security = self.AddEquity(ticker, Resolution.Daily)
39     security.SetFeeModel(CustomFeeModel())
40     security.SetLeverage(self.leverage)
41     
42     # create SymbolData object for etf
43     self.data[ticker] = SymbolData()
44     
45 self.selection_flag: bool = False
46 self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
47 self.settings.daily_precise_end_time = False
48 self.UniverseSettings.Resolution = Resolution.Daily
49 self.AddUniverse(self.FundamentalSelectionFunction)
50 
51def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
52 for security in changes.AddedSecurities:
53     security.SetFeeModel(CustomFeeModel())
54     security.SetLeverage(self.leverage)
55     
56def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
57 # rebalance, when spin off tickers came
58 if not self.selection_flag:
59     return Universe.Unchanged
60 self.selection_flag = False
61 
62 # select stocks, which had spin off
63 selected: List[Symbol] = [x.Symbol for x in fundamental if x.HasFundamentalData and x.Symbol.Value in self.tickers and x.AssetClassification.MorningstarSectorCode] 
64 
65 etfs_tickers: List[str] = [ticker for _, ticker in self.etfs.items()]
66 
67 self.selected_symbols = []
68 for stock in fundamental:
69     symbol: Symbol = stock.Symbol
70     ticker: str = symbol.Value
71     
72     # get last stock price for quantity calculation
73     if symbol in selected:
74         if symbol not in self.data:
75             self.data[symbol] = SymbolData()
76         
77         self.data[symbol].update_price(stock.AdjustedPrice)
78     
79         # get stock's sector
80         sector: str = stock.AssetClassification.MorningstarSectorCode
81         # store stock's sector
82         self.data[symbol].update_sector(sector)
83         self.selected_symbols.append(symbol)
84     # update etfs prices
85     if ticker in etfs_tickers:
86         self.data[ticker].update_price(stock.AdjustedPrice)
87 return self.selected_symbols
88def OnData(self, data: Slice) -> None:
89 spin_offs_last_update_date: Dict[Symbol, datetime.date] = QuantpediaSpinOffs.get_last_update_date()
90 if self.Securities[self.spin_offs].GetLastData() and self.Time.date() > spin_offs_last_update_date[self.spin_offs]:
91     self.Liquidate()
92     return
93 if self.spin_offs in data and data[self.spin_offs]:
94     self.tickers = [x for x in data[self.spin_offs].Tickers]
95     self.selection_flag = True
96  
97 # storing managed symbols, which need to be removed from self.managed_symbols
98 # and it's stock and ETF needs to be liquidated
99 remove_managed_symbols: List[ManagedSymbol] = [] 
100     
101 for managed_symbol in self.managed_symbols:
102     managed_symbol.holding_period += 1
103     
104     # stock has to be liquidate with it's ETF
105     if managed_symbol.holding_period == self.holding_period:
106         remove_managed_symbols.append(managed_symbol)
107         
108         # liquidate stock by selling it's quantity
109         self.MarketOrder(managed_symbol.symbol, -managed_symbol.stock_quantity)
110         # liquidate etf by buying it's quantity
111         self.MarketOrder(managed_symbol.etf, managed_symbol.etf_quantity)
112 
113 # remove managed symbols from self.managed_symbols        
114 for managed_symbol in remove_managed_symbols:
115     self.managed_symbols.remove(managed_symbol)
116 
117 # trade only if there are some stocks selected from FineSelectionFunction
118 if len(self.selected_symbols) == 0:
119     return
120 
121 for symbol in self.selected_symbols:
122     # check if there is a place for trading current stock
123     if len(self.managed_symbols) < self.max_traded_stocks:
124         # get stock's etf accoring to sector
125         if symbol not in self.data:
126             continue
127         etf: str = self.etfs[self.data[symbol].sector]
128         
129         # XLC doesn't have price
130         if self.data[etf].last_price == 0:
131             continue
132         
133         # this weight corresponds to stock and stock's etf according to sector
134         weight: float = self.Portfolio.TotalPortfolioValue / self.max_traded_stocks / 2
135         # calculate stock quantity
136         stock_quantity: int = np.floor(weight / self.data[symbol].last_price)
137         # calculate etf quantity
138         etf_quantity: int = np.floor(weight / self.data[etf].last_price)
139         
140         # create object of ManagedSymbol class, with stock's symbol,  stock's etf according to sector and their quantities
141         managed_symbol: ManagedSymbol = ManagedSymbol(symbol, etf, stock_quantity, etf_quantity)
142         
143         # long stock
144         self.MarketOrder(symbol, stock_quantity)
145         # short etf
146         self.MarketOrder(etf, -etf_quantity)
147         
148         # store created object of stock's ManagedSymbol
149         self.managed_symbols.append(managed_symbol)
150         
151 # clear stocks, which had spin off
152 self.selected_symbols.clear()
153         
154class ManagedSymbol():
155def __init__(self, symbol: Symbol, etf: str, stock_quantity: int, etf_quantity: int) -> None:
156 self.holding_period: int = 0
157 self.symbol: Symbol = symbol
158 self.etf: str = etf
159 self.stock_quantity: int = stock_quantity
160 self.etf_quantity: int = etf_quantity
161 
162class SymbolData():
163def __init__(self) -> None:
164 self.last_price: float = .0
165 self.sector: int = 0
166 
167def update_price(self, price: float) -> None:
168 self.last_price = price
169def update_sector(self, sector: int) -> None:
170 self.sector = sector
171# Quantpedia data.
172# NOTE: IMPORTANT: Data order must be ascending (datewise)
173class QuantpediaSpinOffs(PythonData):
174_last_update_date:Dict[Symbol, datetime.date] = {}
175@staticmethod
176def get_last_update_date() -> Dict[Symbol, datetime.date]:
177return QuantpediaSpinOffs._last_update_date
178
179def GetSource(self, config: SubscriptionDataConfig, date: datetime, isLiveMode: bool) -> SubscriptionDataSource:
180 return SubscriptionDataSource("data.quantpedia.com/backtesting_data/equity/{0}.csv".format(config.Symbol.Value), SubscriptionTransportMedium.RemoteFile, FileFormat.Csv)
181def Reader(self, config: SubscriptionDataConfig, line: str, date: datetime, isLiveMode: bool) -> BaseData:
182 data = QuantpediaSpinOffs()
183 data.Symbol = config.Symbol
184 
185 if not line[0].isdigit(): return None
186 split: str = line.split(';')
187 
188 data.Time = datetime.strptime(split[0], "%d.%m.%Y") + timedelta(days=1)
189 data['tickers'] = split[1:]
190 if config.Symbol not in QuantpediaSpinOffs._last_update_date:
191     QuantpediaSpinOffs._last_update_date[config.Symbol] = datetime(1,1,1).date()
192 if data.Time.date() > QuantpediaSpinOffs._last_update_date[config.Symbol]:
193     QuantpediaSpinOffs._last_update_date[config.Symbol] = data.Time.date()
194 return data
195# custom fee model
196class CustomFeeModel(FeeModel):
197def GetOrderFee(self, parameters: OrderFeeParameters) -> OrderFee:
198 fee: float = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
199 return OrderFee(CashAmount(fee, "USD"))