“该策略涵盖NYSE/AMEX/Nasdaq上市的普通股，数据来自CRSP、Compustat和Thomson Reuters Institutional（13f）。首先，股票按持股广度变化分为三组：持股广度变化最大的为“LOW”，最小的为“HIGH”。然后，依据11个异常变量（如资产增长、动量、资产回报率等），在每组内将股票排序为五分位。策略在异常变量多头股票的“HIGH”组做多，异常变量空头股票的“LOW”组做空。投资组合为价值加权，并预留两个月时间差确保可交易性。”
资产类别：股票 | 区域：美国 | 频率：每月 | 市场：股权 | 关键词：持股，股票因子
策略概述
数据集包含所有在NYSE/AMEX/Nasdaq上市的普通股。样本期从1981年5月到2018年5月。以下公司被排除：在投资组合形成日期前股价低于一美元的股票和金融行业的公司。数据来自多个来源：会计数据来自CRSP/Compustat合并数据库的年度和季度文件，股票回报数据来自CRSP月度文件，机构投资者持股数据来自Thomson Reuters Institutional（13f）持股S34文件。对于存在申报日期与报告日期差异的情况，使用CRSP累积股票和价格调整因子来逆转13f的拆股调整。机构投资者持股数据与CRSP和CRSP/Compustat通过CUSIP编号和报告日期合并。为了确保策略的可交易性，机构持股数据与投资组合形成之间预留两个月的时间差。
然后，根据持股广度变化将所有股票按季度变化分为三分位投资组合。前一季度持股广度变化最大（最负）的股票属于最低组（LOW），持股广度变化最小（最正）的股票属于最高组（HIGH）。接下来，根据11个异常变量中的每一个，在每个广度变化组内对股票进行排序并将其分为五分位投资组合。这些变量包括资产增长（AG）、失败概率（FP）、毛利率（GP）、投资资产（IA）、长期股票发行（LSI）、动量（MOM）、净运营资产（NOA）、净派息（NPAY）、净股票发行（NSI）、经营应计（OA）、o分数（OS）、资产回报率（ROA）。根据改进的策略，在异常变量多头的股票和持股广度变化最高的三分位投资组合中买入股票；在异常变量空头的股票和持股广度变化最低的三分位投资组合中做空股票。投资组合为价值加权。
策略合理性
为了更好地理解基于异常的策略来源，研究作者测试了两种可能的解释：知情交易假说和卖空限制假说。根据知情交易假说，知情的机构投资者可能掌握有关未来盈利意外的私人信息。持股广度的变化（即持有股票的投资者数量的变化）被视为未来价格变化的指标。机构投资者的进入和退出可以预测未来的股票回报。研究支持了这一理论，测试并确认了控制未来盈利意外后广度变化的预测能力消失的假设。第二种假说认为，这种预测能力可以通过卖空限制来解释。然而，它只能解释持股广度变化中空头部分的负阿尔法，而不能解释多头部分的正阿尔法。此外，研究还提供了其不显著性的证据。
论文来源
Changes in Ownership Breadth and Capital Market Anomalies [点击浏览原文]
- 杨儒武，罗格斯大学纽瓦克分校商学院金融与经济系
- 许伟克，克莱姆森大学金融系
<摘要>
我们研究了知情机构投资者的进入和退出与市场异常信号的相互作用如何影响策略表现。异常多头部分在机构投资者进入后获得了更多的正阿尔法，而空头部分在机构投资者退出后获得了更多的负阿尔法。通过买入异常多头且机构投资者进入的股票、卖出异常空头且机构投资者退出的股票，增强型的基于异常的策略表现优于原始策略，每月Fama-French（2015）五因子阿尔法增加19-54个基点。机构投资者的进入和退出捕捉了知情交易和盈利意外，从而增强了这些异常。
回测表现
| 年化收益率 | 8.34% |
| 波动率 | 17.03% |
| Beta | -0.075 |
| 夏普比率 | 0.49 |
| 索提诺比率 | N/A |
| 最大回撤 | N/A |
| 胜率 | 50% |
完整python代码
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
