“该策略通过结合动量和竞争指标交易纽约证券交易所、美国证券交易所和纳斯达克股票，做多高动量、低竞争股票，做空低动量股票，每月重新平衡。”
资产类别: 股票 | 地区: 美国 | 周期: 每月 | 市场: 股票 | 关键词: 竞争、动量
I. 策略概要
该策略专注于纽约证券交易所、美国证券交易所和纳斯达克的普通股（股票代码10或11），不包括金融公司和价格低于1美元的股票。它利用动量竞争指标，通过将每只股票的12个月回报（在t-2结束）标准化为z分数来计算。共同基金根据其持股的价值加权回报放置在相同的动量空间中。如果基金的空间接近度低于选定的阈值，则认为这些基金是竞争对手，竞争指标反映了邻近基金之间的相似性。
在t-1，股票按竞争程度分为三分位数。在最低竞争三分位数内，股票按传统的12个月动量进一步分为五分位数。该策略做多最高动量五分位数，做空最低动量五分位数。投资组合按价值加权，每月重新平衡，利用动量和竞争指标来捕捉低效率。
II. 策略合理性
该论文假设，当买方竞争较低时，动量利润较高。随着更多投资者进入具有盈利机会的市场，买方价格压力增加，从而减少了动量策略的租金。当竞争受到限制且套利性较差时，动量更为有效，尤其是当竞争对手基金具有相关投资信号时。该策略侧重于大型股，共同基金在其中充当边际投资者，从而可以衡量买方竞争。结果表明，当竞争较低时，动量价差显著，产生139个基点，但当竞争较高时，动量价差变得不显著。这强调了买方竞争在确定动量盈利能力方面的经济意义。
III. 来源论文
Buy-Side Competition and Momentum Profits [点击查看论文]
- 杰拉德·霍伯格、尼廷·库马尔和纳格普尔南德·普拉巴拉。南加州大学马歇尔商学院——金融与商业经济学系。印度商学院（ISB），海得拉巴。约翰霍普金斯大学凯瑞商学院。
<摘要>
我们开发了一种衡量动量投资买方竞争的指标，并表明它可以解释动量利润。当竞争较低时，每月动量价差为139个基点，而当竞争较高时，动量价差可以忽略不计。在价值加权投资组合和较大市值股票等更具投资性和交易成本更低的策略中，这些结果更为显著。在负偏度较小且夏普比率和索蒂诺比率较好的情况下，可以获得更好的阿尔法。传统上与动量利润相关的几种股票特征无法解释我们的结果。
IV. 回测表现
| 年化回报 | 14.27% |
| 波动率 | 22.98% |
| β值 | 0.027 |
| 夏普比率 | 0.58 |
| 索提诺比率 | 0.315 |
| 最大回撤 | N/A |
| 胜率 | 52% |
V. 完整的 Python 代码
from AlgorithmImports import *
import data_tools
from typing import List, Dict, Set
from itertools import combinations
from pandas.core.frame import dataframe
from pandas.core.series import Series
import numpy as np
# endregion
class BuySideCompetitionandMomentumProfits(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2010, 1, 1)
self.SetCash(100000)
self.exchange_codes:List[str] = ['NYS', 'NAS', 'ASE']
self.market: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
self.competition_quantile: int = 3
self.momentum_quantile: int = 5
self.leverage: int = 5
self.month_period: int = 21
self.period: int = 12
self.competition_period: int = 9
self.target_granularity: float = 0.08858
self.min_share_price: int = 1
self.fundamental_count: int = 1000
self.fundamental_sorting_key = lambda x: x.DollarVolume
self.data: Dict[Symbol, SymbolData] = {}
self.weight: Dict[Symbol, float] = {}
self.stock_competition: Dict[str, RollingWindow] = {}
self.holdings_by_fund: Dict[str, FundHoldings] = {}
self.ticker_universe: Set = set() # every ticker stored in hedge fund holdings data
self.funds_tickers: Dict[str, dict[datetime.date, list]] = {}
hedge_fund_file_content: str = self.Download('data.quantpedia.com/backtesting_data/equity/hedge_fund_holdings/hedge_funds_holdings.json')
hedge_funds_data: List[Dict[str, Dict[str, str]]] = json.loads(hedge_fund_file_content)
for hedge_fund_data in hedge_funds_data:
hedge_fund_names: List[str] = list(hedge_fund_data.keys())
hedge_fund_names.remove('date')
date: datetime.date = datetime.strptime(hedge_fund_data['date'], '%d.%m.%Y').date()
for hedge_fund_name in hedge_fund_names:
if hedge_fund_name not in self.holdings_by_fund:
self.holdings_by_fund[hedge_fund_name] = data_tools.FundHoldings(hedge_fund_name)
holding_list: List[StockHolding] = []
holdings: List[dict] = hedge_fund_data[hedge_fund_name]['stocks']
for holding in holdings:
ticker: str = holding['ticker']
number_of_shares: int = int(holding['#_of_shares'])
weight: float = float(holding['weight'])
self.ticker_universe.add(ticker)
if ticker not in self.funds_tickers:
# initialize dictionary for stock's ticker
self.funds_tickers[ticker] = {}
if date not in self.funds_tickers[ticker]:
# initialize list, where will be all funds, which hold this stock in this date
self.funds_tickers[ticker][date] = []
# add fund with stock weight in that fund to list == tuple (hedge_fund_name, weight)
self.funds_tickers[ticker][date].append((hedge_fund_name, weight))
holding_list.append(data_tools.StockHolding(ticker, number_of_shares, weight))
self.holdings_by_fund[hedge_fund_name].holdings_by_date[date] = holding_list
self.last_date: datetime.date = max([max(x.holdings_by_date) for x in self.holdings_by_fund.values()])
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(self.market), self.TimeRules.AfterMarketOpen(self.market), self.Selection)
def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
for security in changes.AddedSecurities:
security.SetFeeModel(data_tools.CustomFeeModel())
security.SetLeverage(self.leverage)
for security in changes.RemovedSecurities:
if security.Symbol in self.data:
self.data.pop(security.Symbol)
def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
# monthly selection
if not self.selection_flag:
return Universe.Unchanged
# update the rolling window every month
for stock in fundamental:
symbol: Symbol = stock.Symbol
if symbol in self.data:
self.data[symbol].update_data(stock.AdjustedPrice)
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.Symbol.Value in self.ticker_universe
and x.AssetClassification.MorningstarSectorCode != MorningstarSectorCode.FinancialServices
and x.SecurityReference.ExchangeId in self.exchange_codes
]
if len(selected) > self.fundamental_count:
selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
# warmup price rolling windows
for stock in selected:
symbol: Symbol = stock.Symbol
if symbol in self.data:
continue
self.data[symbol] = data_tools.SymbolData(self.period)
history: dataframe = self.History(symbol, self.period * self.month_period, Resolution.Daily)
if history.empty:
self.Log(f"Not enough data for {symbol} yet.")
continue
closes: Series = history.loc[symbol].close.groupby(pd.Grouper(freq='M')).last()
for time, close in closes.items():
self.data[symbol].update_data(close)
selected_dict: Dict[Symbol, Fundamental] = {x.Symbol.Value: x for x in selected if self.data[x.Symbol].is_ready()}
stock_z_score: Dict[str, float] = {}
fund_z_score: Dict[str, float] = {}
if len(selected_dict) != 0:
cross_mean: float = np.mean([sym_data.get_momentum() for sym, sym_data in self.data.items() if sym.Value in selected_dict and sym_data.is_ready()])
cross_std: float = np.std([sym_data.get_momentum() for sym, sym_data in self.data.items() if sym.Value in selected_dict and sym_data.is_ready()])
# calculate z-score of stocks
stock_z_score: Dict[str, float] = {ticker: (self.data[selected_dict[ticker].Symbol].get_momentum() - cross_mean) / cross_std for ticker in self.ticker_universe \
if ticker in selected_dict and self.data[selected_dict[ticker].Symbol].is_ready()}
# calculate z-score of funds
for fund, fund_data in self.holdings_by_fund.items():
last_date: List[datetime.date] = [x for x in list(fund_data.holdings_by_date.keys()) if x < self.Time.date()]
if len(last_date) == 0:
continue
last_date: datetime.date = max(last_date)
fund_z_score[fund] = sum([(x.weight / 100) * stock_z_score[x.ticker] for x in fund_data.holdings_by_date[last_date] if x.ticker in selected_dict])
if len(fund_z_score) == 0:
return Universe.Unchanged
normalization_constant: float = max(list(fund_z_score.values()))
# stock level competitions
COMP: Dict[Symbol, float] = {}
# calculate fund-level competition and stock-level competition
for ticker, date in self.funds_tickers.items():
if ticker not in selected_dict:
continue
last_dates: List[datetime.date] = [x for x in date.keys() if x < self.Time.date()]
last_date: datetime.date = max(last_dates) if len(last_dates) > 0 else None
if last_date is None:
continue
if len(date[last_date]) > 1:
fund_competition: List[float] = []
for fund, weight in date[last_date]:
fund_competition.append(sum([normalization_constant - abs(fund_z_score[fund] - fund_z_score[x]) for x, w in date[last_date] \
if x != fund and abs(fund_z_score[fund] - fund_z_score[x]) <= self.target_granularity]))
if len(fund_competition) > 2 and sum(fund_competition) != 0:
COMP[selected_dict[ticker].Symbol] = np.mean(fund_competition)
# if ticker not in self.stock_competition:
# self.stock_competition[ticker] = RollingWindow[float](self.competition_period)
# self.stock_competition[ticker].Add(np.mean(fund_competition))
# if self.stock_competition[ticker].IsReady:
# COMP[selected_dict[ticker].Symbol] = np.mean(list(self.stock_competition[ticker])[::-1][:5])
# remove keys with duplicate COMP values for the sake of not random backtest
COMP_no_repl: Dict[float, Symbol] = {}
for key, val in COMP.items():
COMP_no_repl.setdefault(val, key)
COMP_no_repl: Dict[Symbol, float] = dict((v, k) for k, v in COMP_no_repl.items())
# sort and divide to quantiles
if len(COMP_no_repl) >= self.competition_quantile * self.momentum_quantile:
sorted_stock_level_competitions: List[Symbol] = sorted(COMP_no_repl, key=COMP_no_repl.get)
quantile: int = int(len(sorted_stock_level_competitions) / self.competition_quantile)
lowest_competition: List[Symbol] = sorted_stock_level_competitions[:quantile]
# get momemtum for second sorting
lowest_competitions: Dict[Symbol, float] = {sym: sym_data.get_momentum() for sym, sym_data in self.data.items() if sym in lowest_competition and sym.Value in selected_dict}
sorted_lowest_competitions: List[Symbol] = sorted(lowest_competitions, key=lowest_competitions.get, reverse=True)
quantile: int = int(len(sorted_lowest_competitions) / self.momentum_quantile)
long: List[Symbol] = sorted_lowest_competitions[:quantile]
short: List[Symbol] = sorted_lowest_competitions[-quantile:]
# calculate weights based on values
for i, portfolio in enumerate([long, short]):
mc_sum: float = sum(list(map(lambda symbol: selected_dict[symbol.Value].MarketCap, portfolio)))
for symbol in portfolio:
self.weight[symbol] = ((-1)**i) * selected_dict[symbol.Value].MarketCap / mc_sum
return list(self.weight.keys())
def OnData(self, data: Slice) -> None:
# rebalance monthly
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 Selection(self) -> None:
if self.Time.date() > self.last_date:
self.Liquidate()
return
self.selection_flag = True 