“该策略识别成交量飙升，构建一个在负回报期间贝塔值最低的10%ETF成分股的投资组合。这些股票被买入并持有40天，等权重。”
资产类别: 股票 | 地区: 美国 | 周期: 每日 | 市场: 股票 | 关键词: ETF成分股
I. 策略概要
投资范围包括来自9个行业ETF、标普500 ETF和小盘股ETF的股票。该策略识别成交量飙升，定义为成交量高于平均值三个标准差的日子。如果成交量飙升伴随着负回报，投资者会创建一个等权重的投资组合，其中包含相对于ETF贝塔值最低的10%ETF成分股。这些股票被买入并持有40天。投资组合以等权重重新平衡。
II. 策略合理性
ETF中的扭曲和反转之所以发生，是因为并非所有成分股都以相同的方式或程度受到外部冲击的影响。这些“局外人”可能会因冲击的性质而异，冲击的性质可能包括商品价格下跌、负面盈利意外或政治事件等因素。成分股之间的差异越大，出现扭曲的机会就越多。这些不同的敞口导致潜在的市场低效，可以通过有针对性的投资策略加以利用。
III. 来源论文
The Revenge of the Stock Pickers [点击查看论文]
- Hailey Lynch 等人
<摘要>
当交易所交易基金（ETF）围绕某个主题进行大量交易时，其成分股之间的相关性会显著增加。即使是一些对主题本身几乎没有或负面敞口的证券，也开始与其他ETF成分股同步交易。换句话说，由于ETF投资者对证券层面的信息不敏感，他们经常“鱼龙混杂”。随着个股价格随ETF被拉高或拉低，这些错误定价可能会变得显著，而利用它们实现的利润可能会为选股者提供机会。
IV. 回测表现
| 年化回报 | 18% |
| 波动率 | N/A |
| β值 | 0.008 |
| 夏普比率 | N/A |
| 索提诺比率 | -4.171 |
| 最大回撤 | N/A |
| 胜率 | 57% |
V. 完整的 Python 代码
from AlgorithmImports import *
from QC100UniverseSelectionModel import QC100UniverseSelectionModel
from collections import deque
from scipy import stats
class StockPickingETFConstituents(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2000, 1, 1)
self.SetCash(100000)
self.period:int = 21
self.SetWarmup(self.period, Resolution.Daily)
# Source: https://github.com/QuantConnect/Lean/blob/master/Algorithm.Framework/Selection/QC500UniverseSelectionModel.py
self.UniverseSettings.Resolution = Resolution.Daily
self.SetUniverseSelection(QC100UniverseSelectionModel(n_of_symbols = 100, select_every_n_months = 3))
self.quantile:int = 10
# daily price data
self.data:dict[Symbol, deque] = {}
self.market:Symbol = self.AddEquity('OEF', Resolution.Daily).Symbol
self.day_holding_period:int = 40
self.managed_queue:list[RebalanceQueueItem] = []
def OnSecuritiesChanged(self, changes):
# newly added proxy S&P stocks
for security in changes.AddedSecurities:
symbol:Symbol = security.Symbol
security.SetFeeModel(CustomFeeModel())
security.SetLeverage(10)
if symbol not in self.data:
self.data[symbol] = deque(maxlen = self.period)
# delete removed S&P stock from data storage
for security in changes.RemovedSecurities:
symbol:Symbol = security.Symbol
if symbol in self.data:
del self.data[symbol]
def OnData(self, data) -> None:
# store daily data for universe
for symbol in self.data:
if symbol in data and data[symbol]:
price:float = data[symbol].Value
volume:float = data[symbol].Volume
self.data[symbol].append((price, volume))
market_closes:list[float] = []
trade_flag:bool = False
# market etf data is ready
if self.market in self.data and len(self.data[self.market]) == self.data[self.market].maxlen:
market_closes = [x[0] for x in self.data[self.market]]
volumes:list[float] = [x[1] for x in self.data[self.market]]
volume_mean:float = np.mean(volumes)
volume_std:float = np.std(volumes)
recent_volume:float = volumes[-1]
# volume spike has not occured
if recent_volume > volume_mean + 3 * volume_std:
# last day's return was negative
last_day_return:float = market_closes[-1] / market_closes[-2] - 1
if last_day_return < 0:
trade_flag = True
market_closes:np.ndarray = np.array(market_closes)
if trade_flag:
stock_beta:dict[Symbol, float] = {}
for symbol in self.data:
if symbol == self.market: continue
# stock data is ready
if (symbol in self.data and len(self.data[symbol]) == self.data[symbol].maxlen):
# beta calculation
stock_closes:np.ndarray = np.array([x[0] for x in self.data[symbol]])
market_returns:np.ndarray = (market_closes[1:] - market_closes[:-1]) / market_closes[:-1]
stock_returns:np.ndarray = (stock_closes[1:] - stock_closes[:-1]) / stock_closes[:-1]
# manual beta calc
cov = np.cov(market_returns, stock_returns)[0][1]
market_variance = np.std(market_returns) ** 2
beta = cov / market_variance
# beta, alpha, r_value, p_value, std_err = stats.linregress(market_returns, stock_returns)
stock_beta[symbol] = beta
if len(stock_beta) >= self.quantile:
# beta sorting
sorted_by_beta:list = sorted(stock_beta.items(), key = lambda x: x[1], reverse = True)
quantile:int = int(len(sorted_by_beta) / self.quantile)
long:list[Symbol] = [x[0] for x in sorted_by_beta[-quantile:]]
long_w:float = self.Portfolio.TotalPortfolioValue / self.day_holding_period / len(long)
long_symbol_q:list[tuple[Symbol, float]] = [(x, np.floor(long_w / self.data[x][-1][0])) for x in long]
# append long portfolio to managed queue
self.managed_queue.append(RebalanceQueueItem(long_symbol_q))
# rebalance portfolio
remove_item:RebalanceQueueItem = None
for item in self.managed_queue:
if item.holding_period == self.day_holding_period:
for symbol, quantity in item.symbol_q:
self.MarketOrder(symbol, -quantity)
remove_item = item
elif item.holding_period == 0:
open_symbol_q:list[tuple[Symbol, float]] = []
for symbol, quantity in item.symbol_q:
if symbol in data and data[symbol]:
self.MarketOrder(symbol, quantity)
open_symbol_q.append((symbol, quantity))
# only opened orders will be closed
item.symbol_q = open_symbol_q
item.holding_period += 1
# we need to remove closed part of portfolio after loop. Otherwise it will miss one item in self.managed_queue
if remove_item:
self.managed_queue.remove(remove_item)
class RebalanceQueueItem():
def __init__(self, symbol_q:list) -> None:
# symbol/quantity collections
self.symbol_q:list[tuple[Symbol, float]] = symbol_q
self.holding_period:int = 0
# custom fee model
class CustomFeeModel(FeeModel):
def GetOrderFee(self, parameters):
fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
return OrderFee(CashAmount(fee, "USD"))