“该策略通过利用共享分析师覆盖率交易纽约证券交易所、美国证券交易所和纽约证券交易所MKT股票，做多高关联股票回报，做空低关联股票回报，并每月进行价值加权重新平衡。”
资产类别: 股票 | 地区: 美国 | 周期: 每月 | 市场: 股票 | 关键词: 动量
I. 策略概要
该策略专注于分析师覆盖的纽约证券交易所、美国证券交易所和纽约证券交易所MKT普通股，不包括价格低于5美元的股票。如果股票共享分析师覆盖率，则使用IBES数据和过去12个月发布的盈利预测来确定股票之间的关联。在每个月底，股票根据关联股票投资组合回报（CS RET）排名为五分位数，CS RET计算为所有关联股票的加权平均回报。该策略做多CS RET最高的五分位数，做空最低的五分位数。投资组合按价值加权，并每月重新平衡，利用共享分析师覆盖率进行回报预测。
II. 策略合理性
关联股票之间的关系是由投资者处理信息的能力有限所驱动的，尤其是在许多相关公司之间。研究表明，与传统行业同行相比，分析师共同覆盖的同行能更好地解释回报和基本面的横截面变化。分析师同行能够研究公司特定的关联，这与之前汇总股票或关注单一维度的研究不同。分析师覆盖率适用于大多数公开交易的公司，并捕获跨多个维度的关联。这种方法增强了跨资产回报的可预测性，在分析师共同覆盖的股票中，这种预测性更强，为理解和预测公司特定的回报动态提供了一个稳健的框架。
III. 来源论文
Shared Analyst Coverage and Cross-Asset Momentum Effects [点击查看论文]
- 阿里，乌斯曼，太平洋投资管理公司（PIMCO）
<摘要>
通过共享分析师覆盖率识别公司关联，我们发现关联公司（CF）动量因子产生了每月1.68%的阿尔法（t = 9.67）。在跨度回归中，在控制CF动量后，行业、地域、客户、客户/供应商行业、单部门到多部门和技术动量因子的阿尔法不显著/为负。横截面回归和发达国际市场也存在类似的结果。卖方分析师对关联公司的消息反应迟缓。这些效应在复杂和间接的关联中更为强烈。与投资者注意力有限一致，这些结果表明，动量溢出效应是由共享分析师覆盖率捕获的统一现象。
IV. 回测表现
| 年化回报 | 11.22% |
| 波动率 | 19.28% |
| β值 | 0.133 |
| 夏普比率 | 0.56 |
| 索提诺比率 | -0.066 |
| 最大回撤 | N/A |
| 胜率 | 47% |
V. 完整的 Python 代码
from AlgorithmImports import *
from dateutil.relativedelta import relativedelta
from pandas.core.series import Series
from pandas.core.frame import dataframe
class ConnectedStocksMomentumPortfolio(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2010, 1, 1)
self.SetCash(100_000)
self.weight: Dict[Symbol, float] = {}
self.quantile: int = 5
self.price_data: Dict[Symbol, RollingWindow] = {}
self.m_period: int = 12
self.d_period: int = 21
self.universe_selection_period: int = 1
self.recent_estimate_date_by_analyst: Dict[str, Dict[str, datetime.date]] = {}
market: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
self.required_exchanges: List[str] = ['NYS', 'ASE', 'NAS']
self.already_subscribed: List[Symbol] = []
self.leverage: int = 10
self.min_share_price: float = 5.
self.fundamental_count: int = 500
self.fundamental_sorting_key = lambda x: x.DollarVolume
self.selection_flag: bool = False
self.rebalance_flag: bool = False
self.UniverseSettings.Resolution = Resolution.Daily
self.AddUniverse(self.FundamentalSelectionFunction)
self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
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: Symbol = stock.Symbol
# store daily price
if symbol in self.price_data:
self.price_data[symbol.Value].Add(stock.AdjustedPrice)
if not self.selection_flag:
return Universe.Unchanged
# select new universe once a period
if self.Time.month % self.universe_selection_period != 0:
self.rebalance_flag = True
return Universe.Unchanged
selected: List[Fundamental] = [x for x in fundamental if x.HasFundamentalData
and x.Market == 'usa'
and x.MarketCap != 0
and x.Price >= self.min_share_price
and x.SecurityReference.ExchangeId in self.required_exchanges
]
if len(selected) > self.fundamental_count:
selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
min_date: datetime.date = self.Time.date() - relativedelta(months=self.m_period)
cf_ret: Dict[Fundamental, float] = {}
for stock in selected:
symbol: Symbol = stock.Symbol
i_ticker: str = stock.Symbol.Value
# subscribe Estimize data
if symbol not in self.already_subscribed:
self.AddData(EstimizeEstimate, symbol)
self.already_subscribed.append(symbol)
# warmup price rolling windows
if symbol.Value not in self.price_data:
self.price_data[symbol.Value] = RollingWindow[float](self.d_period)
history: dataframe = self.History(symbol, self.d_period, Resolution.Daily)
if history.empty:
self.Log(f"Not enough data for {symbol} yet.")
continue
closes: Series = history.loc[symbol].close
for time, close in closes.items():
self.price_data[symbol.Value].Add(close)
if self.price_data[symbol.Value].IsReady:
# nij indexed by j
n_ij: Dict[str, int] = {}
for analyst, ticker_estimate_dates in self.recent_estimate_date_by_analyst.items():
# i was covered by analyst
if i_ticker in ticker_estimate_dates:
# check period of the last coverage
if ticker_estimate_dates[i_ticker] >= min_date:
for j_ticker, date_list in ticker_estimate_dates.items():
if j_ticker != i_ticker:
# price data for j is ready
if j_ticker in self.price_data and self.price_data[j_ticker].IsReady:
# check period of the last coverage
if ticker_estimate_dates[j_ticker] >= min_date:
# found connected stocks covered by an analyst
ticker_pair:tuple[str, str] = (i_ticker, j_ticker)
# increment of analysts who cover both tickers i and j
if j_ticker not in n_ij:
n_ij[j_ticker] = 0
n_ij[j_ticker] += 1
# calculate CF RET
N:int = len(n_ij)
if N != 0:
cf_ret[stock] = (1 / sum(list(n_ij.values())) * sum([nij * (self.price_data[j_t][0] / self.price_data[j_t][self.d_period - 1] - 1) for j_t, nij in n_ij.items()]))
self.rebalance_flag = True
if len(cf_ret) >= self.quantile:
# CF RET sorting
sorted_by_cf_ret: List = sorted(cf_ret.items(), key = lambda x:x[1], reverse=True)
quantile: int = int(len(sorted_by_cf_ret) / self.quantile)
long: List[Fundamental] = [x[0] for x in sorted_by_cf_ret[:quantile]]
short: List[Fundamental] = [x[0] for x in sorted_by_cf_ret[-quantile:]]
# market cap weighting
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, slice: Slice) -> None:
# store latest EPS Estimize estimate
estimize = slice.Get(EstimizeEstimate)
for symbol, value in estimize.items():
ticker: str = symbol.Value
if value.AnalystId not in self.recent_estimate_date_by_analyst:
self.recent_estimate_date_by_analyst[value.AnalystId] = {}
if ticker not in self.recent_estimate_date_by_analyst[value.AnalystId]:
self.recent_estimate_date_by_analyst[value.AnalystId][ticker] = datetime.min
self.recent_estimate_date_by_analyst[value.AnalystId][ticker] = value.CreatedAt.date()
if not self.rebalance_flag:
return
if not self.selection_flag:
return
self.selection_flag = False
self.rebalance_flag = False
# trade execution
portfolio: List[PortfolioTarget] = [PortfolioTarget(symbol, w) for symbol, w in self.weight.items() if symbol in slice and slice[symbol]]
self.SetHoldings(portfolio, True)
self.weight.clear()
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"))