Institutional Ownership Effect

1from AlgorithmImports import *
2#endregion
3class InstitutionalOwnershipEffect(QCAlgorithm):
4def Initialize(self):
5self.SetStartDate(2005, 1, 1)
6self.SetCash(100_000)
7
8self.dates: List[datetime.date] = []
9self.institutional_ownership: Dict[Symbol, Dict] = {}
10
11self.data: Dict[Symbol, SymbolData] = {}
12self.period: int = 3 * 8 * 21 # Eight quarter period
13self.one_quarter_data: Dict[Symbol, List[float]] = {} # Storing one quarter data about each stock from our universe
14self.eight_quarter_data: Dict[Symbol, RollingWindow] = {} # Storing eight quarter data about each stock from our universe
15
16self.quantile: int = 10
17self.leverage: int = 5
18self.last_investment_date = None
19self.symbol: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
20
21self.sp100_stocks: List[Symbol] = [] # 'AGN', 'UTX', 'BRKB' # No data about institutional ownership
22
23csv_string_file: str = self.Download('data.quantpedia.com/backtesting_data/economic/institutional_ownership/institutional_ownership_in_millions.csv')
24lines: List[str] = csv_string_file.split('\r\n')
25for i in range(len(lines)):
26    line_split: List[str] = lines[i].split(',')
27    if i == 0: # First row are headers of columns
28        for j in range(1, len(line_split)): # Take all headers, which are stock tickers
29            ticker: str = line_split[j]
30            symbol: Symbol = self.AddEquity(ticker, Resolution.Daily).Symbol
31            
32            # Warmup volume data.
33            self.data[symbol] = SymbolData(self.period)
34            history: DataFrame = self.History(symbol, self.period, Resolution.Daily)
35            if history.empty:
36                self.Log(f"Not enough data for {symbol} yet")
37                continue
38            volumes: Series = history.loc[symbol].volume
39            for time, volume in volumes.items():
40                self.data[symbol].update(volume)
41            
42            self.one_quarter_data[symbol] = []
43            self.eight_quarter_data[symbol] = RollingWindow[float](8)
44            self.institutional_ownership[symbol] = {} # Create dictionary for all institutional ownership data
45            
46            # Subscript current stock and append symbol to universe, which we will use for this strategy 
47            self.sp100_stocks.append(symbol)
48    else:
49        date: datetime.date = datetime.strptime(line_split[0], "%Y-%m-%d").date()
50        
51        self.dates.append(date) # Create list of dates for institutional annoucements
52        
53        for j, symbol in zip(range(1, len(line_split)), self.sp100_stocks):
54            # Based on symbol store date and institutional ownership value as a nested dictionary
55            # In nested dictionary date figures as a key and institutional ownership as a value
56            self.institutional_ownership[symbol][date] = (line_split[j]) 
57
58self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
59self.count_month: int = 1
60self.selection_flag: bool = False
61self.Schedule.On(self.DateRules.MonthEnd(self.symbol), self.TimeRules.AfterMarketOpen(self.symbol), self.Selection)
62
63def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
64for security in changes.AddedSecurities:
65    security.SetFeeModel(CustomFeeModel())
66    security.SetLeverage(self.leverage)
67def OnData(self, slice: Slice) -> None:
68date = None
69inst_turn_in_div: Dict[Symbol, float] = {}
70
71current_date: datetime.date = self.Time.date()
72day_before: datetime.date = (self.Time - timedelta(days=1)).date()
73two_days_before: datetime.date = (self.Time - timedelta(days=2)).date()
74three_days_before: datetime.date = (self.Time - timedelta(days=3)).date()
75
76for symbol in self.sp100_stocks:
77    # Update daily volume data.
78    if symbol in slice:
79        if slice[symbol]:
80            self.data[symbol].update(slice[symbol].Volume)
81    
82    # Institutional ownership could be annouced on weekend. This prevents missing it.
83    # Without three days look ahead it was missing some annoucements
84    if (current_date in self.dates) and (current_date != self.last_investment_date):
85        self.last_investment_date = current_date
86        date = current_date
87    elif (day_before in self.dates) and (day_before != self.last_investment_date):
88        self.last_investment_date = day_before
89        date = day_before
90    elif (two_days_before in self.dates) and (two_days_before != self.last_investment_date):
91        self.last_investment_date = two_days_before
92        date = two_days_before
93    elif (three_days_before in self.dates) and (three_days_before != self.last_investment_date):
94        self.last_investment_date = three_days_before
95        date = three_days_before
96        
97    # If institutional ownership was announced store institutional_ownership value about each stock from our universe
98    if date in self.dates:
99        if self.institutional_ownership[symbol][date] != '':
100            self.one_quarter_data[symbol].append(float(self.institutional_ownership[symbol][date]))
101
102    if not self.selection_flag:
103        continue
104    absolute_change = None
105    if len(self.one_quarter_data[symbol]) >= 2:
106        absolute_change: float = self.one_quarter_data[symbol][-1] - self.one_quarter_data[symbol][0]
107        total_one_quarter: float = sum(self.one_quarter_data[symbol])
108        self.eight_quarter_data[symbol].Add(total_one_quarter)
109
110        # If eight quarter institutional ownership data are ready for chosen stock, then trade
111        if self.data[symbol].is_ready() and self.eight_quarter_data[symbol].IsReady:  
112            if symbol != 'ADP':
113                inst_turn_in_div_num: float = absolute_change / self.data[symbol].total_volume()
114                
115                total_inst_shares: float = sum([x for x in self.eight_quarter_data[symbol]])
116                inst_turn_in_div_num: float = inst_turn_in_div_num / total_inst_shares
117                
118                inst_turn_in_div[symbol] = inst_turn_in_div_num
119    
120    self.one_quarter_data[symbol].clear()
121
122if not self.selection_flag:
123    return
124self.selection_flag = False
125
126long: List[Symbol] = []
127short: List[Symbol] = []
128if len(inst_turn_in_div) >= self.quantile:
129    quantile: int = int(len(inst_turn_in_div) / self.quantile)
130    sorted_by_inst_turn_in_div = [x[0] for x in sorted(inst_turn_in_div.items(), key=lambda item: item[1])]
131    # long the lowest ‘InstTurnIndiv’ decile
132    long = sorted_by_inst_turn_in_div[:quantile]
133    # short the highest ‘InstTurnIndiv’ decile
134    short = sorted_by_inst_turn_in_div[-quantile:]
135
136# Trade execution.
137invested: List[Symbol] = [x.Key for x in self.Portfolio if x.Value.Invested]
138for symbol in invested:
139    if symbol not in long + short:
140        self.Liquidate(symbol)
141    
142for symbol in long:
143    self.SetHoldings(symbol, 1 / len(long))
144        
145for symbol in short:
146    self.SetHoldings(symbol, -1 / len(short))
147def Selection(self) -> None:
148if self.count_month == 3: # The portfolio is rebalance quarterly
149    self.selection_flag = True
150    self.count_month = 1
151else:
152    self.count_month += 1
153    
154class SymbolData():
155def __init__(self, period: int) -> None:
156self._volumes: RollingWindow = RollingWindow[float](period)
157
158def update(self, volume) -> None:
159self._volumes.Add(volume)
160
161def is_ready(self) -> bool:
162return self._volumes.IsReady
163
164def total_volume(self) -> float:
165return sum(list(self._volumes))
166
167class CustomFeeModel(FeeModel):
168def GetOrderFee(self, parameters):
169fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
170return OrderFee(CashAmount(fee, "USD"))