The dataset consists of all NYSE/AMEX/Nasdaq common stocks. The sample period lasts from May 1981 to May 2018. Following entities are excluded: stocks with share prices less than one dollar at the beginning of the portfolio formation date and firms in the financial sectors.
ASSET CLASS: stocks | REGION: United States | FREQUENCY:
Monthly | MARKET: equities | KEYWORD: Ownership, Equity Factors
I. STRATEGY IN A NUTSHELL
The strategy targets all NYSE, AMEX, and NASDAQ common stocks, excluding firms with share prices below $1 and financial sector firms. Using data from CRSP, Compustat, and Thomson Reuters 13f filings, stocks are sorted into terciles based on quarterly changes in institutional ownership breadth. Within each tercile, stocks are further sorted into quintiles according to 11 anomaly variables: asset growth, failure probability, gross profitability, investment-asset ratio, long-term equity issuance, momentum, net operating assets, net payout, net stock issuance, operating accruals, o-score, and return on assets. The long leg consists of stocks with the highest anomaly scores in the top ownership breadth change tercile, while the short leg consists of stocks with the lowest anomaly scores in the bottom ownership breadth change tercile. Portfolios are value-weighted and rebalanced quarterly, with a two-month lag applied to ensure tradability.
II. ECONOMIC RATIONALE
The strategy exploits signals from informed institutional investors. Changes in ownership breadth indicate entries and exits of well-informed investors, predicting future stock returns. Evidence supports the informed trading hypothesis, as controlling for future earnings surprises removes the predictive power of breadth changes. Short-selling constraints explain only part of the negative alpha for short positions and are otherwise insignificant.
III. SOURCE PAPER
Changes in Ownership Breadth and Capital Market Anomalies. [Click to Open PDF]
Yangru Wu, Rutgers University, Newark – School of Business – Department of Finance & Economics; Weike Xu, Clemson University – Department of Finance
<Abstract>
We investigate how the interaction of entries and exits of informed institutional investors with market anomaly signals affects strategy performance. The long legs of anomalies earn more positive alphas following entries, while the short legs earn more negative alphas following exits. The enhanced anomaly-based strategies of buying stocks in the long legs of anomalies with entries and shorting stocks in the short legs with exits outperform the original anomalies with an increase of 19-54 bps per month in the Fama-French (2015) five-factor alpha. The entries and exits of institutional investors capture informed trading and earnings surprises thereby enhancing the anomalies.
IV. BACKTEST PERFORMANCE
| Annualised Return | 8.34% |
| Volatility | 17.03% |
| Beta | -0.075 |
| Sharpe Ratio | 0.49 |
| Sortino Ratio | N/A |
| Maximum Drawdown | N/A |
| Win Rate | 50% |
V. FULL PYTHON CODE
from AlgorithmImports import *
from typing import Dict, List, Set
from data_tools import QuantpediaHegdeFunds, CustomFeeModel, SymbolData
# endregion
class ChangesInOwnershipBreadthPredictPerformanceOfEquityFactors(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2010, 1, 1)
self.SetCash(100000)
self.leverage:int = 5
self.months_lag:int = 2
self.breadth_selection_quantile:int = 3
self.anomaly_selection_quantile:int = 5
self.mom_period:int = 21
self.total_assets_period:int = 2
self.mean_weights_period:int = 2
self.max_missing_days:int = 3 * 31
self.last_hedge_funds_update:datetime.date|None = None
self.quantities:Dict[Symbol, float] = {}
self.tickers_with_data:Dict[str, List[float]] = []
self.data:Dict[str, SymbolData] = {}
self.exchanges:List[str] = ['NYS', 'NAS', 'ASE']
self.anomalies_symbols:List[str] = ['AG', 'GP', 'MOM', 'ROA', 'NSI', 'NPAY', 'OA']
self.market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
self.hedge_funds_symbol:Symbol = self.AddData(QuantpediaHegdeFunds, 'hedge_funds_holdings.json', Resolution.Daily).Symbol
self.selection_flag:bool = False
self.rebalance_flag:bool = False
self.UniverseSettings.Resolution = Resolution.Daily
self.AddUniverse(self.CoarseSelectionFunction, self.FineSelectionFunction)
def OnSecuritiesChanged(self, changes:SecurityChanges) -> None:
for security in changes.AddedSecurities:
security.SetFeeModel(CustomFeeModel())
security.SetLeverage(self.leverage)
def CoarseSelectionFunction(self, coarse:List[CoarseFundamental]) -> List[Symbol]:
for equity in coarse:
ticker:Symbol = equity.Symbol.Value
if ticker in self.data:
self.data[ticker].update_prices(equity.AdjustedPrice)
if not self.selection_flag:
return Universe.Unchanged
self.selection_flag = False
self.rebalance_flag = True
# select only stocks, which have current and previous mean weight
selected_stocks:List[Symbol] = []
for stock in coarse:
symbol:Symbol = stock.Symbol
ticker:str = symbol.Value
if ticker not in self.tickers_with_data:
continue
# warm up prices if needed
if not self.data[ticker].prices_ready():
history = self.History(symbol, self.mom_period, Resolution.Daily)
if not history.empty:
closes = history.loc[symbol].close
for _, close in closes.iteritems():
self.data[ticker].update_prices(close)
selected_stocks.append(stock.Symbol)
self.tickers_with_data.clear()
return selected_stocks
def FineSelectionFunction(self, fine:List[FineFundamental]) -> List[Symbol]:
curr_date:datetime.date = self.Time.date()
stocks_with_data:List[FineFundamental] = []
for stock in fine:
if stock.MarketCap != 0 and stock.SecurityReference.ExchangeId in self.exchanges and stock.FinancialStatements.BalanceSheet.TotalAssets.ThreeMonths != 0 \
and stock.FinancialStatements.IncomeStatement.GrossProfit.ThreeMonths != 0 and stock.OperationRatios.ROA.ThreeMonths != 0 \
and stock.FinancialStatements.CashFlowStatement.IssuanceOfCapitalStock.ThreeMonths != 0 and stock.FinancialStatements.CashFlowStatement.RepurchaseOfCapitalStock.ThreeMonths != 0 \
and stock.ValuationRatios.TotalYield != 0 and stock.FinancialStatements.CashFlowStatement.CommonStockIssuance.ThreeMonths != 0 \
and stock.FinancialStatements.BalanceSheet.CurrentAssets.Value != 0 and stock.FinancialStatements.BalanceSheet.CurrentLiabilities.Value != 0:
ticker:str = stock.Symbol.Value
stock_data:SymbolData = self.data[ticker]
# make sure data are consecutive
if not stock_data.anomaly_data_still_coming(curr_date, self.max_missing_days):
# each anomaly's data will be reset except momentum
stock_data.reset_anomalies_data()
total_assets:float = stock.FinancialStatements.BalanceSheet.TotalAssets.ThreeMonths
gross_profit:float = stock.FinancialStatements.IncomeStatement.GrossProfit.ThreeMonths
roa:float = stock.OperationRatios.ROA.ThreeMonths
issuance_of_stock:float = stock.FinancialStatements.CashFlowStatement.IssuanceOfCapitalStock.ThreeMonths
repurchase_of_stock:float = stock.FinancialStatements.CashFlowStatement.RepurchaseOfCapitalStock.ThreeMonths
net_stock_issuance:float = issuance_of_stock - repurchase_of_stock
market_cap:float = stock.MarketCap
total_yield:float = stock.ValuationRatios.TotalYield
common_stock_issuance:float = stock.FinancialStatements.CashFlowStatement.CommonStockIssuance.ThreeMonths
net_payout:float = (total_yield * market_cap) / (common_stock_issuance / market_cap)
operating_assets:float = stock.FinancialStatements.BalanceSheet.CurrentAssets.Value
operating_liabilities:float = stock.FinancialStatements.BalanceSheet.CurrentLiabilities.Value
net_operating_assets:float = operating_assets - operating_liabilities
operating_accruals:float = stock_data.update_operating_accruals(net_operating_assets, total_assets) \
if stock_data.net_operating_assets_ready() else None
stock_data.update_net_operating_assets(net_operating_assets)
if operating_accruals != None:
stock_data.update_operating_accruals(operating_accruals)
# update stock's anomalies data
stock_data.update_total_assets_values(total_assets)
stock_data.update_gross_profit(gross_profit)
stock_data.update_roa(roa)
stock_data.update_net_stock_issuance(net_stock_issuance)
stock_data.update_net_payout(net_payout)
stock_data.set_last_anomalies_update(curr_date)
if stock_data.is_ready():
stocks_with_data.append(stock)
# make sure there are enough stocks for both seletion
if len(stocks_with_data) < (self.breadth_selection_quantile * self.anomaly_selection_quantile):
return Universe.Unchanged
# firstly sort stocks based on their mean weight change - breadth
breadth_selection_quantile:int = int(len(stocks_with_data) / self.breadth_selection_quantile)
sorted_by_breadth:List[FineFundamental] = [
x for x in sorted(stocks_with_data, key=lambda stock: self.data[stock.Symbol.Value].get_ownership_breadth())]
low_stocks:List[FineFundamental] = sorted_by_breadth[:breadth_selection_quantile]
high_stocks:List[FineFundamental] = sorted_by_breadth[-breadth_selection_quantile:]
stocks_for_trade:Set = set()
total_anomalies:float = float(len(self.anomalies_symbols))
weight:float = self.Portfolio.TotalPortfolioValue / 2. / total_anomalies
anomalies_data:Dict[str, Dict[FineFundamental, float]] = self.CalculateAnomalies(low_stocks)
# perform short selection and calculate stocks quantities
for anomaly_symbol, stocks_with_values in anomalies_data.items():
sorted_by_anomaly_values:List[FineFundamental] = [x[0] for x in sorted(stocks_with_values.items(), key=lambda item: item[1])]
anomaly_selection_quantile:int = int(len(sorted_by_anomaly_values) / self.anomaly_selection_quantile)
short_leg:List[FineFundamental] = sorted_by_anomaly_values[:anomaly_selection_quantile]
total_cap:float = sum(list(map(lambda stock: stock.MarketCap, short_leg)))
for stock in short_leg:
stock_price:float = self.data[stock.Symbol.Value].get_last_price()
quantity:float = np.floor((weight * (stock.MarketCap / total_cap)) / stock_price)
self.quantities[stock.Symbol] = -quantity
stocks_for_trade.add(stock.Symbol)
# reset anomalies data for next calculation
anomalies_data:Dict[str, Dict[FineFundamental, float]] = self.CalculateAnomalies(high_stocks)
# perform long selection and calculate stocks quantities
for anomaly_symbol, stocks_with_values in anomalies_data.items():
sorted_by_anomaly_values:List[FineFundamental] = [x[0] for x in sorted(stocks_with_values.items(), key=lambda item: item[1])]
anomaly_selection_quantile:int = int(len(sorted_by_anomaly_values) / self.anomaly_selection_quantile)
long_leg:List[FineFundamental] = sorted_by_anomaly_values[-anomaly_selection_quantile:]
total_cap:float = sum(list(map(lambda stock: stock.MarketCap, long_leg)))
for stock in long_leg:
stock_price:float = self.data[stock.Symbol.Value].get_last_price()
quantity:float = np.floor((weight * (stock.MarketCap / total_cap)) / stock_price)
self.quantities[stock.Symbol] = quantity
stocks_for_trade.add(stock.Symbol)
return list(stocks_for_trade)
def OnData(self, data:Slice) -> None:
curr_date:datetime.date = self.Time.date()
if self.hedge_funds_symbol in data and data[self.hedge_funds_symbol]:
stocks_data:Dict[str, List[float]] = data[self.hedge_funds_symbol].GetProperty('stocks_with_weights')
for ticker, weights in stocks_data.items():
mean_weight:float = np.mean(weights)
if ticker not in self.data:
self.data[ticker] = SymbolData(self.mom_period, self.total_assets_period, self.mean_weights_period)
# make sure mean weight values are consecutive
if not self.data[ticker].weight_data_still_coming(curr_date, self.max_missing_days):
self.data[ticker].reset_weights()
self.data[ticker].update_mean_weights(curr_date, mean_weight)
# stock will be selected in CoarseSelectionFuction only, if it has mean values ready
if self.data[ticker].mean_weight_values_ready():
self.tickers_with_data.append(ticker)
self.last_hedge_funds_update = curr_date
self.selection_flag = True
if self.last_hedge_funds_update != None and (curr_date - self.last_hedge_funds_update).days > self.max_missing_days:
# liquidate portfolio, when data stop coming
self.Liquidate()
if not self.rebalance_flag:
return
self.rebalance_flag = False
self.Liquidate()
for symbol, quantity in self.quantities.items():
if self.Securities[symbol].IsTradable and self.Securities[symbol].Price != 0:
self.MarketOrder(symbol, quantity)
self.quantities.clear()
def CalculateAnomalies(self, stocks:List[FineFundamental]) -> Dict[str, Dict[FineFundamental, float]]:
anomalies_data:Dict[str, Dict[FineFundamental, float]] = { anomaly_symbol: {} for anomaly_symbol in self.anomalies_symbols }
for stock in stocks:
ticker:str = stock.Symbol.Value
stock_data:SymbolData = self.data[ticker]
anomalies_data['AG'][stock] = stock_data.get_asset_growth()
anomalies_data['GP'][stock] = stock_data.get_gross_profit()
anomalies_data['MOM'][stock] = stock_data.get_momentum()
anomalies_data['ROA'][stock] = stock_data.get_roa()
anomalies_data['NSI'][stock] = stock_data.get_net_stock_issuance()
anomalies_data['NPAY'][stock] = stock_data.get_net_payout()
anomalies_data['OA'][stock] = stock_data.get_operating_accruals()
return anomalies_data
