Liquidity Volatility in Stocks

1from AlgorithmImports import *
2from typing import List, Dict
3import data_tools
4from dateutil.relativedelta import relativedelta
5# endregion
6
7class LiquidityVolatilityinStocks(QCAlgorithm):
8
9def Initialize(self):
10self.SetStartDate(2000, 1, 1)
11self.SetCash(100000)
12
13self.market:Symbol = self.AddEquity("SPY", Resolution.Daily).Symbol
14 
15self.tickers_to_ignore:List[str] = ['KELYB', 'BRKB', 'SGA']
16
17self.fundamental_count:int = 500
18self.fundamental_sorting_key = lambda x: x.DollarVolume
19
20self.leverage:int = 5
21self.first_quantile:int = 3
22self.second_quantile:int = 5
23self.short_period:int = 12
24self.long_period:int = 36
25
26self.data:Dict[Symbol, SymbolData] = {}
27self.weight:Dict[Symbol, float] = {}
28
29self.single_sort_flag:bool = False
30self.selection_flag:bool = False
31self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
32self.UniverseSettings.Resolution = Resolution.Daily
33self.AddUniverse(self.FundamentalSelectionFunction)
34self.Schedule.On(self.DateRules.MonthStart(self.market), self.TimeRules.AfterMarketOpen(self.market), self.Selection)
35
36def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
37for security in changes.AddedSecurities:
38    security.SetFeeModel(data_tools.CustomFeeModel())
39    security.SetLeverage(self.leverage)
40
41def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
42# store daily stock prices
43for stock in fundamental:
44    symbol:Symbol = stock.Symbol
45    if stock.Volume != 0:
46        volume:float = stock.Volume
47    else:
48        continue
49
50    if symbol in self.data:
51        self.data[symbol].update_daily_data(stock.AdjustedPrice, volume)
52
53# monthly selection
54if not self.selection_flag:
55    return Universe.Unchanged
56
57selected:List[Fundamental] = [x for x in fundamental if x.HasFundamentalData and x.MarketCap != 0 and x.Market == 'usa' \
58                        and ((x.SecurityReference.ExchangeId == "NYS") or (x.SecurityReference.ExchangeId == "NAS") or (x.SecurityReference.ExchangeId == "ASE")) \
59                        and x.Symbol.Value not in self.tickers_to_ignore]
60
61if len(selected) > self.fundamental_count:
62    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
63
64liquidity_level:Dict[Symbol, float] = {}
65coefficient_variation:Dict[Symbol, float] = {}
66
67# price warmup
68for stock in selected:
69    symbol:Symbol = stock.Symbol
70    if symbol not in self.data:
71        self.data[symbol] = data_tools.SymbolData(self.short_period, self.long_period)
72        history:DataFrame = self.History(symbol, start=self.Time.date() - relativedelta(months=1), end=self.Time.date())
73        if history.empty:
74            self.Log(f"Not enough data for {symbol} yet.")
75            continue
76        data:pd.DataFrame = history.loc[symbol]
77        for time, row in data.iterrows():
78            if 'volume' not in row:
79                continue
80            self.data[symbol].update_daily_data(row.close, row.volume)
81
82    if self.data[symbol].is_ready():
83        coeff_variation:float = self.data[symbol].coefficient_variation_calculation()
84        if coeff_variation is not None:
85            liquidity_level[stock] = self.data[symbol].get_liquidity_level()
86            coefficient_variation[stock] = coeff_variation
87
88if len(coefficient_variation) == 0 or len(liquidity_level) == 0:
89    return Universe.Unchanged
90
91liquidity_portfolios:List[Tuple[List[Fundamental]]] = []
92
93if len(liquidity_level) >= self.first_quantile * self.second_quantile:
94    # single sort by coefficient of variation
95    if self.single_sort_flag:
96        sorted_liquidity_symbols = sorted(coefficient_variation, key=coefficient_variation.get)
97        quantile:int = int(len(sorted_liquidity_symbols) / self.second_quantile)
98        long:List[Fundamental] = sorted_liquidity_symbols[:quantile]
99        short:List[Fundamental] = sorted_liquidity_symbols[-quantile:]
100
101        liquidity_portfolios.append((long, short))
102
103    # double sort by liquidity level and coefficient of variation
104    else:
105        sorted_liquidity_symbols:List[Fundamental] = sorted(liquidity_level, key=liquidity_level.get)
106        quantile:int = int(len(sorted_liquidity_symbols) / self.first_quantile)
107        first_group:List[Fundamental] = list(sorted_liquidity_symbols)[:quantile]
108        second_group:List[Fundamental] = list(sorted_liquidity_symbols)[quantile:-quantile]
109        third_group:List[Fundamental] = list(sorted_liquidity_symbols)[-quantile:]
110
111        for group in [first_group, second_group, third_group]:
112            sorted_group = sorted({symbol: value for symbol, value in coefficient_variation.items() if symbol in group}.items(), key=lambda x: x[1])
113            quantile:int = int(len(sorted_group) / self.second_quantile)
114            liquidity_portfolios.append(([x[0] for x in sorted_group[:quantile]], [x[0] for x in sorted_group[-quantile:]]))
115
116    # calculate weights based on market cap
117    for group in liquidity_portfolios:
118        for i, portfolio in enumerate(group):
119            mc_sum:float = sum(list(map(lambda stock: stock.MarketCap , portfolio)))
120            for stock in portfolio:
121                self.weight[stock.Symbol] = (((-1)**i) * stock.MarketCap / mc_sum) * (1 / len(liquidity_portfolios))
122    
123return list(self.weight.keys())
124
125def OnData(self, data: Slice) -> None:
126if not self.selection_flag:
127    return
128self.selection_flag = False
129
130# trade execution
131portfolio:List[PortfolioTarget] = [PortfolioTarget(symbol, w) for symbol, w in self.weight.items() if symbol in data and data[symbol]]
132self.SetHoldings(portfolio, True)
133
134self.weight.clear()
135
136def Selection(self) -> None:
137self.selection_flag = True