“通过动量和卖空兴趣交易CRSP股票，做多低卖空兴趣、高动量股票，做空高卖空兴趣、高动量股票，使用价值加权、每月重新平衡的投资组合。”
资产类别: 股票 | 地区: 美国 | 周期: 每月 | 市场: 股票 | 关键词: 卖空、动量
I. 策略概要
投资范围包括CRSP股票，不包括ADR和ETF。股票首先根据六个月的动量分为十分位，然后在每个动量十分位内根据卖空兴趣水平进一步划分。在最高动量十分位中，做多卖空兴趣最低的股票，做空卖空兴趣最高的股票。该策略采用价值加权，每月重新平衡，并利用动量和卖空兴趣之间的关系来产生回报。
II. 策略合理性
该策略利用了卖空者的专业知识，他们通常是经验丰富的交易员，能够识别高估的股票。高卖空兴趣的股票与未来较低的回报相关联，因此卖空这些股票是该方法的一个关键组成部分。加入动量进一步增强了策略，因为过去动量高的股票通常被高估，这种错误定价被卖空者识别。该策略通过跟随“聪明钱”运作，这是金融市场中一个成熟的原则。通过结合动量和卖空兴趣，该策略有效地利用了老练卖空者的洞察力和行动来产生回报。其功能基于经过验证的市场行为。
III. 来源论文
Short Selling Activity and Future Returns: Evidence from FinTech Data [点击查看论文]
- Gargano, Antonio。C.T.鲍尔商学院
<摘要>
我们使用来自领先金融科技公司（S3 Partners）的新数据集来研究卖空兴趣预测美国股票回报横截面的能力。我们发现卖空兴趣（即卖空股票数量占已发行股票数量的比例）是一个看跌指标，这与理论预测和卖空者是知情交易者的直觉一致。在最高（最低）卖空兴趣十分位中做多（做空）的对冲投资组合，在等权重股票时产生-7.6%的年度四因子Fama-French阿尔法，在根据市值加权股票时产生-6.24%的年度四因子Fama-French阿尔法。以过去回报为条件可以提高卖空兴趣的预测准确性：仅使用过去六个月涨幅最大的股票的对冲卖空兴趣投资组合产生-17.88%的阿尔法。控制其他已知股票回报驱动因素（例如规模、价值和流动性）的多变量回归证实了这些发现的有效性。在Fama-MacBeth和面板回归中，我们发现卖空兴趣增加一个标准差预示着未来调整后回报下降4.3%至9.3%。
IV. 回测表现
| 年化回报 | 15.66% |
| 波动率 | 17.8% |
| β值 | -0.054 |
| 夏普比率 | 0.88 |
| 索提诺比率 | 0.436 |
| 最大回撤 | N/A |
| 胜率 | 50% |
V. 完整的 Python 代码
from AlgorithmImports import *
from io import StringIO
from typing import List, Dict
from pandas.core.frame import dataframe
from numpy import isnan
class ShortSellingActivityAndMomentum(QCAlgorithm):
def Initialize(self) -> None:
self.SetStartDate(2017, 1, 1) # short interest data starts at 12-2017
self.SetCash(100_000)
self.tickers_to_ignore: List[str] = ['NE']
self.data: Dict[Symbol, SymbolData] = {}
self.weight: Dict[Symbol, float] = {} # storing symbols, with their weights for trading
self.quantile: int = 4
self.leverage: int = 5
self.period: int = 6 * 21 # need 6 months of daily prices
market: Symbol = self.AddEquity("SPY", Resolution.Daily).Symbol
# source: https://www.finra.org/finra-data/browse-catalog/equity-short-interest/data
text: str = self.Download('data.quantpedia.com/backtesting_data/economic/short_volume.csv')
self.short_volume_df: dataframe = pd.read_csv(StringIO(text), delimiter=';')
self.short_volume_df['date'] = pd.to_datetime(self.short_volume_df['date']).dt.date
self.short_volume_df.set_index('date', inplace=True)
# self.fundamental_count: int = 1000
# self.fundamental_sorting_key = lambda x: x.MarketCap
self.selection_flag: bool = False
self.UniverseSettings.Resolution = Resolution.Daily
self.AddUniverse(self.FundamentalSelectionFunction)
self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
self.settings.daily_precise_end_time = False
self.Schedule.On(self.DateRules.MonthStart(market), self.TimeRules.AfterMarketOpen(market), self.Selection)
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 the rolling window every day
for stock in fundamental:
symbol = stock.Symbol
# store daily price
if symbol in self.data:
self.data[symbol].update(stock.AdjustedPrice)
# monthly rebalance
if not self.selection_flag:
return Universe.Unchanged
# check last date on custom data
if self.Time.date() > self.short_volume_df.index[-1] or self.Time.date() < self.short_volume_df.index[0]:
self.Liquidate()
return Universe.Unchanged
# select top n stocks by dollar volume
selected: List[Fundamental] = [
x for x in fundamental
if x.HasFundamentalData
and x.MarketCap != 0
and x.Symbol.Value not in self.tickers_to_ignore
]
# if len(selected) > self.fundamental_count:
# selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
momentums: Dict[Symbol, float] = {} # storing stocks momentum
market_cap: Dict[Symbol, float] = {} # storing stocks market capitalization
# warmup price rolling windows
for stock in selected:
symbol: Symbol = stock.Symbol
ticker: str = symbol.Value
if symbol not in self.data:
# create SymbolData object for specific stock symbol
self.data[symbol] = SymbolData(self.period)
# get history daily prices
history: dataframe = self.History(symbol, self.period, Resolution.Daily)
if history.empty:
self.Log(f"Not enough data for {symbol} yet.")
continue
closes: Series = history.loc[symbol].close
# store history daily prices into RollingWindow
for _, close in closes.items():
self.data[symbol].update(close)
if ticker in self.short_volume_df.columns:
if isnan(self.short_volume_df[self.short_volume_df.index <= self.Time.date()][ticker][-1]):
continue
self.data[symbol].update_short_interest(self.short_volume_df[self.short_volume_df.index <= self.Time.date()][ticker][-1] / stock.CompanyProfile.SharesOutstanding)
if not self.data[symbol].is_ready():
continue
# store stock market capitalization
market_cap[symbol] = stock.MarketCap
# calculate stock momentum
momentum = self.data[symbol].performance()
# store stock momentum
momentums[symbol] = momentum
# not enough stocks for quartile selection
if len(momentums) < self.quantile:
return Universe.Unchanged
# perform quartile selection
quantile: int = int(len(momentums) / 4)
sorted_by_momentum: List[Symbol] = [x[0] for x in sorted(momentums.items(), key=lambda item: item[1])]
# get top momentum stocks
top_by_momentum: List[Symbol] = sorted_by_momentum[-quantile:]
# check if there are enough data for next quartile selection on top stocks by momentum
if len(top_by_momentum) < self.quantile:
return Universe.Unchanged
# perform quartile selection on top stocks by momentum
quantile = int(len(top_by_momentum) / self.quantile)
sorted_by_short_interest: List[Symbol] = [x for x in sorted(top_by_momentum, key=lambda symbol: self.data[symbol].short_interest)]
# in the top momentum quartile, short the highest short interest quartile and long the quartile with the lowest short interest
short: List[Symbol] = sorted_by_short_interest[-quantile:]
long: List[Symbol] = sorted_by_short_interest[:quantile]
# calculate total long capitalization and total short capitalization
for i, portfolio in enumerate([long, short]):
mc_sum: float = sum(list(map(lambda symbol: market_cap[symbol], portfolio)))
for symbol in portfolio:
self.weight[symbol] = ((-1)**i) * market_cap[symbol] / mc_sum
return list(self.weight.keys())
def OnData(self, data: Slice) -> None:
# rebalance montly
if not self.selection_flag:
return
self.selection_flag = False
# trade execution
portfolio: List[PortfolioTarget] = [PortfolioTarget(symbol, w) for symbol, w in self.weight.items() if symbol in data and data[symbol]]
self.SetHoldings(portfolio, True)
self.weight.clear()
def Selection(self) -> None:
self.selection_flag = True
class SymbolData():
def __init__(self, period: int) -> None:
self.closes: RollingWindow = RollingWindow[float](period)
self.short_interest: Union[None, float] = None
def update(self, close: float) -> None:
self.closes.Add(close)
def update_short_interest(self, short_interest_value: float) -> None:
self.short_interest = short_interest_value
def is_ready(self) -> bool:
return self.closes.IsReady and self.short_interest
def performance(self) -> float:
closes: List[float] = [x for x in self.closes]
return (closes[0] - closes[-1]) / closes[-1]
# custom fee model
class CustomFeeModel(FeeModel):
def GetOrderFee(self, parameters: OrderFeeParameters) -> OrderFee:
fee: float = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
return OrderFee(CashAmount(fee, "USD"))