“该策略利用行业回报差异（基于SIC分类与HP分类的差异）构建投资组合，每周进行再平衡并按等权重配置。对回报差异最负的行业进行多头交易，对回报差异最正的行业进行空头交易。”
资产类别：股票 | 地区：美国 | 频率：每周 | 市场：股票市场 | 关键词：分类效应
I. 策略概述
该策略针对NYSE、AMEX和NASDAQ的股票，结合两种行业回报数据：
- 官方行业回报（基于SIC分类）。
- 基本面行业回报（基于Hoberg和Phillips (HP) 方法）。
HP分类通过分析公司10-K表格中的产品描述，计算产品相似度来识别行业内的实际竞争者。
每周，股票根据前一周的官方行业回报与基本面行业回报的差异（回报差异）排序为五分位组合：
- Q1组合：回报差异最负的股票（官方回报明显低于基本面回报）。
- Q5组合：回报差异最正的股票（官方回报明显高于基本面回报）。
策略构建一个多空组合，对Q1股票做多，对Q5股票做空，投资组合按等权重配置，每周重新平衡。
II. 策略合理性
学术研究表明，在高频率（如每周）下，股票与其官方行业的回报表现出强协动性，但与基本面行业的协动性较弱。然而，在较低频率下，这种关系会逆转。这种现象支持有限理性假说，即某些投资者倾向于过度依赖官方行业的波动，在短期内对行业冲击反应过度，而这些错误会随着时间的推移逐步被修正。
通过利用这种短期分类偏差，策略抓住了由市场错定价引发的可预测性回报。
III. 论文来源
Categorization Bias in the Stock Market [点击浏览原文]
- 作者：Philipp Kr¨uger、Augustin Landier、David Thesmar
- 机构：日内瓦大学金融研究所、图卢兹经济学院、巴黎HEC商学院及CEPR
<摘要>
本文提供了金融市场中存在分类偏差的证据。一些投资者通过行业的视角看待个别公司，这种分类化思维导致股票回报的错误定价和可预测性。我们通过构建基于Hoberg和Phillips (2010a,b) 方法的相关公司篮子，测量公司官方SIC行业回报与其基本面行业回报之间的差异。
研究发现，股票在短期内与其官方行业表现出强协动性，但随后逐步向其基本面行业（HP分类）的表现回归。利用行业分类引发的错误定价进行多空策略，可以产生统计显著且经济意义显著的风险调整后超额收益。
此外，研究还表明金融分析师也受到行业分类偏差的影响：当一家公司所属的官方行业与其基本面不符时，分析师往往会对官方行业的信息赋予过高权重，从而导致可预测的预测错误。这进一步支持了分类偏差对市场行为和分析师决策的深远影响。
IV. 回测表现
| 年化收益率 | 21.36% |
| 波动率 | 12.35% |
| Beta | 0.103 |
| 夏普比率 | 1.05 |
| 索提诺比率 | 0.023 |
| 最大回撤 | N/A |
| 胜率 | 49% |
V. 完整python代码
from AlgorithmImports import *
from typing import List, Dict
import pandas as pd
import numpy as np
# endregion
class CategorizationEffectInStocks(QCAlgorithm):
def Initialize(self) -> None:
self.SetStartDate(2002, 1, 1)
self.SetCash(100_000)
self.UniverseSettings.Leverage = 5
self.UniverseSettings.Resolution = Resolution.Daily
self.AddUniverse(self.FundamentalSelectionFunction)
self.Settings.MinimumOrderMarginPortfolioPercentage = 0.0
self.settings.daily_precise_end_time = False
self.exchange_codes: List[str] = ['NYS', 'NAS', 'ASE']
self.week_period: int = 5
self.quantile: int = 5
self.selection_flag: bool = False
self.prices: Dict[str, RollingWindow] = {}
self.firm_similarity: Dict[int, Dict[str, Dict[str, float]]] = {}
self.yearly_universes: Dict[int, List[str]] = {}
self.long_symbols: List[Symbol] = []
self.short_symbols: List[Symbol] = []
csv: str = self.Download('data.quantpedia.com/backtesting_data/equity/industries/firm_similarity.csv')
lines: List[str] = csv.split('\r\n')
for line in lines[1:]: # Skip header
if line == '':
continue
line_split: List[str] = line.split(';')
year: int = int(line_split[0].split('-')[0])
if year not in self.firm_similarity:
self.firm_similarity[year]: Dict = {}
if year not in self.yearly_universes:
self.yearly_universes[year]: List = []
industry_comp_ticker: str = line_split[1]
if industry_comp_ticker not in self.firm_similarity[year]:
self.firm_similarity[year][industry_comp_ticker]: Dict = {}
if industry_comp_ticker not in self.yearly_universes[year]:
self.yearly_universes[year].append(industry_comp_ticker)
company_ticker: str = line_split[2]
score: float = float(line_split[3])
self.firm_similarity[year][industry_comp_ticker][company_ticker]: float = score
self.yearly_universes[year].append(company_ticker)
market: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
self.Schedule.On(
self.DateRules.EveryDay(market),
self.TimeRules.BeforeMarketClose(market),
self.Selection)
def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
for security in changes.AddedSecurities:
security.SetFeeModel(CustomFeeModel())
def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
# Update daily prices
for f in fundamental:
ticker: str = f.Symbol.Value
if ticker in self.prices:
self.prices[ticker].Add(f.Price)
if not self.selection_flag:
return Universe.Unchanged
prev_year: int = self.Time.year - 1
if prev_year not in self.firm_similarity:
return Universe.Unchanged
# Select universe
curr_year_universe: List[str] = self.yearly_universes[prev_year]
selected: List[Fundamental] = [f for f in fundamental if f.HasFundamentalData
and f.Symbol.Value in curr_year_universe
and f.SecurityReference.ExchangeId in self.exchange_codes
and not np.isnan(f.AssetClassification.MorningstarIndustryGroupCode)]
warmed_up_stocks: List[Fundamental] = []
# Warm up stock prices
for f in selected:
symbol: Symbol = f.Symbol
ticker: str = symbol.Value
if ticker not in self.prices:
self.prices[ticker] = RollingWindow[float](self.week_period)
history: pd.dataframe = self.History(symbol, self.week_period, Resolution.Daily)
if history.empty:
continue
closes: pd.Series = history.loc[symbol].close
for _, close in closes.items():
self.prices[ticker].Add(close)
if self.prices[ticker].IsReady:
warmed_up_stocks.append(f)
csv_industries: Dict[str, Dict[str, float]] = self.firm_similarity[prev_year]
industry_code_ticker: Dict[str, str] = {}
industries_groups: Dict[str, List[Symbol]] = {}
industry_diff_by_symbol: Dict[Symbol, float] = {}
symbol_by_ticker: Dict[str, Symbol] = {}
# Create industries groups
for f in warmed_up_stocks:
symbol: Symbol = f.Symbol
ticker: str = symbol.Value
industry_group_code: str = f.AssetClassification.MorningstarIndustryGroupCode
if ticker in csv_industries:
# Create match of csv industry group with QC industry group
industry_code_ticker[industry_group_code]: str = ticker
# Create match between stock's ticker and it's symbol, because this stock can be traded
symbol_by_ticker[ticker] = symbol
if industry_group_code not in industries_groups:
industries_groups[industry_group_code]: List[Symbol] = []
industries_groups[industry_group_code].append(f.Symbol)
# Calculate difference between industry performances
for industry_group_code, ticker in industry_code_ticker.items():
industry_universe: List[Symbol] = industries_groups[industry_group_code]
# Official industry performance calculation
official_industry_perf: float = np.mean([self.Performance(self.prices[symbol.Value])
for symbol in industry_universe])
# Fundamental industry performance calculation
tickers_similarities: Dict[str, float] = self.firm_similarity[prev_year][ticker]
fundamental_industry_perf: float = sum([self.Performance(self.prices[stock_ticker]) * similarity_score
for stock_ticker, similarity_score in tickers_similarities.items()
if stock_ticker in self.prices and self.prices[stock_ticker].IsReady])
total_industry_score: float = sum(list(tickers_similarities.values()))
if fundamental_industry_perf != 0 and total_industry_score != 0:
fundamental_industry_perf /= total_industry_score
trade_symbol: Symbol = symbol_by_ticker[ticker]
curr_industries_diff: float = official_industry_perf - fundamental_industry_perf
industry_diff_by_symbol[trade_symbol] = fundamental_industry_perf
# Make sure there are enough stocks for selection
if len(industry_diff_by_symbol) < self.quantile:
return Universe.Unchanged
quantile: int = int(len(industry_diff_by_symbol) / self.quantile)
sorted_by_diff: List[Symbol] = [x[0] for x in sorted(industry_diff_by_symbol.items(), key=lambda item: item[1])]
self.long_symbols = sorted_by_diff[:quantile]
self.short_symbols = sorted_by_diff[-quantile:]
return self.long_symbols + self.short_symbols
def OnData(self, slice: Slice) -> None:
# Rebalance weekly
if not self.selection_flag:
return
self.selection_flag = False
# Trade execution
targets: List[PortfolioTarget] = []
for i, portfolio in enumerate([self.long_symbols, self.short_symbols]):
for symbol in portfolio:
if slice.ContainsKey(symbol) and slice[symbol] is not None:
targets.append(PortfolioTarget(symbol, ((-1) ** i) / len(portfolio)))
self.SetHoldings(targets, True)
self.long_symbols.clear()
self.short_symbols.clear()
def Performance(self, prices_roll_window: RollingWindow) -> float:
return (prices_roll_window[0] / prices_roll_window[prices_roll_window.Count - 1]) - 1
def Selection(self) -> None:
if self.Time.weekday() == 0:
self.selection_flag = True
class CustomFeeModel(FeeModel):
def GetOrderFee(self, parameters: OrderFeeParameters) -> OrderFee:
fee: float = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
return OrderFee(CashAmount(fee, "USD"))
