Catching “Falling Knife” Stocks

1from AlgorithmImports import *
2from typing import List, Dict
3import numpy as np
4from numpy import isnan
5class CatchingFallingKnifeStocks(QCAlgorithm):
6def Initialize(self):
7self.SetStartDate(2000, 1, 1)
8self.SetCash(100000)
9self.fundamental_sorting_key = lambda x: x.MarketCap
10self.fundamental_count:int = 3000
11self.period:int = 500
12
13self.long:List[Symbol] = []
14# Daily data
15self.data:Dict[Symbol, SymbolData] = {}
16self.min_share_price:int = 5
17self.leverage:int = 3
18
19self.symbol:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
20if self.symbol not in self.data:
21    self.data[self.symbol] = SymbolData(self.symbol, self.period)
22    history:DataFrame = self.History(self.symbol, self.period, Resolution.Daily)
23    if history.empty:
24        self.Log(f"Note enough data for {symbol} yet")
25    else:
26        closes:Series = history.loc[self.symbol].close[:-1]
27        for time, close in closes.items():
28            self.data[self.symbol].update(close)
29
30self.last_month:int = -1
31self.selection_flag:bool = True
32self.settings.daily_precise_end_time = False
33self.settings.minimum_order_margin_portfolio_percentage = 0.
34self.UniverseSettings.Resolution = Resolution.Daily
35self.AddUniverse(self.FundamentalSelectionFunction)
36self.market: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
37self.schedule.on(self.date_rules.month_start(self.market),
38                self.time_rules.after_market_open(self.market),
39                self.selection)
40
41def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
42for security in changes.AddedSecurities:
43    security.SetFeeModel(CustomFeeModel())
44    security.SetLeverage(self.leverage)
45def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
46for stock in fundamental:
47    symbol:Symbol = stock.Symbol
48    
49    # Store daily price.
50    if symbol in self.data:
51        self.data[symbol].update(stock.AdjustedPrice)
52if not self.selection_flag:
53    return Universe.Unchanged
54selected:List[Fundamental] = [
55    x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and x.Price > self.min_share_price and x.MarketCap != 0 \
56    and not isnan(x.OperationRatios.TotalDebtEquityRatio.ThreeMonths) and x.OperationRatios.TotalDebtEquityRatio.ThreeMonths > 0 \
57    and not isnan(x.AssetClassification.MorningstarIndustryGroupCode) and x.AssetClassification.MorningstarIndustryGroupCode != 0
58]
59
60if len(selected) > self.fundamental_count:
61    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
62group:Dict[str, List[Symbol]] = {}
63# Warmup price rolling windows.
64for stock in selected:
65    symbol:Symbol = stock.Symbol
66    if symbol not in self.data:
67        self.data[symbol] = SymbolData(symbol, self.period)
68        history = self.History(symbol, self.period, Resolution.Daily)
69        if history.empty:
70            self.Log(f"Not enough data for {symbol} yet")
71            continue
72        closes = history.loc[symbol].close
73        for time, close in closes.items():
74            self.data[symbol].update(close)
75    if not self.data[symbol].is_ready():
76        continue
77    industry_group_code = stock.AssetClassification.MorningstarIndustryGroupCode
78    # Debt to equity ratio.
79    debt_to_equity = stock.OperationRatios.TotalDebtEquityRatio.ThreeMonths
80    self.data[symbol]._debt_to_equity = debt_to_equity
81    
82    # Adding stocks in groups
83    if not industry_group_code in group:
84        group[industry_group_code] = []
85    
86    group[industry_group_code].append(self.data[symbol])
87
88if self.symbol in self.data and self.data[self.symbol].is_ready():
89    spy_ret:float = self.data[self.symbol].performance()
90            
91    for industry_code in group:
92        industry_debt_to_equity_10th_percentile:float = np.percentile([symbol_data._debt_to_equity for symbol_data in group[industry_code]], 10)
93            
94        # Stocks that suffered losses of 50 percent or more than s&p
95        # and
96        # stocks that have a Debt/Equity ratio within at least 10% of the lowest in the industry
97        long:List[Symbol] = [symbol_data._symbol for symbol_data in group[industry_code] if symbol_data.performance() <= (spy_ret - 0.5) \
98                            and symbol_data._debt_to_equity <= industry_debt_to_equity_10th_percentile]
99        
100        for symbol in long:
101            self.long.append(symbol)
102            
103return self.long
104def OnData(self, data: Slice) -> None:
105if not self.selection_flag: 
106    return
107self.selection_flag = False
108
109# Trade execution.        
110targets:List[PortfolioTarget] = []
111for symbol in self.long:
112    if symbol in data and data[symbol]:
113        targets.append(PortfolioTarget(symbol, 1 / len(self.long)))
114
115self.SetHoldings(targets, True)
116self.long.clear()
117def selection(self) -> None:
118self.selection_flag = True
119class SymbolData():
120def __init__(self, symbol:Symbol, period:int) -> None:
121self._symbol:Symbol = symbol
122self._price:RollingWindow = RollingWindow[float](period)
123self._debt_to_equity:float = 0.
124
125def is_ready(self) -> bool:
126return self._price.IsReady
127
128def update(self, close:float) -> None:
129self._price.Add(close)
130
131def performance(self) -> float:
132return (self._price[0] / self._price[self._price.Count - 1] - 1)
133# Custom fee model.
134class CustomFeeModel(FeeModel):
135def GetOrderFee(self, parameters):
136fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
137return OrderFee(CashAmount(fee, "USD"))