“该策略通过长期机构交易对CRSP股票进行排序，做多减少交易的股票，做空增加交易的股票，使用价值加权投资组合，每季度重新平衡。”
资产类别: 股票 | 地区: 美国 | 周期: 每季度 | 市场: 股票 | 关键词: 机构交易
I. 策略概要
该策略针对CRSP数据库中价值在5美元至1,000美元之间的股票。每个季度初，股票根据长期机构交易（LIT）进行排序，LIT定义为金融机构在过去10个季度累计净交易量，并按已发行股票数量进行缩放。机构持股数据来源于汤森路透13F数据库。该策略涉及做多机构交易减少幅度最大的十分位股票，做空机构交易增加幅度最大的十分位股票。投资组合采用价值加权，每季度重新平衡，以利用机构交易动态。
II. 策略合理性
该研究支持以下假设：机构长期买卖行为会导致与基本面出现显著偏离。机构倾向于在多个时期内朝同一方向交易，这是由购买基本面改善的股票和出售基本面下降的股票所产生的动量效应驱动的。然而，公司基本面的转变最终会导致机构交易和股票回报的显著逆转，使价格回归其基本面价值。
III. 来源论文
Long-Term Institutional Trades and the Cross-Section of Returns [点击查看论文]
- James Bulsiewicz。费尔利·迪金森大学
<摘要>
我研究了长期机构交易与未来回报之间的关系，发现金融机构在过去十个季度累计净买入的股票数量与未来回报呈负相关。根据这一指标构建的多空投资组合获得了9.9%的年化平均Carhart阿尔法。总的来说，我发现长期机构交易包含的未来回报信息是现有短期机构交易指标尚未捕捉到的。
IV. 回测表现
| 年化回报 | 8.24% |
| 波动率 | 11.04% |
| β值 | 0.013 |
| 夏普比率 | 0.75 |
| 索提诺比率 | -0.025 |
| 最大回撤 | N/A |
| 胜率 | 53% |
V. 完整的 Python 代码
from AlgorithmImports import *
from dateutil.relativedelta import relativedelta
#endregion
class LongTermInstitutionalTradesAndTheCrossSectionOfReturns(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2005, 1, 1) # institutional ownership data starts at 2005
self.SetCash(100_000)
self.period: int = 10 # need n data points of institutional ownership data
self.quantile: int = 10
self.leverage: int = 5
self.weight: Dict[Symbol, float] = {}
self.institutional_data: Dict[str, RollingWindow] = {}
market: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
# subscribe to Quantpedia institutional ownership data
self.institutional_symbol: Symbol = self.AddData(QuantpediaInstitutionalOwnership, 'INSTITUTIONAL_OWNERSHIP_IN_PERCENTS', Resolution.Daily).Symbol
self.selection_flag: bool = False
self.settings.daily_precise_end_time = False
self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
self.UniverseSettings.Resolution = Resolution.Daily
self.AddUniverse(self.FundamentalSelectionFunction)
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]:
# monthly rebalance
if not self.selection_flag:
return Universe.Unchanged
if self.Time.date() > QuantpediaInstitutionalOwnership.get_last_update_date():
return Universe.Unchanged
selected: List[Fundamental] = [
x for x in fundamental
if x.Symbol.Value in self.institutional_data
and x.MarketCap != 0
]
LIT: Dict[Fundamental, float] = {
stock: self.CalculateLIT(self.institutional_data[stock.Symbol.Value])
for stock in selected if self.institutional_data[stock.Symbol.Value].IsReady}
# not enough stocks for decile selection
if len(LIT) < self.quantile:
return Universe.Unchanged
# perform decile selection
quantile: int = int(len(LIT) / self.quantile)
sorted_by_LIT: List[Fundamental] = sorted(LIT, key=LIT.get)
# long stocks with the highest decrease
long: List[Fundamental] = sorted_by_LIT[:quantile]
# short stocks with the lowest decrease
short: List[Fundamental] = sorted_by_LIT[-quantile:]
# calculate weights
for i, portfolio in enumerate([long, short]):
mc_sum: float = sum(map(lambda x: x.MarketCap, portfolio))
for stock in portfolio:
self.weight[stock.Symbol] = ((-1) ** i) * stock.MarketCap / mc_sum
return list(self.weight.keys())
def OnData(self, data: Slice) -> None:
if self.institutional_symbol in data and data[self.institutional_symbol]:
# get objects with tickers and their values from data object
tickers_values_objects: List = [x for x in data[self.institutional_symbol].GetProperty('Tickers_Values')]
# update LIT rolling window for each stock ticker
self.UpdateInstitutionalDataRollingWindow(tickers_values_objects)
# monthly rebalance
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 UpdateInstitutionalDataRollingWindow(self, tickers_values_objects: List[float]) -> None:
''' update rolling window for each stock in self.institutional_data with new institutional data point '''
''' if rolling window doesn't exists it will be created '''
for ticker_value in tickers_values_objects:
# create list from ticker and value object
ticker_value = [x for x in ticker_value]
# get ticker from list
ticker: str = ticker_value[0]
# get value from list
value: float = ticker_value[1]
# initialize rolling window for institutional data for specific stock ticker
if ticker not in self.institutional_data:
self.institutional_data[ticker] = RollingWindow[float](self.period)
# make sure value isn't missing
if value:
self.institutional_data[ticker].Add(value)
def CalculateLIT(self, institutional_data_rolling_window: RollingWindow) -> float:
''' calculates and returns stock's LIT(total change of institutional ownership data) '''
institutional_data: List[float] = list(institutional_data_rolling_window)
prev_data_point: float = institutional_data[0]
total_change: float = 0
for data_point in institutional_data[1:]: # offset first data point
# increase of institutional data
if prev_data_point < data_point:
total_change += (data_point - prev_data_point)
# decrease of institutional data
else:
total_change -= (prev_data_point - data_point)
prev_data_point = data_point
return total_change
def Selection(self) -> None:
self.selection_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"))
# Quantpedia data.
# NOTE: IMPORTANT: Data order must be ascending (datewise)
# Name of the csv file has to be in uppercase
class QuantpediaInstitutionalOwnership(PythonData):
_last_update_date: datetime.date = datetime(1,1,1).date()
@staticmethod
def get_last_update_date() -> datetime.date:
return QuantpediaInstitutionalOwnership._last_update_date
def __init__(self):
self.header = None
def GetSource(self, config, date, isLiveMode):
return SubscriptionDataSource("data.quantpedia.com/backtesting_data/economic/institutional_ownership/{0}.csv".format(config.Symbol.Value), SubscriptionTransportMedium.RemoteFile, FileFormat.Csv)
def Reader(self, config, line, date, isLiveMode):
data = QuantpediaInstitutionalOwnership()
data.Symbol = config.Symbol
# initialize header with stock tickers from csv files
if not self.header:
# split header of csv file
line_split = line.split(',')
# store stock tickers in header list
self.header = [ticker for ticker in line_split[1:]] # exclude 'date'
if not line[0].isdigit():
return None
line_split = line.split(',')
# make two months lag of data
data.Time = datetime.strptime(line_split[0], '%Y-%m-%d') + relativedelta(months=2)
# store last update date
if data.Time.date() > QuantpediaInstitutionalOwnership._last_update_date:
QuantpediaInstitutionalOwnership._last_update_date = data.Time.date()
tickers_values = []
# store values with their tickers
for (ticker), (value) in zip(self.header, line_split[1:]):
if value == '':
tickers_values.append([ticker, None])
else:
tickers_values.append([ticker, float(value)])
data['tickers_values'] = tickers_values
return data