机构持股效应

“该策略投资于NYSE、AMEX和NASDAQ的股票，根据机构持股周转率进行交易。对周转率最低的十分位组合做多，对周转率最高的十分位组合做空。投资组合采用等权重配置，并每季度重新平衡。”

资产类别：股票 | 地区：美国 | 频率：季度 | 市场：股票市场 | 关键词：机构

I. 策略概述

该策略专注于NYSE、AMEX和NASDAQ中股价高于1美元的股票，每季度根据机构持股周转率（InstTurnIndiv）对股票进行分类。InstTurnIndiv基于13f文件数据计算，为机构持股在过去八个季度的绝对变动，取平均值后除以机构总持股量。根据InstTurnIndiv将股票分为十分位，对周转率最低的十分位（机构持股最稳定的股票）做多，对周转率最高的十分位（机构持股周转率最高的股票）做空。投资组合按等权重配置，并每季度重新平衡，以利用持股稳定性对收益的影响。

II. 策略合理性

学术研究表明，机构投资者凭借其信息优势进行交易，这导致从机构向散户或个体投资者转移的大量资金流动会使相关股票的预期回报下降。持股周转率较高的股票往往伴随着较大的价格波动和溢价，而这些溢价在未来会回归正常，导致低于平均水平的回报率。通过投资持股更稳定的股票，该策略旨在捕捉较高的风险调整后回报。

III. 论文来源

Do Institutions Pay to Play? Turnover of Institutional Ownership and Stock Returns [点击浏览原文]

• Dimitrov, Gatchev

<摘要>

我们研究了股票收益与机构持股周转率之间的关系。基于十个投资组合的分析发现，机构持股周转率最高的股票组合在随后一年中的收益比最低的持股周转率组合低8.9%。这些发现与Harrison和Kreps（1978）及Scheinkman和Xiong（2003）的理论一致，即交易利润的预期导致高持股周转率和资产价格溢价。

进一步分析表明，机构与个体投资者之间的交易导致的持股周转率是股票未来收益的负面影响主要来源，而机构之间的交易对收益的影响较小。这与机构在与个体投资者交易时预期更高利润的观点一致。对于流动性较低或信息不对称较强的股票，这种负相关关系更加显著。

IV. 回测表现

年化收益率	15%
波动率	20.1%
Beta	-0.022
夏普比率	0.71
索提诺比率	-0.016
最大回撤	N/A
胜率	49%

V. 完整python代码

from AlgorithmImports import *
#endregion
class InstitutionalOwnershipEffect(QCAlgorithm):
    def Initialize(self):
        self.SetStartDate(2005, 1, 1)
        self.SetCash(100_000)
        
        self.dates: List[datetime.date] = []
        self.institutional_ownership: Dict[Symbol, Dict] = {}
        
        self.data: Dict[Symbol, SymbolData] = {}
        self.period: int = 3 * 8 * 21 # Eight quarter period
        self.one_quarter_data: Dict[Symbol, List[float]] = {} # Storing one quarter data about each stock from our universe
        self.eight_quarter_data: Dict[Symbol, RollingWindow] = {} # Storing eight quarter data about each stock from our universe
        
        self.quantile: int = 10
        self.leverage: int = 5
        self.last_investment_date = None
        self.symbol: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
        
        self.sp100_stocks: List[Symbol] = [] # 'AGN', 'UTX', 'BRKB' # No data about institutional ownership
        
        csv_string_file: str = self.Download('data.quantpedia.com/backtesting_data/economic/institutional_ownership/institutional_ownership_in_millions.csv')
        lines: List[str] = csv_string_file.split('\r\n')
        for i in range(len(lines)):
            line_split: List[str] = lines[i].split(',')
            if i == 0: # First row are headers of columns
                for j in range(1, len(line_split)): # Take all headers, which are stock tickers
                    ticker: str = line_split[j]
                    symbol: Symbol = self.AddEquity(ticker, Resolution.Daily).Symbol
                    
                    # Warmup volume data.
                    self.data[symbol] = SymbolData(self.period)
                    history: dataframe = self.History(symbol, self.period, Resolution.Daily)
                    if history.empty:
                        self.Log(f"Not enough data for {symbol} yet")
                        continue
                    volumes: Series = history.loc[symbol].volume
                    for time, volume in volumes.items():
                        self.data[symbol].update(volume)
                    
                    self.one_quarter_data[symbol] = []
                    self.eight_quarter_data[symbol] = RollingWindow[float](8)
                    self.institutional_ownership[symbol] = {} # Create dictionary for all institutional ownership data
                    
                    # Subscript current stock and append symbol to universe, which we will use for this strategy 
                    self.sp100_stocks.append(symbol)
            else:
                date: datetime.date = datetime.strptime(line_split[0], "%Y-%m-%d").date()
                
                self.dates.append(date) # Create list of dates for institutional annoucements
                
                for j, symbol in zip(range(1, len(line_split)), self.sp100_stocks):
                    # Based on symbol store date and institutional ownership value as a nested dictionary
                    # In nested dictionary date figures as a key and institutional ownership as a value
                    self.institutional_ownership[symbol][date] = (line_split[j]) 
        
        self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
        self.count_month: int = 1
        self.selection_flag: bool = False
        self.Schedule.On(self.DateRules.MonthEnd(self.symbol), self.TimeRules.AfterMarketOpen(self.symbol), self.Selection)
        
    def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
        for security in changes.AddedSecurities:
            security.SetFeeModel(CustomFeeModel())
            security.SetLeverage(self.leverage)
    def OnData(self, slice: Slice) -> None:
        date = None
        inst_turn_in_div: Dict[Symbol, float] = {}
        
        current_date: datetime.date = self.Time.date()
        day_before: datetime.date = (self.Time - timedelta(days=1)).date()
        two_days_before: datetime.date = (self.Time - timedelta(days=2)).date()
        three_days_before: datetime.date = (self.Time - timedelta(days=3)).date()
        
        for symbol in self.sp100_stocks:
            # Update daily volume data.
            if symbol in slice:
                if slice[symbol]:
                    self.data[symbol].update(slice[symbol].Volume)
            
            # Institutional ownership could be annouced on weekend. This prevents missing it.
            # Without three days look ahead it was missing some annoucements
            if (current_date in self.dates) and (current_date != self.last_investment_date):
                self.last_investment_date = current_date
                date = current_date
            elif (day_before in self.dates) and (day_before != self.last_investment_date):
                self.last_investment_date = day_before
                date = day_before
            elif (two_days_before in self.dates) and (two_days_before != self.last_investment_date):
                self.last_investment_date = two_days_before
                date = two_days_before
            elif (three_days_before in self.dates) and (three_days_before != self.last_investment_date):
                self.last_investment_date = three_days_before
                date = three_days_before
                
            # If institutional ownership was announced store institutional_ownership value about each stock from our universe
            if date in self.dates:
                if self.institutional_ownership[symbol][date] != '':
                    self.one_quarter_data[symbol].append(float(self.institutional_ownership[symbol][date]))
        
            if not self.selection_flag:
                continue
            absolute_change = None
            if len(self.one_quarter_data[symbol]) >= 2:
                absolute_change: float = self.one_quarter_data[symbol][-1] - self.one_quarter_data[symbol][0]
                total_one_quarter: float = sum(self.one_quarter_data[symbol])
                self.eight_quarter_data[symbol].Add(total_one_quarter)
        
                # If eight quarter institutional ownership data are ready for chosen stock, then trade
                if self.data[symbol].is_ready() and self.eight_quarter_data[symbol].IsReady:  
                    if symbol != 'ADP':
                        inst_turn_in_div_num: float = absolute_change / self.data[symbol].total_volume()
                        
                        total_inst_shares: float = sum([x for x in self.eight_quarter_data[symbol]])
                        inst_turn_in_div_num: float = inst_turn_in_div_num / total_inst_shares
                        
                        inst_turn_in_div[symbol] = inst_turn_in_div_num
            
            self.one_quarter_data[symbol].clear()
        
        if not self.selection_flag:
            return
        self.selection_flag = False
        
        long: List[Symbol] = []
        short: List[Symbol] = []
        if len(inst_turn_in_div) >= self.quantile:
            quantile: int = int(len(inst_turn_in_div) / self.quantile)
            sorted_by_inst_turn_in_div = [x[0] for x in sorted(inst_turn_in_div.items(), key=lambda item: item[1])]
            # long the lowest ‘InstTurnIndiv’ decile
            long = sorted_by_inst_turn_in_div[:quantile]
            # short the highest ‘InstTurnIndiv’ decile
            short = sorted_by_inst_turn_in_div[-quantile:]
        
        # Trade execution.
        invested: List[Symbol] = [x.Key for x in self.Portfolio if x.Value.Invested]
        for symbol in invested:
            if symbol not in long + short:
                self.Liquidate(symbol)
            
        for symbol in long:
            self.SetHoldings(symbol, 1 / len(long))
                
        for symbol in short:
            self.SetHoldings(symbol, -1 / len(short))
    def Selection(self) -> None:
        if self.count_month == 3: # The portfolio is rebalance quarterly
            self.selection_flag = True
            self.count_month = 1
        else:
            self.count_month += 1
            
class SymbolData():
    def __init__(self, period: int) -> None:
        self._volumes: RollingWindow = RollingWindow[float](period)
        
    def update(self, volume) -> None:
        self._volumes.Add(volume)
        
    def is_ready(self) -> bool:
        return self._volumes.IsReady
        
    def total_volume(self) -> float:
        return sum(list(self._volumes))
        
class CustomFeeModel(FeeModel):
    def GetOrderFee(self, parameters):
        fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
        return OrderFee(CashAmount(fee, "USD"))