Influence of Liquidity, Institutional Ownership & Lottery Effect on Stocks

1from AlgorithmImports import *
2from pandas.core.frame import DataFrame
3from dateutil.relativedelta import relativedelta
4# endregion
5
6class InfluenceofLiquidityInstitutionalOwnershipLotteryEffectonStocks(QCAlgorithm):
7
8def Initialize(self):
9self.SetStartDate(2004, 1, 1)
10self.SetCash(100000)
11
12data_delay_months:int = 3
13file_contents:str = self.Download('data.quantpedia.com/backtesting_data/economic/institutional_ownership/institutional_ownership_in_percents.csv')
14lines:List[str] = file_contents.split('\r\n')
15self.tickers:List[str] = lines[0].split(',')[1:]
16
17dict_list:List[Dict[str, float]] = []
18
19for line in lines[1:]:
20    line_split:List[str] = line.split(',')
21    date = (datetime.strptime(line_split[0], "%Y-%m-%d") + relativedelta(months=data_delay_months)).date()
22
23    temp_dict:Dict[str, float] = { 'date' : date }
24
25    for i in range(1, len(line_split)):
26        ticker:str = self.tickers[i-1]
27        temp_dict[ticker] = float(line_split[i]) if line_split[i] != '' else 0.
28    
29    dict_list.append(temp_dict)
30
31io_df:DataFrame = pd.DataFrame(dict_list, columns=['date'] + self.tickers)
32self.io_df:DataFrame = io_df.set_index('date')
33
34self.leverage:int = 3
35self.selection_month:int = 12
36self.period:int = 12 * 21
37self.quantile:int = 3
38self.price_data:Dict[Symbol, RollingWindow] = {}
39self.weight:Dict[Symbol, float] = {}
40
41self.market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
42self.selection_flag:bool = False
43self.Settings.MinimumOrderMarginPortfolioPercentage = 0
44self.UniverseSettings.Resolution = Resolution.Daily
45self.AddUniverse(self.CoarseSelectionFunction, self.FineSelectionFunction)
46self.Schedule.On(self.DateRules.MonthStart(self.market), self.TimeRules.AfterMarketOpen(self.market), self.Selection)
47
48def OnSecuritiesChanged(self, changes:SecurityChanges) -> None:
49for security in changes.AddedSecurities:
50    security.SetFeeModel(CustomFeeModel())
51    security.SetLeverage(self.leverage)
52
53def CoarseSelectionFunction(self, coarse:List[CoarseFundamental]) -> List[Symbol]:
54# store daily stock prices
55for stock in coarse:
56    symbol:Symbol = stock.Symbol.Value
57
58    if symbol in self.price_data:
59        self.price_data[symbol].Add(stock.AdjustedPrice)
60
61# monthly selection
62if not self.selection_flag:
63    return Universe.Unchanged
64
65selection:List[Symbol] = [x.Symbol for x in coarse if x.HasFundamentalData and x.Symbol.Value in list(self.io_df.columns)]
66
67# warmup price rolling windows
68for symbol in selection:
69    if symbol in self.price_data:
70        continue
71    
72    self.price_data[symbol] = RollingWindow[float](self.period)
73   
74    history:DataFrame = self.History(symbol, self.period, Resolution.Daily)
75    if history.empty:
76        self.Log(f"Not enough data for {symbol} yet.")
77        continue
78    closes:pd.Series = history.loc[symbol].close
79    for time, close in closes.iteritems():
80        self.price_data[symbol].Add(close)
81
82return [x for x in selection if self.price_data[x].IsReady]
83
84def FineSelectionFunction(self, fine:List[FineFundamental]) -> List[Symbol]:
85stock_by_ticker:Dict[str, FineFundamental] = { x.Symbol.Value : x for x in fine }
86
87# filter stocks
88if not self.io_df.empty and len(stock_by_ticker) != 0:
89    last_io_values_sorted:DataFrame = self.io_df[self.io_df.index <= self.Time.date()]
90    if len(last_io_values_sorted) != 0:
91        last_io_values_sorted:pd.Series = last_io_values_sorted.iloc[-1].sort_values(ascending=False)
92        last_io_values_sorted = last_io_values_sorted[last_io_values_sorted != 0]
93        quantile:int = len(last_io_values_sorted) // self.quantile
94        bottom_io_tickers:List[str] = list(last_io_values_sorted[-quantile:].index)
95
96        bottom_io_MAX:Dict[FineFundamental, float] = { stock_by_ticker[x] : max(np.array(list(self.price_data[stock_by_ticker[x].Symbol]))[:-1] / np.array(list(self.price_data[stock_by_ticker[x].Symbol]))[1:] - 1) for x in bottom_io_tickers if x in stock_by_ticker }
97        MAX_median:float = np.median(list(bottom_io_MAX.values()))
98        
99        long:List[FineFundamental] = [stock for stock, MAX in bottom_io_MAX.items() if MAX <= MAX_median]
100        short:List[FineFundamental] = [stock for stock, MAX in bottom_io_MAX.items() if MAX > MAX_median]
101
102        # value weighting
103        market_cap_long:float = sum([x.MarketCap for x in long])
104        market_cap_short:float = sum([x.MarketCap for x in short])
105
106        for stock in long:
107            self.weight[stock.Symbol] = stock.MarketCap / market_cap_long
108
109        for stock in short:
110            self.weight[stock.Symbol] = -stock.MarketCap / market_cap_short
111
112return list(self.weight.keys())
113
114def OnData(self, data: Slice) -> None:
115if not self.selection_flag:
116    return
117self.selection_flag = False
118
119# trade execution
120invested:List[Symbol] = [x.Key for x in self.Portfolio if x.Value.Invested]
121for symbol in invested:
122    if symbol not in self.weight:
123        self.Liquidate(symbol)
124
125for symbol, w in self.weight.items():
126    self.SetHoldings(symbol, w)
127
128self.weight.clear()
129
130def Selection(self) -> None:
131if self.Time.month == self.selection_month:
132    self.selection_flag = True
133
134# custom fee model
135class CustomFeeModel(FeeModel):
136def GetOrderFee(self, parameters):
137fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
138return OrderFee(CashAmount(fee, "USD"))