Fundamental Strength and the 52-Week High Anomaly

1from numpy import floor, isnan
2from AlgorithmImports import *
3from pandas.core.frame import DataFrame
4import data_tools
5from functools import reduce
6class FundamentalStrength52WeekHigh(QCAlgorithm):
7def Initialize(self):
8self.SetStartDate(2000, 1, 1)
9self.SetCash(100000)
10
11self.period:int = 52 * 5
12self.holding_period:int = 6
13self.quantile:int = 5
14self.leverage:int = 5
15self.min_share_price:float = 5.
16self.exchange_codes:List[str] = ['NYS', 'NAS', 'ASE']
17
18self.data:Dict[Symbol, data_tools.SymbolData] = {}
19self.last_fine:List[Symbol] = []
20self.managed_queue:List[data_tools.RebalanceQueueItem] = []
21self.financial_statement_names:List[str] = [
22    'EarningReports.BasicAverageShares.ThreeMonths',
23    'EarningReports.BasicEPS.TwelveMonths',
24    'OperationRatios.ROA.ThreeMonths',
25    'OperationRatios.GrossMargin.ThreeMonths',
26    'FinancialStatements.CashFlowStatement.CashFlowFromContinuingOperatingActivities.ThreeMonths',
27    'FinancialStatements.IncomeStatement.NormalizedIncome.ThreeMonths',
28    'FinancialStatements.BalanceSheet.LongTermDebt.ThreeMonths',
29    'FinancialStatements.BalanceSheet.TotalAssets.ThreeMonths',
30    'FinancialStatements.BalanceSheet.OrdinarySharesNumber.ThreeMonths',
31    'FinancialStatements.IncomeStatement.TotalRevenueAsReported.ThreeMonths',
32    'ValuationRatios.PERatio',
33    'OperationRatios.CurrentRatio.ThreeMonths',
34]
35market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
36
37self.fundamental_count:int = 3000
38self.fundamental_sorting_key = lambda x: x.MarketCap
39self.selection_flag = False
40self.UniverseSettings.Resolution = Resolution.Daily
41self.AddUniverse(self.FundamentalSelectionFunction)
42self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
43self.Schedule.On(self.DateRules.MonthStart(market), self.TimeRules.AfterMarketOpen(market), self.Selection)
44self.settings.daily_precise_end_time = False
45def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
46for security in changes.AddedSecurities:
47    security.SetFeeModel(data_tools.CustomFeeModel())
48    security.SetLeverage(self.leverage)
49        
50def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
51# update the rolling window every day
52for stock in fundamental:
53    symbol:Symbol = stock.Symbol
54    # Store monthly price.
55    if symbol in self.data:
56        self.data[symbol].update_price(stock.AdjustedPrice)
57if not self.selection_flag:
58    return Universe.Unchanged
59selected:List[Fundamental] = [
60    x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and x.Price > self.min_share_price and \
61    all((not isnan(self.rgetattr(x, statement_name)) and self.rgetattr(x, statement_name) != 0) for statement_name in self.financial_statement_names)
62]
63if len(selected) > self.fundamental_count:
64    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
65score = {}
66nearness = {}
67# warmup price rolling windows
68for stock in selected:
69    symbol:Symbol = stock.Symbol
70    if symbol not in self.data:
71        self.data[symbol] = data_tools.SymbolData(symbol, self.period)
72        history = self.History(symbol, self.period, Resolution.Daily)
73        if history.empty:
74            self.Log(f"Not enough data for {symbol} yet.")
75            continue
76        closes = history.loc[symbol].close
77        for time, close in closes.items():
78            self.data[symbol].update_price(close)
79    
80    if self.data[symbol].is_ready():
81        nearness[symbol] = self.data[symbol].nearness()
82        
83        # FSCORE calc            
84        roa:float = stock.OperationRatios.ROA.ThreeMonths
85        cfo:float = stock.FinancialStatements.CashFlowStatement.CashFlowFromContinuingOperatingActivities.ThreeMonths
86        leverage:float = stock.FinancialStatements.BalanceSheet.LongTermDebt.ThreeMonths / stock.FinancialStatements.BalanceSheet.TotalAssets.ThreeMonths
87        liquidity:float = stock.OperationRatios.CurrentRatio.ThreeMonths
88        equity_offering:float = stock.FinancialStatements.BalanceSheet.OrdinarySharesNumber.ThreeMonths
89        gross_margin:float = stock.OperationRatios.GrossMargin.ThreeMonths
90        turnover:float = stock.FinancialStatements.IncomeStatement.TotalRevenueAsReported.ThreeMonths / stock.FinancialStatements.BalanceSheet.TotalAssets.ThreeMonths
91        
92        symbol_data = self.data[symbol]
93        # check if data has previous year's data ready and their values are consecutive
94        if (not symbol_data.data_is_set()) or (symbol not in self.last_fine):
95            symbol_data.update_data(roa, leverage, liquidity, equity_offering, gross_margin, turnover)
96            continue
97        score[symbol] = 0
98        if roa > 0:
99            score[symbol] += 1
100        if cfo > 0:
101            score[symbol] += 1
102        if roa > symbol_data.ROA:   # ROA change is positive
103            score[symbol] += 1
104        if cfo > roa:
105            score[symbol] += 1
106        if leverage < symbol_data.Leverage:
107            score[symbol] += 1
108        if liquidity > symbol_data.Liquidity:
109            score[symbol] += 1
110        if equity_offering < symbol_data.Equity_offering:
111            score[symbol] += 1
112        if gross_margin > symbol_data.Gross_margin:
113            score[symbol] += 1
114        if turnover > symbol_data.Turnover:
115            score[symbol] += 1
116        
117        # assing new (this year's) data
118        symbol_data.update_data(roa, leverage, liquidity, equity_offering, gross_margin, turnover)
119
120long:List[Symbol] = []
121short:List[Symbol] = []
122    
123if len(score) != 0 and len(nearness) >= self.quantile:
124    # nearness sorting and F score sorting
125    sorted_by_nearness:List[Symbol] = sorted(nearness, key = nearness.get, reverse = True)
126    quantile:int = int(len(sorted_by_nearness) / self.quantile)
127    high_by_nearness = sorted_by_nearness[:quantile]
128    low_by_nearness = sorted_by_nearness[-quantile:]
129    long = [x[0] for x in score.items() if x[1] >= 7 and x[0] in high_by_nearness]
130    short = [x[0] for x in score.items() if x[1] <= 3 and x[0] in low_by_nearness]
131    
132    if len(long) != 0 and len(short) != 0:
133        long_w:float = self.Portfolio.TotalPortfolioValue / self.holding_period / len(long)
134        short_w:float = self.Portfolio.TotalPortfolioValue / self.holding_period / len(short)
135        # symbol/quantity collection
136        long_symbol_q:List[Tuple] = [(x, floor(long_w / self.data[x].LastPrice)) for x in long]
137        short_symbol_q:List[Tuple] = [(x, -floor(short_w / self.data[x].LastPrice)) for x in short]
138        
139        self.managed_queue.append(data_tools.RebalanceQueueItem(long_symbol_q + short_symbol_q))
140
141self.last_fine = [x.Symbol for x in selected]
142
143return long + short
144def OnData(self, data: Slice) -> None:
145if not self.selection_flag:
146    return
147self.selection_flag = False
148remove_item = None
149# rebalancing portfolio
150for item in self.managed_queue:
151    if item.holding_period == self.holding_period:
152        # liquidate
153        for symbol, quantity in item.symbol_q:
154            self.MarketOrder(symbol, -quantity)
155                    
156        remove_item = item
157    
158    # trade execution
159    if item.holding_period == 0:
160        open_symbol_q = []
161        
162        for symbol, quantity in item.symbol_q:
163            if data.ContainsKey(symbol):
164                self.MarketOrder(symbol, quantity)
165                open_symbol_q.append((symbol, quantity))
166                    
167        # only opened orders will be closed        
168        item.symbol_q = open_symbol_q
169        
170    item.holding_period += 1
171    
172# we need to remove closed part of portfolio after loop. Otherwise it will miss one item in self.managed_queue
173if remove_item:
174    self.managed_queue.remove(remove_item)
175        
176def Selection(self) -> None:
177self.selection_flag = True
178# https://gist.github.com/wonderbeyond/d293e7a2af1de4873f2d757edd580288
179def rgetattr(self, obj, attr, *args):
180def _getattr(obj, attr):
181    return getattr(obj, attr, *args)
182return reduce(_getattr, [obj] + attr.split('.'))