The strategy trades CRSP stocks by sorting on long-term institutional trades, going long on decreased trades, short on increased trades, using value-weighted portfolios rebalanced quarterly.
ASSET CLASS: stocks | REGION: United States | FREQUENCY:
Quarterly | MARKET: equities | KEYWORD: Stock
I. STRATEGY IN A NUTSHELL
The strategy trades U.S. stocks based on long-term institutional trades, going long on stocks with declining institutional activity and short on stocks with increasing activity, rebalanced quarterly.
II. ECONOMIC RATIONALE
Prolonged institutional buying or selling creates temporary mispricings; subsequent reversals in trades and fundamentals provide opportunities for value-aligned long-short positions.
III. SOURCE PAPER
Long-Term Institutional Trades and the Cross-Section of Returns [Click to Open PDF]
James Bulsiewicz.Fairleigh Dickinson University.
<Abstract>
I investigate the relation between long-term institutional trades and future returns, and find that the cumulative number of shares purchased in net by financial institutions over the prior ten quarters is negatively related to future returns. A long-short portfolio constructed on this measure earns an annualized average Carhart alpha of 9.9%. Overall, I find that long-term institutional trades contain information about future returns that is not already captured by existing short-term institutional trades measures.
IV. BACKTEST PERFORMANCE
| Annualised Return | 8.24% |
| Volatility | 11.04% |
| Beta | -0.013 |
| Sharpe Ratio | 0.75 |
| Sortino Ratio | -0.025 |
| Maximum Drawdown | N/A |
| Win Rate | 53% |
V. FULL PYTHON CODE
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 dataVI. Backtest Performance
