“该策略涉及根据空头头寸变化和内部人需求对CRSP股票进行排序,形成两个投资组合。第一个被做空,第二个被做多,每月进行再平衡并等权重。”

I. 策略概要

该策略涉及CRSP股票,使用市场调整回报(投资组合回报减去CRSP市场回报)。在每个季度末,股票根据空头头寸变化和内部人需求变化分为两组(增加和减少)。这些组再根据空头头寸和内部人需求变化的中位数细分为低和高子组。

构建两个投资组合:一个包含空头头寸增加最多且内部人需求减少最多的股票,另一个包含空头头寸减少最多且内部人需求增加最多的股票。该策略做空第一个投资组合,做多第二个投资组合,股票等权重并每月重新平衡。

II. 策略合理性

卖空成本高昂,涉及借贷成本和费用,这促使卖空者仅在其拥有 superior 信息时才采取行动。这些卖家擅长分析公共数据,如公司新闻和公共订单流。企业内部人员掌握有价值的私人信息,也对公司价值具有预测能力。

研究发现,结合卖空和内幕交易信息比单独使用任何一种来源都能产生更高的利润。这两个因素都有助于投资组合的表现,利用了这些交易者相对于其他交易者的信息优势。当信息差距缩小,例如在经济低迷时期信息不对称程度较低的环境中,这种优势就会减弱,因为不确定性降低了知情交易者优势的可信度。结果与信息假说一致,强调了这些信息优势在预测回报方面的价值。

III. 来源论文

When Short Sellers and Corporate Insiders Agree on Stock Pricing [点击查看论文]

<摘要>

作者提出了一种利用卖空者和公司内部人交易信息的策略。他们发现,该策略在至少一年的时间里获得了统计上显著且经济上有意义的风险调整回报,这主要源于知情投资者和不知情投资者之间的信息不对称。基于这一发现,他们接着表明,该策略在高信息不对称环境和经济扩张时期效果最佳。这些结果对投资从业者具有重要启示。对高信息不对称公司感兴趣的投资者在做出投资决策时可以参考空头头寸和内部人需求的信息。

IV. 回测表现

年化回报10.6%
波动率19.83%
β值-0.295
夏普比率0.53
索提诺比率N/A
最大回撤N/A
胜率36%

V. 完整的 Python 代码

from AlgorithmImports import *
from collections import deque
import pandas as pd
import numpy as np
from io import StringIO
#endregion
class WhenShortSellersandCorporateInsidersAgreeonStockPricing(QCAlgorithm):
    def Initialize(self):
        self.SetStartDate(2013, 1, 1)
        self.SetCash(100000)
        self.symbols = [
                        'AAPL','MSFT','AMZN','FB','BRKB','GOOGL','GOOG','JPM','JNJ','V',
                        'PG','XOM','UNH','BAC','MA','T','DIS','INTC','HD','VZ',
                        'MRK','PFE','CVX','KO','CMCSA','CSCO','PEP','WFC','C','BA',
                        'ADBE','WMT','CRM','MCD','MDT','BMY','ABT','NVDA','NFLX','AMGN',
                        'PM','PYPL','TMO','COST','ABBV','ACN','HON','NKE','UNP','UTX',
                        # 'NEE','IBM','TXN','AVGO','LLY','ORCL','LIN','SBUX','AMT','LMT',
                        # 'GE','MMM','DHR','QCOM','CVS','MO','LOW','FIS','AXP','BKNG',
                        # 'UPS','GILD','CHTR','CAT','MDLZ','GS','USB', 'CI','ANTM','BDX',
                        # 'TJX','ADP','TFC','CME','SPGI','COP','INTU','ISRG','CB','SO',
                        # 'D','FISV','PNC','DUK','SYK','ZTS','MS','RTN','AGN','BLK'
                        ]
                        
        self.period = 3 * 21
        self.quantile = 4
        self.max_SI_missing_days = 5
    
        # Create custom universe.
        self.UniverseSettings.Resolution = Resolution.Daily
        self.AddUniverseSelection(FineFundamentalUniverseSelectionModel(self.SelectCoarse, self.SelectFine))
        # dataframe with insider trades for every stock.
        self.insiders_trading = {}
        
        # Daily short interest data.
        self.short_interest = {}
        
        # Short interest and investor demand quarterly pairs.
        self.data = {}
        
        self.long = []
        self.short = []
        
        self.symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
        
        for symbol in self.symbols:
            # Import insiders trading data.
            csv_string_file = self.Download(f'data.quantpedia.com/backtesting_data/economic/insiders_trading/{symbol}.csv')
            if csv_string_file == "": continue
            parser = lambda x: pd.datetime.strptime(x, "%Y-%m-%d")
            self.insiders_trading[symbol] = pd.read_csv(StringIO(csv_string_file), sep=';', parse_dates=['Tran.Date'], date_parser=parser)
            # Import short interest daily data.
            self.AddData(NasdaqCustomColumns, 'FINRA/FNSQ_' + symbol, Resolution.Daily)
            self.short_interest[symbol] = deque(maxlen = self.period)
        self.selection_flag = False
        self.rebalance_flag = False
        
        self.Schedule.On(self.DateRules.MonthEnd(self.symbol), self.TimeRules.BeforeMarketClose(self.symbol), self.Selection)
    def OnSecuritiesChanged(self, changes):
        for security in changes.AddedSecurities:
            security.SetFeeModel(CustomFeeModel())
            security.SetLeverage(10)
    def SelectCoarse(self, coarse):
        if not self.selection_flag:
            return Universe.Unchanged
        
        return [Symbol.Create(x, SecurityType.Equity, Market.USA) for x in self.symbols]
    def SelectFine(self, fine):
        fine = [x for x in fine if x.EarningReports.BasicAverageShares.ThreeMonths > 0 and x.Symbol.Value in self.insiders_trading]
        # Short interests and demand diffs.
        data_change = {}    
        for stock in fine:
            symbol = stock.Symbol
            ticker = symbol.Value
            if self.Securities['FINRA/FNSQ_' + ticker].GetLastData() and (self.Time.date() - self.Securities['FINRA/FNSQ_' + ticker].GetLastData().Time.date()).days > self.max_SI_missing_days:
                self.data[symbol].clear()
                continue
            # Last month's short_interest data is ready.
            if len(self.short_interest[ticker]) == self.short_interest[ticker].maxlen:
                # Calculate investor demand.
                buys = [row['Shares'] for index, row in self.insiders_trading[ticker].iterrows() if row['Symbol'] == ticker and row['Tran.Date'] >= (self.Time - timedelta(days = self.period)) and row['Tran.Date'] <= self.Time and row['Action'] == 'B'] 
                sells = [row['Shares'] for index, row in self.insiders_trading[ticker].iterrows() if row['Symbol'] == ticker and row['Tran.Date'] >= (self.Time - timedelta(days = self.period)) and row['Tran.Date'] <= self.Time and row['Action'] == 'S']
                        
                total_buy_shares = sum(buys)
                total_sell_shares = sum(sells)
                
                if len(buys) != 0 or len(sells) != 0:
                    demand = (total_buy_shares - total_sell_shares) / stock.EarningReports.BasicAverageShares.ThreeMonths
                    
                    # Calculate quarterly short interest.
                    short_interest = sum([x[0] for x in self.short_interest[ticker]]) / sum([x[1] for x in self.short_interest[ticker]])
                    
                    # Store quarterly data pairs.                    
                    if symbol not in self.data:
                        self.data[symbol] = deque()
                    self.data[symbol].append([short_interest, demand])
                
                    # If there is at least of 4 quarters of data ready.
                    if len(self.data[symbol]) >= 4:
                        short_interest_diff = np.diff([x[0] for x in self.data[symbol]])
                        demand_diff = np.diff([x[1] for x in self.data[symbol]])
                        
                        data_change[symbol] = [np.median(short_interest_diff), np.median(demand_diff)]
            
        if len(data_change) >= self.quantile:
            # Sorting by short interest and demand diffs.
            sorted_by_interest_change = sorted(data_change.items(), key = lambda x: x[1][0], reverse = True)
            quantile = int(len(sorted_by_interest_change) / self.quantile)
            high_by_interest_change = [x[0] for x in sorted_by_interest_change[:quantile]]
            low_by_interest_change = [x[0] for x in sorted_by_interest_change[-quantile:]]
            
            sorted_by_demand_change = sorted(data_change.items(), key = lambda x: x[1][1], reverse = True)
            quantile  = int(len(sorted_by_demand_change) / self.quantile)
            high_by_demand_change = [x[0] for x in sorted_by_demand_change[:quantile]]
            low_by_demand_change = [x[0] for x in sorted_by_demand_change[-quantile:]]
            
            self.long = [x for x in high_by_interest_change if x in low_by_demand_change]
            self.short = [x for x in low_by_interest_change if x in high_by_demand_change]
        
        return self.long + self.short
    def OnData(self, data):
        # Store short interest data.
        for symbol in self.symbols:
            look_up_symbol = 'FINRA/FNSQ_' + symbol
            if look_up_symbol in data and data[look_up_symbol]:
                short_vol = data[look_up_symbol].GetProperty("SHORTVOLUME")
                total_vol = data[look_up_symbol].GetProperty("TOTALVOLUME")
                
                if symbol in self.short_interest:
                    self.short_interest[symbol].append((short_vol, total_vol))
        
        # rebalance once a month     
        if not self.rebalance_flag:
            return
        self.selection_flag = False
        self.rebalance_flag = False
        
        # Trade execution
        stocks_invested = [x.Key for x in self.Portfolio if x.Value.Invested]
        for symbol in stocks_invested:
            if symbol not in self.long + self.short:
                self.Liquidate(symbol)
        long_count = len(self.long)
        short_count = len(self.short)
        
        for symbol in self.long:
            if symbol in data and data[symbol]:
                self.SetHoldings(symbol, 1 / long_count)
        for symbol in self.short:
            if symbol in data and data[symbol]:
                self.SetHoldings(symbol, -1 / short_count)
               
    def Selection(self):
        if self.Time.month in [3,6,9,12]:
            # clear long and short selection once every selection period, so that portoflio can be rabalanced monthly even without new selection
            self.long.clear()
            self.short.clear()
            self.selection_flag = True
        
        self.rebalance_flag = True
            
# Custom fee model
class CustomFeeModel(FeeModel):
    def GetOrderFee(self, parameters):
        fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
        return OrderFee(CashAmount(fee, "USD"))
# Quandl short interest data.
class NasdaqCustomColumns(NasdaqDataLink):
    def __init__(self) -> None:
        self.ValueColumnName = 'shortvolume'    # also 'TOTALVOLUME' is accesible

发表评论

了解 Quant Buffet 的更多信息

立即订阅以继续阅读并访问完整档案。

继续阅读