FSCORE-Enhanced Short-Term Reversal Strategy on Large-Cap Stocks

1from AlgoLib import *
2from typing import List, Dict
3from numpy import floor, isnan
4from functools import reduce
5from pandas.core.frame import DataFrame
6import data_tools
7
8class CombiningFSCOREShortTermReversals(XXX):
9
10def Initialize(self):
11self.SetStartDate(2000, 1, 1)
12self.SetCash(100000)
13
14self.exchange_codes:List[str] = ['NYS', 'NAS', 'ASE']
15
16self.financial_statement_names:List[str] = [
17    'EarningReports.BasicAverageShares.ThreeMonths',
18    'EarningReports.BasicEPS.TwelveMonths',
19    'OperationRatios.ROA.ThreeMonths',
20    'OperationRatios.GrossMargin.ThreeMonths',
21    'FinancialStatements.CashFlowStatement.CashFlowFromContinuingOperatingActivities.ThreeMonths',
22    'FinancialStatements.IncomeStatement.NormalizedIncome.ThreeMonths',
23    'FinancialStatements.BalanceSheet.LongTermDebt.ThreeMonths',
24    'FinancialStatements.BalanceSheet.TotalAssets.ThreeMonths',
25    'FinancialStatements.BalanceSheet.OrdinarySharesNumber.ThreeMonths',
26    'FinancialStatements.IncomeStatement.TotalRevenueAsReported.ThreeMonths',
27    'ValuationRatios.PERatio',
28    'OperationRatios.CurrentRatio.ThreeMonths',
29]
30
31self.SetSecurityInitializer(lambda x: x.SetMarketPrice(self.GetLastKnownPrice(x)))
32
33self.leverage:int = 10
34self.min_share_price:int = 5
35self.period = 21
36
37self.long:List[Symbol] = []
38self.short:List[Symbol] = []
39
40self.stock_data:Dict[Symbol, data_tools.StockData] = {}
41self.data:Dict[Symbol, data_tools.SymbolData] = {}
42
43market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
44
45self.fundamental_count:int = 3000
46self.fundamental_sorting_key = lambda x: x.MarketCap
47
48self.selection_flag = False
49self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
50self.UniverseSettings.Resolution = Resolution.Daily
51self.AddUniverse(self.FundamentalSelectionFunction)
52self.Schedule.On(self.DateRules.MonthEnd(market), self.TimeRules.AfterMarketOpen(market), self.Selection)
53   
54def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
55for security in changes.AddedSecurities:
56    security.SetFeeModel(data_tools.CustomFeeModel())
57    security.SetLeverage(self.leverage)
58
59def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
60# Update the rolling window every day.
61for stock in fundamental:
62    symbol:Symbol = stock.Symbol
63
64    # Store daily price.
65    if symbol in self.data:
66        self.data[symbol].update(stock.AdjustedPrice)
67
68if not self.selection_flag:
69    return Universe.Unchanged
70
71selected:List[Fundamental] = [x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and x.AdjustedPrice > self.min_share_price \
72    and all((not isnan(self.rgetattr(x, statement_name)) and self.rgetattr(x, statement_name) != 0) for statement_name in self.financial_statement_names) \
73    and x.SecurityReference.ExchangeId in self.exchange_codes]
74if len(selected) > self.fundamental_count:
75    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
76
77# Warmup price rolling windows.
78for stock in selected:
79    symbol:Symbol = stock.Symbol
80
81    if symbol in self.data:
82        continue
83    
84    self.data[symbol] = data_tools.SymbolData(symbol, self.period)
85    history:DataFrame = self.History(symbol, self.period, Resolution.Daily)
86    if history.empty:
87        self.Log(f"Not enough data for {symbol} yet.")
88        continue
89    closes:Series = history.loc[symbol].close
90    for time, close in closes.items():
91        self.data[symbol].update(close)                  
92
93# BM sorting
94sorted_by_market_cap:List[Fundamental] = sorted(selected, key = lambda x: x.MarketCap, reverse = True)
95lenght:int = int((len(sorted_by_market_cap) / 100) * 40)
96top_by_market_cap:List[Fundamental] = [x for x in sorted_by_market_cap[:lenght]]
97
98fine_symbols:List[Symbol] = [x.Symbol for x in top_by_market_cap]
99
100score_performance:Dict[Symbol, Tuple[float]] = {}
101
102for stock in top_by_market_cap:
103    symbol:Symbol = stock.Symbol
104    
105    if not self.data[symbol].is_ready():
106        continue
107
108    if symbol not in self.stock_data:
109        self.stock_data[symbol] = data_tools.StockData()   # Contains latest data.
110        
111    roa:float = stock.OperationRatios.ROA.ThreeMonths
112    cfo:float = stock.FinancialStatements.CashFlowStatement.CashFlowFromContinuingOperatingActivities.ThreeMonths
113    leverage:float = stock.FinancialStatements.BalanceSheet.LongTermDebt.ThreeMonths / stock.FinancialStatements.BalanceSheet.TotalAssets.ThreeMonths
114    liquidity:float = stock.OperationRatios.CurrentRatio.ThreeMonths
115    equity_offering:float = stock.FinancialStatements.BalanceSheet.OrdinarySharesNumber.ThreeMonths
116    gross_margin:float = stock.OperationRatios.GrossMargin.ThreeMonths
117    turnover:float = stock.FinancialStatements.IncomeStatement.TotalRevenueAsReported.ThreeMonths / stock.FinancialStatements.BalanceSheet.TotalAssets.ThreeMonths
118    
119    # Check if data has previous year's data ready.
120    stock_data = self.stock_data[symbol]
121    if (stock_data.ROA == 0) or (stock_data.Leverage == 0) or (stock_data.Liquidity == 0) or (stock_data.Equity_offering == 0) or (stock_data.Gross_margin == 0) or (stock_data.Turnover == 0):
122        stock_data.Update(roa, leverage, liquidity, equity_offering, gross_margin, turnover)
123        continue
124
125    score:int = 0
126
127    if roa > 0:
128        score += 1
129    if cfo > 0:
130        score += 1
131    if roa > stock_data.ROA:   # ROA change is positive
132        score += 1
133    if cfo > roa:
134        score += 1
135    if leverage < stock_data.Leverage:
136        score += 1
137    if liquidity > stock_data.Liquidity:
138        score += 1
139    if equity_offering < stock_data.Equity_offering:
140        score += 1
141    if gross_margin > stock_data.Gross_margin:
142        score += 1
143    if turnover > stock_data.Turnover:
144        score += 1
145    
146    score_performance[symbol] = (score, self.data[symbol].performance())
147    
148    # Update new (this year's) data.
149    stock_data.Update(roa, leverage, liquidity, equity_offering, gross_margin, turnover)
150
151# Clear out not updated data.
152for symbol in self.stock_data:
153    if symbol not in fine_symbols:
154        self.stock_data[symbol] = data_tools.StockData()
155
156# Performance sorting and F score sorting.
157self.long = [x[0] for x in score_performance.items() if x[1][0] >= 7 and x[1][1] < 0]
158self.short = [x[0] for x in score_performance.items() if x[1][0] <= 3 and x[1][1] > 0]
159
160return self.long + self.short
161
162def OnData(self, data: Slice) -> None:
163if not self.selection_flag:
164    return
165self.selection_flag = False
166
167# Trade execution.
168targets:List[PortfolioTarget] = []
169for i, portfolio in enumerate([self.long, self.short]):
170    for symbol in portfolio:
171        if symbol in data and data[symbol]:
172            targets.append(PortfolioTarget(symbol, ((-1) ** i) / len(portfolio)))
173
174self.SetHoldings(targets, True)
175
176self.long.clear()
177self.short.clear()
178
179def Selection(self) -> None:
180self.selection_flag = True
181
182# https://gist.github.com/wonderbeyond/d293e7a2af1de4873f2d757edd580288
183def rgetattr(self, obj, attr, *args):
184def _getattr(obj, attr):
185    return getattr(obj, attr, *args)
186return reduce(_getattr, [obj] + attr.split('.'))