The investment universe consists of: common stocks traded at NYSE, NASDAQ or AMEX
ASSET CLASS: stocks | REGION: United States | FREQUENCY:
Monthly | MARKET: equities | KEYWORD: Dispersion Factor
I. STRATEGY IN A NUTSHELL
The strategy trades common stocks on NYSE, NASDAQ, and AMEX based on exposure to cross-industry dispersion (CID). Stocks with low CID beta (β_CID) are longed, and those with high β_CID are shorted. Portfolios are value-weighted and rebalanced monthly.
II. ECONOMIC RATIONALE
CID captures sector-specific labor income risk. Stocks sensitive to CID tend to underperform during sectoral shifts, reflecting human capital risk. The CID premium is distinct from traditional volatility, VIX, or known uncertainty measures.
III. SOURCE PAPER
Labor Income Risk and the Cross-Section of Expected Returns [Click to Open PDF]
Mykola Pinchuk, University of Rochester, Simon Business School
<Abstract>
This paper explores asset pricing implications of unemployment risk from sectoral shifts. I proxy for this risk using cross-industry dispersion (CID), defined as a mean absolute deviation of returns of 49 industry portfolios. CID peaks during periods of accelerated sectoral reallocation and heightened uncertainty. I find that expected stock returns are related cross-sectionally to the sensitivities of returns to innovations in CID. Annualized returns of the stocks with high sensitivity to CID are 5.9% lower than the returns of the stocks with low sensitivity. Abnormal returns with respect to the best factor model are 3.5%, suggesting that common factors can not explain this return spread. Stocks with high sensitivity to CID are likely to be the stocks, which benefited from sectoral shifts. CID positively predicts unemployment through its long-term component, consistent with the hypothesis that CID is a proxy for unemployment risk from sectoral shifts.
IV. BACKTEST PERFORMANCE
| Annualised Return | 6.04% |
| Volatility | 13.67% |
| Beta | 0.031 |
| Sharpe Ratio | 0.44 |
| Sortino Ratio | N/A |
| Maximum Drawdown | N/A |
| Win Rate | 49% |
V. FULL PYTHON CODE
from AlgorithmImports import *
from typing import List, Dict
from collections import deque
from pandas.core.frame import dataframe
import statsmodels.api as sm
from dateutil.relativedelta import relativedelta
# endregion
class CrossIndustryDispersionFactor(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2000, 1, 1)
self.SetCash(100000)
self.stock_price_data:Dict[Symbol, deque] = {}
self.ff_price_data:Dict[Symbol, deque] = {}
self.leverage:int = 3
self.month_period:int = 2 * 12
self.period:int = self.month_period * 30
self.quantile:int = 5
self.SetWarmup(self.period, Resolution.Daily)
self.market:Symbol = self.AddEquity("SPY", Resolution.Daily).Symbol
self.ff_industry:Symbol = self.AddData(QuantpediaFamaFrench, "fama_french_49_industry_VW", Resolution.Daily).Symbol
self.weight:Dict[Symbol, float] = {}
self.coarse_count:int = 1000
self.selection_flag:bool = False
self.exchanges:List[str] = ['NYS', 'NAS', 'ASE']
self.UniverseSettings.Resolution = Resolution.Daily
self.AddUniverse(self.CoarseSelectionFunction, self.FineSelectionFunction)
self.Schedule.On(self.DateRules.MonthStart(self.market), self.TimeRules.AfterMarketOpen(self.market), self.Selection)
def CoarseSelectionFunction(self, coarse:List[CoarseFundamental]) -> List[Symbol]:
# update price every day
for stock in coarse:
symbol = stock.Symbol
if symbol in self.stock_price_data:
self.stock_price_data[symbol].append((self.Time, stock.AdjustedPrice))
if not self.selection_flag:
return Universe.Unchanged
if self.coarse_count < 3000:
selected:list = sorted([x for x in coarse if x.HasFundamentalData and x.AdjustedPrice >= 1],
key=lambda x: x.DollarVolume, reverse=True)[:self.coarse_count]
else:
selected:list = [x for x in coarse if x.HasFundamentalData and x.AdjustedPrice >= 1]
# warmup prices
for stock in selected:
symbol:Symbol = stock.Symbol
if symbol in self.stock_price_data:
continue
self.stock_price_data[symbol] = deque(maxlen=self.period)
history = self.History(symbol, self.period, Resolution.Daily)
if history.empty:
self.Log(f"Not enough data for {symbol} yet.")
continue
closes = history.loc[symbol].close
for time, close in closes.iteritems():
self.stock_price_data[symbol].append((time, close))
return [x.Symbol for x in selected if len(self.stock_price_data[x.Symbol]) == self.stock_price_data[x.Symbol].maxlen]
def FineSelectionFunction(self, fine:List[FineFundamental]) -> List[Symbol]:
# filter fine
fine:Dict[Symbol, FineFundamental] = {x.Symbol : x for x in fine if x.MarketCap != 0 and x.SecurityReference.ExchangeId in self.exchanges }
if len(fine) >= self.coarse_count:
fine = {x[0] : x[1] for x in sorted(fine.items(), key = lambda x: x[1].MarketCap, reverse=True)[:self.coarse_count]}
# create FF df out of deques
assets:Dict = {symbol : [i[1] for i in deq] for symbol, deq in self.ff_price_data.items()}
ff_df:dataframe = pd.dataframe(assets, index=[i[0] for i in list(self.ff_price_data.values())[0]])
# daily FF returns to daily equity
ff_df[QuantpediaFamaFrench._columns] = (1 + ff_df[QuantpediaFamaFrench._columns]).cumprod()
# calculate monthly return
ff_df = ff_df.groupby(pd.Grouper(freq='M')).last()
ff_df = ff_df.pct_change().iloc[-self.month_period-1:-1]
CID:dataframe = abs(ff_df[QuantpediaFamaFrench._columns].sub(ff_df[self.market], axis=0)).mean(axis=1)
# create stock df out of deques
assets:Dict = {symbol : [i[1] for i in self.stock_price_data[symbol]] for symbol, _ in fine.items()}
stock_df:dataframe = pd.dataframe(assets, index=[i[0] for i in list(self.stock_price_data.values())[0]])
# calculate monthly stock returns
stock_df = stock_df.groupby(pd.Grouper(freq='M')).last()
stock_df = stock_df.pct_change().iloc[-self.month_period-1:-1]
# run regression
x:np.ndarray = CID.values
y:np.ndarray = stock_df.values
model = self.multiple_linear_regression(x, y)
beta_values:np.ndarray = model.params[1]
asset_cols:List[str] = list(stock_df.columns)
# store beta by symbol
beta_by_symbol:Dict[Symbol, float] = {}
for i, beta in enumerate(beta_values):
beta_by_symbol[fine[asset_cols[i]]] = beta
if len(beta_by_symbol) >= self.quantile:
# sort by beta
sorted_beta:List[Tuple] = sorted(beta_by_symbol, key=beta_by_symbol.get, reverse=True)
quantile:int = len(sorted_beta) // self.quantile
long:List[FineFundamental] = sorted_beta[-quantile:]
short:List[FineFundamental] = sorted_beta[:quantile]
total_market_cap_long:float = sum([x.MarketCap for x in long])
total_market_cap_short:float = sum([x.MarketCap for x in short])
for stock in long:
self.weight[stock.Symbol] = stock.MarketCap / total_market_cap_long
for stock in short:
self.weight[stock.Symbol] = -stock.MarketCap / total_market_cap_short
return list(self.weight.keys())
def OnSecuritiesChanged(self, changes:SecurityChanges) -> None:
for security in changes.AddedSecurities:
security.SetFeeModel(CustomFeeModel())
security.SetLeverage(self.leverage)
def OnData(self, data: Slice) -> None:
# store FF industry data and market data
if data.ContainsKey(self.ff_industry) and data[self.ff_industry]:
if data.ContainsKey(self.market) and data[self.market]:
# init deques
for symbol in [self.market] + QuantpediaFamaFrench._columns:
if symbol not in self.ff_price_data:
self.ff_price_data[symbol] = deque(maxlen=self.period)
# FF daily returns
for col in QuantpediaFamaFrench._columns:
self.ff_price_data[col].append((self.Time, data[self.ff_industry].GetProperty(col) / 100.))
# market daily prices
self.ff_price_data[self.market].append((self.Time, data[self.market].Value))
if not self.selection_flag:
return
self.selection_flag = False
# liquidate
stocks_invested:List[Symbol] = [x.Key for x in self.Portfolio if x.Value.Invested]
for symbol in stocks_invested:
if symbol not in self.weight:
self.Liquidate(symbol)
# trade execution
for symbol, w in self.weight.items():
if symbol in data and data[symbol]:
self.SetHoldings(symbol, w)
self.weight.clear()
def Selection(self) -> None:
if self.ff_price_data:
if all(len(self.ff_price_data[col]) == self.period for col in QuantpediaFamaFrench._columns) and \
len(self.ff_price_data[self.market]) == self.period:
self.selection_flag = True
def multiple_linear_regression(self, x:np.ndarray, y:np.ndarray):
# x = np.array(x).T
x = sm.add_constant(x)
result = sm.OLS(endog=y, exog=x).fit()
return result
# Quantpedia data
# NOTE: IMPORTANT: Data order must be ascending (datewise)
class QuantpediaFamaFrench(PythonData):
def GetSource(self, config, date, isLiveMode):
return SubscriptionDataSource("data.quantpedia.com/backtesting_data/equity/fama_french/fama_french_49_industry_VW.csv", SubscriptionTransportMedium.RemoteFile, FileFormat.Csv)
_last_update_date:datetime.date = datetime(1,1,1).date()
_columns:List[str] = list()
@staticmethod
def get_last_update_date() -> Dict[Symbol, datetime.date]:
return QuantpediaFamaFrench._last_update_date
def Reader(self, config, line, date, isLiveMode):
data = QuantpediaFamaFrench()
data.Symbol = config.Symbol
if not line[0].isdigit():
QuantpediaFamaFrench._columns = line.split(',')[1:] # skip 'Date' columns
return None
split = line.split(',')
data.Time = datetime.strptime(split[0], "%Y-%m-%d") + relativedelta(months=1)
if data.Time.date() > QuantpediaFamaFrench._last_update_date:
QuantpediaFamaFrench._last_update_date = data.Time.date()
for i, col_name in enumerate(QuantpediaFamaFrench._columns):
data[col_name] = float(split[i+1])
return data
# custom fee model
class CustomFeeModel(FeeModel):
def GetOrderFee(self, parameters):
fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
return OrderFee(CashAmount(fee, "USD"))