The strategy ranks stocks into deciles, trades the best and worst deciles, uses MA50/MA200 signals to refine positions monthly, and rebalances portfolios annually in June.
ASSET CLASS: stocks | REGION: Global | FREQUENCY:
Monthly | MARKET: equities | KEYWORD: Classical Equity , Anomalies Combined, Trendfollowing Filter
I. STRATEGY IN A NUTSHELL: Global Equity Anomaly Decile Strategy with Trend Filter
This global equity strategy ranks all stocks annually in June using eight accounting variables from the prior fiscal year, assigning them to deciles. Equal-weighted portfolios are formed, with a long-short spread: long the top decile (Decile High) and short the bottom decile (Decile Low). Rebalancing occurs annually. A monthly trend filter is applied using 50- and 200-day moving averages (MA50 and MA200): stocks in Decile High are removed if MA50 < MA200, and stocks in Decile Low are removed if MA50 > MA200, refining positions based on higher-frequency price information.
II. ECONOMIC RATIONALE
Equity anomalies often rely on low-frequency accounting data, leading to annual rebalancing. Using high-frequency price trends via moving averages allows investors to update portfolio views, improving performance by reacting to more timely market information while maintaining the low-frequency signal.
III. SOURCE PAPER
Anomalies Enhanced: A Portfolio Rebalancing Approach[Click to Open PDF]
Yufeng Han,Dayong Huang and Guofu Zhou.University of North Carolina (UNC) at Charlotte – Finance.University of North Carolina (UNC) at Greensboro – Bryan School of Business & Economics.Washington University in St. Louis – John M. Olin Business School.
<Abstract>
Many anomalies are based on firm characteristics and are rebalanced yearly, ignoring any information during the year. In this paper, we provide dynamic trading strategies to rebalance the anomaly portfolios monthly. For eight major anomalies, we find that these dynamic trading strategies substantially enhance their economic importance, with improvements in the Fama and French (2015) five-factor risk-adjusted abnormal return ranging from 0.40% to 0.75% per month. The results are robust to a number of controls. Our findings indicate that many well known anomalies are more profitable than previously thought, yielding new challenges for their theoretical explanations.
IV. BACKTEST PERFORMANCE
| Annualised Return | 16.18% |
| Volatility | 15.48% |
| Beta | 0.05 |
| Sharpe Ratio | 1.05 |
| Sortino Ratio | -0.008 |
| Maximum Drawdown | N/A |
| Win Rate | 43% |
V. FULL PYTHON CODE
import numpy as np
from AlgorithmImports import *
from functools import reduce
from numpy import isnan
from pandas.core.frame import dataframe
class ClassicalEquityAnomaliesCombinedwithTrendfollowingFilter(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2000, 1, 1)
self.SetCash(100000)
market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
self.financial_statement_names:List[str] = [
'ValuationRatios.PBRatio',
'FinancialStatements.IncomeStatement.GrossProfit.TwelveMonths',
'FinancialStatements.IncomeStatement.TotalOperatingIncomeAsReported.TwelveMonths',
'OperationRatios.TotalAssetsGrowth.OneYear',
'FinancialStatements.CashFlowStatement.CapitalExpenditure.TwelveMonths',
'EarningReports.BasicAverageShares.TwelveMonths',
# accruals data
'FinancialStatements.BalanceSheet.CurrentAssets.TwelveMonths',
'FinancialStatements.BalanceSheet.CashAndCashEquivalents.TwelveMonths',
'FinancialStatements.BalanceSheet.CurrentLiabilities.TwelveMonths',
'FinancialStatements.BalanceSheet.CurrentDebt.TwelveMonths',
'FinancialStatements.BalanceSheet.IncomeTaxPayable.TwelveMonths',
'FinancialStatements.IncomeStatement.DepreciationAndAmortization.TwelveMonths',
'FinancialStatements.BalanceSheet.TotalAssets.TwelveMonths'
]
# Daily sma data.
self.data:Dict[Symbol, SymbolData] = {}
self.short_period:int = 50
self.long_period:int = 200
self.performance_period:int = 12*21
self.quantile:int = 10
self.leverage:int = 5
self.long:List[Symbol] = []
self.short:List[Symbol] = []
self.fundamental_count:int = 1000
self.fundamental_sorting_key = lambda x: x.DollarVolume
self.selection_flag:bool = False
self.UniverseSettings.Resolution = Resolution.Daily
self.AddUniverse(self.FundamentalSelectionFunction)
self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
self.Schedule.On(self.DateRules.MonthEnd(market), self.TimeRules.BeforeMarketClose(market), self.Selection)
self.settings.daily_precise_end_time = False
def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
for security in changes.AddedSecurities:
security.SetFeeModel(CustomFeeModel())
security.SetLeverage(self.leverage)
def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
# Update SMA every day.
for stock in fundamental:
symbol:Symbol = stock.Symbol
# Store monthly price.
if symbol in self.data:
self.data[symbol].update(self.Time, stock.AdjustedPrice)
if not self.selection_flag:
return Universe.Unchanged
selected:List[Fundamental] = [
x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and \
all((not isnan(self.rgetattr(x, statement_name)) and self.rgetattr(x, statement_name) != 0) for statement_name in self.financial_statement_names)
]
if len(selected) > self.fundamental_count:
selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
# Ranked stocks.
rank = {}
current_accruals_data = {}
acc = {}
momentum = {}
net_stock_issue = {}
capex = {}
noa = {}
selected_stocks:List[Fundamental] = []
# Warmup price rolling windows.
for stock in selected:
symbol:Symbol = stock.Symbol
if symbol not in self.data:
self.data[symbol] = SymbolData(self.short_period, self.long_period, self.performance_period)
history:dataframe = self.History(symbol, self.performance_period, Resolution.Daily)
if history.empty:
self.Log(f"Not enough data for {symbol} yet.")
continue
closes:pd.Series = history.loc[symbol].close
for time, close in closes.items():
self.data[symbol].update(time, close)
if self.data[symbol].is_ready():
selected_stocks.append(stock)
rank[symbol] = 0
momentum[symbol] = self.data[symbol].performance()
# Accural calc
current_accruals_data = AccrualsData(stock.FinancialStatements.BalanceSheet.CurrentAssets.TwelveMonths, stock.FinancialStatements.BalanceSheet.CashAndCashEquivalents.TwelveMonths,
stock.FinancialStatements.BalanceSheet.CurrentLiabilities.TwelveMonths, stock.FinancialStatements.BalanceSheet.CurrentDebt.TwelveMonths, stock.FinancialStatements.BalanceSheet.IncomeTaxPayable.TwelveMonths,
stock.FinancialStatements.IncomeStatement.DepreciationAndAmortization.TwelveMonths, stock.FinancialStatements.BalanceSheet.TotalAssets.TwelveMonths)
if self.data[symbol].accruals_data:
acc[symbol] = self.CalculateAccruals(current_accruals_data, self.data[symbol].accruals_data)
# Calculate NOA.
noa[symbol] = (stock.FinancialStatements.BalanceSheet.CurrentAssets.TwelveMonths - stock.FinancialStatements.BalanceSheet.CurrentLiabilities.TwelveMonths) / self.data[symbol].accruals_data.TotalAssets
# Calculate stock stock issue.
num_of_shares = stock.EarningReports.BasicAverageShares.TwelveMonths
if self.data[symbol].shares_outstanding:
net_stock_issue[symbol] = num_of_shares / self.data[symbol].shares_outstanding - 1
self.data[symbol].shares_outstanding = num_of_shares
# Calculate stock capital expenditure.
cap_expenditures = stock.FinancialStatements.CashFlowStatement.CapitalExpenditure.TwelveMonths
if self.data[symbol].capex:
capex[symbol] = cap_expenditures / self.data[symbol].capex - 1
self.data[symbol].capex = cap_expenditures
# Update accruals data for next year's calculation.
self.data[symbol].accruals_data = current_accruals_data
# Ensure that consecutive accruals and shares data are available.
for symbol in self.data:
if symbol not in list(map(lambda x: x.Symbol, selected_stocks)):
self.data[symbol].accruals_data = None
self.data[symbol].shares_outstanding = None
self.data[symbol].capex = None
# Sort by book to market.
sorted_by_bm = sorted(selected_stocks, key=lambda x: 1 / x.ValuationRatios.PBRatio)
for index, stock in enumerate(sorted_by_bm):
rank[stock.Symbol] += (index+1)
# Sort by gross profit.
sorted_by_gp = sorted(selected_stocks, key=lambda x: x.FinancialStatements.IncomeStatement.GrossProfit.TwelveMonths / x.FinancialStatements.BalanceSheet.TotalAssets.TwelveMonths)
for index, stock in enumerate(sorted_by_gp):
rank[stock.Symbol] += (index+1)
# Sort by operating profit.
sorted_by_op = sorted(selected_stocks, key=lambda x: x.FinancialStatements.IncomeStatement.TotalOperatingIncomeAsReported.TwelveMonths)
for index, stock in enumerate(sorted_by_op):
rank[stock.Symbol] += (index+1)
# Sort by asset growth.
sorted_by_ag = sorted(selected_stocks, key=lambda x: x.OperationRatios.TotalAssetsGrowth.OneYear, reverse = True)
for index, stock in enumerate(sorted_by_ag):
rank[stock.Symbol] += (index+1)
# Sort by investments.
sorted_by_inv = sorted(capex.items(), key=lambda x: x[1], reverse = True)
for index, symbol_si in enumerate(sorted_by_inv):
rank[symbol_si[0]] += (index+1)
# Sort by net stock issue.
sorted_by_si = sorted(net_stock_issue.items(), key=lambda x: x[1], reverse = True)
for index, symbol_si in enumerate(sorted_by_si):
rank[symbol_si[0]] += (index+1)
# Sort by accruals.
sorted_by_acc = sorted(acc.items(), key=lambda x: x[1])
for index, symbol_acc in enumerate(sorted_by_acc):
rank[symbol_acc[0]] += (index+1)
# Sort by NOA.
sorted_by_noa = sorted(noa.items(), key=lambda x: x[1], reverse = True)
for index, symbol_noa in enumerate(sorted_by_noa):
rank[symbol_noa[0]] += (index+1)
if len(rank) > self.quantile:
# Rank sorting.
sorted_by_rank:List = sorted(rank.items(), key=lambda x: x[1], reverse = True)
quantile:int = int(len(sorted_by_rank) / self.quantile)
top_decile:List[Symbol] = [x[0] for x in sorted_by_rank[:quantile]]
bottom_decile:List[Symbol] = [x[0] for x in sorted_by_rank[-quantile:]]
top_decile_perf:float = np.mean([momentum[x] for x in top_decile])
bottom_decile_perf:float = np.mean([momentum[x] for x in bottom_decile])
if top_decile_perf > bottom_decile_perf:
self.long = top_decile
self.short = bottom_decile
else:
self.long = bottom_decile
self.short = top_decile
return self.long + self.short
def OnData(self, data: Slice) -> None:
if not self.selection_flag:
return
self.selection_flag = False
# order execution
targets:List[PortfolioTarget] = []
for i, portfolio in enumerate([self.long, self.short]):
for symbol in portfolio:
if symbol in data and data[symbol]:
targets.append(PortfolioTarget(symbol, ((-1) ** i) / len(portfolio)))
self.SetHoldings(targets, True)
self.long.clear()
self.short.clear()
def Selection(self) -> None:
# Drop stocks according to trend filter.
symbols_invested:List[Symbol] = [x.Key for x in self.Portfolio if x.Value.Invested]
for symbol in symbols_invested:
if self.Portfolio[symbol].IsLong:
if not self.data[symbol].is_in_uptrend():
self.Liquidate(symbol)
elif self.Portfolio[symbol].IsShort:
if self.data[symbol].is_in_uptrend():
self.Liquidate(symbol)
if self.Time.month == 6:
self.selection_flag = True
def CalculateAccruals(self, current_accrual_data, prev_accrual_data) -> float:
delta_assets = current_accrual_data.CurrentAssets - prev_accrual_data.CurrentAssets
delta_cash = current_accrual_data.CashAndCashEquivalents - prev_accrual_data.CashAndCashEquivalents
delta_liabilities = current_accrual_data.CurrentLiabilities - prev_accrual_data.CurrentLiabilities
delta_debt = current_accrual_data.CurrentDebt - prev_accrual_data.CurrentDebt
delta_tax = current_accrual_data.IncomeTaxPayable - prev_accrual_data.IncomeTaxPayable
dep = current_accrual_data.DepreciationAndAmortization
avg_total = (current_accrual_data.TotalAssets + prev_accrual_data.TotalAssets) / 2
bs_acc = ((delta_assets - delta_cash) - (delta_liabilities - delta_debt-delta_tax) - dep) / avg_total
return bs_acc
# https://gist.github.com/wonderbeyond/d293e7a2af1de4873f2d757edd580288
def rgetattr(self, obj, attr, *args):
def _getattr(obj, attr):
return getattr(obj, attr, *args)
return reduce(_getattr, [obj] + attr.split('.'))
class AccrualsData():
def __init__(self, current_assets, cash_and_cash_equivalents, current_liabilities, current_debt, income_tax_payable, depreciation_and_amortization, total_assets):
self.CurrentAssets = current_assets
self.CashAndCashEquivalents = cash_and_cash_equivalents
self.CurrentLiabilities = current_liabilities
self.CurrentDebt = current_debt
self.IncomeTaxPayable = income_tax_payable
self.DepreciationAndAmortization = depreciation_and_amortization
self.TotalAssets = total_assets
class SymbolData():
def __init__(self, short_sma_period, long_sma_period, performance_period):
self.price = RollingWindow[float](performance_period)
self.short_sma = SimpleMovingAverage(short_sma_period)
self.long_sma = SimpleMovingAverage(long_sma_period)
self.accruals_data = None
self.shares_outstanding = None
self.capex = None
def update(self, time: datetime, value: float):
self.short_sma.Update(time, value)
self.long_sma.Update(time, value)
self.price.Add(value)
def is_ready(self) -> bool:
return (self.short_sma.IsReady and self.long_sma.IsReady and self.price.IsReady)
def is_in_uptrend(self) -> bool:
return (self.short_sma.Current.Value > self.long_sma.Current.Value)
def performance(self) -> float:
return self.price[0] / self.price[self.price.Count - 1] - 1
# Custom fee model.
class CustomFeeModel(FeeModel):
def GetOrderFee(self, parameters):
fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
return OrderFee(CashAmount(fee, "USD"))