“该策略利用杠杆ETF资金流创建情绪指数（SSI），根据对SSI的敏感性交易纽约证券交易所股票，头寸规模按SSI幅度调整，并每月重新平衡。”
资产类别: 股票 | 地区: 美国 | 周期: 每月 | 市场: 股票 | 关键词: 投机情绪
I. 策略概要
该策略利用六种杠杆ETF（三只做多ETF：QLD、SSO、DDM；三只做空ETF：QID、SDS、DXD）构建投机情绪指数（SSI），并以此交易纽约证券交易所股票。SSI的计算方法是：评估ETF的股票百分比变化，计算做多和做空ETF之间的净资金流，并将此净资金流建模为AR(1)过程。AR(1)过程的残差构成SSI。
对于每只纽约证券交易所股票，其对滞后SSI的敏感性通过滚动36个月回归进行估计。然后，每月根据SSI敏感性将股票分为五分位数。该策略仅使用第一（低敏感性）和第五（高敏感性）五分位数构建多空投资组合。当SSI为正时，做多第五五分位数，做空第一五分位数；当SSI为负时，头寸反转。投资组合的敞口随SSI绝对值而调整——SSI越高，杠杆越高；SSI越低，敞口越小，现金持有量越大。投资组合按价值加权，每月重新平衡。这种方法利用杠杆ETF资金流中反映的情绪驱动市场动态。
II. 策略合理性
杠杆ETF主要被投机性交易者使用，因为它们适合短期目标、具有高昂的直接和间接成本、机构持股比例较低以及与非杠杆ETF相比交易量大。杠杆ETF是为投机目的而设计的利基产品。投机情绪指数（SSI）与现有情绪指标不同，除了市场隐含波动率之外，它们之间的相关性很低，但市场隐含波动率具有统计显著的相关性。与大多数只衡量幅度而不衡量方向的情绪指标不同，SSI同时捕捉了这两个方面。正的SSI值表明投机性交易者正在大量买入做多股票敞口，而负值则表明强劲的做空股票敞口。SSI反映了暂时影响资产价格的非基本面需求冲击，导致回报反转。这些反转构成了盈利策略的基础——逆投机情绪而行。
III. 来源论文
Speculation Sentiment [点击查看论文]
- Shaun Davies. 科罗拉多大学博尔德分校 – 利兹商学院
<摘要>
我利用杠杆交易型开放式指数基金（ETF）的一级市场来衡量总体的、不知情的、赌博式的需求，即投机情绪。杠杆ETF一级市场是一个新颖的环境，它提供了可观察到的套利活动，这些活动归因于纠正ETF份额与其标的资产之间的错误定价。套利活动代表了投机需求冲击的幅度和方向，我用它来构建投机情绪指数。该指标与同期市场回报呈负相关（例如，在下跌市场中看涨），并负向预测回报。结果与投机情绪导致市场范围内的价格扭曲随后反转一致。
IV. 回测表现
| 年化回报 | 12.28% |
| 波动率 | 19.49% |
| β值 | -0.001 |
| 夏普比率 | 0.63 |
| 索提诺比率 | -0.011 |
| 最大回撤 | N/A |
| 胜率 | 50% |
V. 完整的 Python 代码
from AlgorithmImports import *
import statsmodels.api as sm
from scipy import stats
# endregion
class TradingBasedonLeveredETFsSpeculationSentiment(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2000, 1, 1)
self.SetCash(100_000)
self.long_etfs: List[str] = ['QLD', 'SSO', 'DDM']
self.short_etfs: List[str] = ['QID', 'SDS', 'DXD']
# previous month's SO
self.previous_so: Dict[str, float] = {
ticker : 0 for ticker in self.long_etfs + self.short_etfs
}
# monthly prices
self.m_price_data: Dict[Symbol, RollingWindow] = {}
# market data
self.m_regression_period: int = 36 + 1
self.market: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
self.m_price_data[self.market]: RollingWindow = RollingWindow[float](self.m_regression_period)
self.net_percent_share_change_series: RollingWindow = RollingWindow[float](self.m_regression_period)
self.shares_out_data:Symbol = self.AddData(ETFSharesOutstanding, 'ETFSharesOutstanding', Resolution.Daily).Symbol
self.weight: Dict[Symbol, float] = {} # traded weight for a given month
# universe selection
self.quantile: int = 5
self.leverage: int = 5
self.leverage_cap: int = 5
self.exchange_codes: List[str] = ['NYS']
self.fundamental_count: int = 500
self.fundamental_sorting_key = lambda x: x.DollarVolume
self.recent_month: int = -1
self.rebalance_flag: bool = False
self.selection_flag: bool = False
self.UniverseSettings.Resolution = Resolution.Daily
self.AddUniverse(self.FundamentalSelectionFunction)
self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
self.settings.daily_precise_end_time = False
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]:
if not self.selection_flag:
return Universe.Unchanged
# update rolling windows every month
for stock in fundamental:
symbol: Symbol = stock.Symbol
if symbol in self.m_price_data:
self.m_price_data[symbol].Add(stock.AdjustedPrice)
self.selection_flag = False
selected: List[Fundamental] = [
f for f in fundamental if f.HasFundamentalData
and f.Market == 'usa'
and f.SecurityReference.ExchangeId in self.exchange_codes
and f.MarketCap != 0
]
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 not in self.m_price_data:
self.m_price_data[symbol] = RollingWindow[float](self.m_regression_period)
history: dataframe = self.History(symbol, self.m_regression_period * 30, Resolution.Daily)
if history.empty:
self.Log(f"Not enough data for {symbol} yet.")
continue
closes: Series = history.loc[symbol].close
closes_len = len(closes.keys())
# find monthly closes
for index, time_close in enumerate(closes.items()):
# index out of bounds check.
if index + 1 < closes_len:
date_month: int = time_close[0].date().month
next_date_month: int = closes.keys()[index + 1].month
# found last day of month
if date_month != next_date_month:
self.m_price_data[symbol].Add(time_close[1])
if not self.net_percent_share_change_series.IsReady:
return Universe.Unchanged
# forming the SSI index
# at this point, both market price data and net share change series data are ready
market_prices: np.ndarray = np.array([x for x in self.m_price_data[self.market]])
market_returns: np.ndarray = market_prices[:-1] / market_prices[1:] - 1
share_changes: np.ndarray = np.array([x for x in self.net_percent_share_change_series])[:-1]
ols_model = self.multiple_linear_regression(share_changes, market_returns)
# time series of residuals form the Speculation Sentiment Index
SSI: np.ndarray = ols_model.resid
SSI_sensitivity: Dict[Fundamental, float] = {}
self.daily_return: dataframe = pd.dataframe()
for stock in selected :
symbol: Symbol = stock.Symbol
if self.m_price_data[symbol].IsReady:
# at this point, both stock price data and net share change series data are ready
stock_prices: np.ndarray = np.array([x for x in self.m_price_data[symbol]])
stock_returns: np.ndarray = stock_prices[:-1] / stock_prices[1:] - 1
self.daily_return[stock] = stock_returns
# estimate monthly sensitivity to lagged SSI
ols_model = self.multiple_linear_regression(SSI[1:], stock_returns[:-1])
SSI_sensitivity[stock] = ols_model.params[1]
if len(SSI_sensitivity) >= self.quantile:
# sorting by SSI sensitivity
sorted_by_sensitivity: List[Fundamental] = sorted(SSI_sensitivity, key = SSI_sensitivity.get, reverse=True)
quantile: int = int(len(sorted_by_sensitivity) / self.quantile)
long: List[Fundamental] = []
short: List[Fundamental] = []
if SSI[0] > 0:
long = sorted_by_sensitivity[-quantile:]
short = sorted_by_sensitivity[:quantile]
else:
long = sorted_by_sensitivity[:quantile]
short = sorted_by_sensitivity[-quantile:]
# value 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
# magnitude of SSI scales the portfolio itself; average leverage is equal to zero
abs_ssi_values: np.ndarray = abs(SSI)
percentile: float = stats.percentileofscore(abs_ssi_values[1:], abs_ssi_values[0]) / 100.
leverage: float = 1 - ((0.5 - percentile) * 2) if percentile < 0.5 else 1 + ((percentile - 0.5) * 2)
self.weight = {
x : w * leverage for x, w in self.weight.items()
}
self.rebalance_flag = True
return list(self.weight.keys())
def OnData(self, data: Slice) -> None:
if (self.Time.date() > ETFSharesOutstanding.get_last_update_date()):
self.selection_flag = False
self.Liquidate()
return
# SO data came in
if self.shares_out_data in data and data[self.shares_out_data]:
# universe selection and rebalance is done once a month when new SO data comes
if self.Time.month != self.recent_month:
self.selection_flag = True
self.recent_month = self.Time.month
long_psc_sum: float = 0
short_psc_sum: float = 0
data_ready_ticker_cnt: int = 0 # n of tickers with SO data ready
for ticker in self.long_etfs + self.short_etfs:
so: float = data[self.shares_out_data].GetProperty(ticker)
# ticker has previous month's SO data
if self.previous_so[ticker] != 0:
data_ready_ticker_cnt += 1
percent_share_change: float = -1 + (so / self.previous_so[ticker])
if ticker in self.long_etfs:
long_psc_sum += percent_share_change
else:
short_psc_sum += percent_share_change
self.previous_so[ticker] = so
# every ticker had SO data ready
if data_ready_ticker_cnt == len(self.long_etfs + self.short_etfs):
psc_net: float = long_psc_sum - short_psc_sum
self.net_percent_share_change_series.Add(psc_net)
# rebalance once a month
if not self.rebalance_flag:
return
self.rebalance_flag = False
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 multiple_linear_regression(self, x:np.ndarray, y:np.ndarray, add_residual:bool = True):
x:np.ndarray = np.array(x).T
if add_residual:
x = sm.add_constant(x)
result = sm.OLS(endog=y, exog=x).fit()
return result
# Custom data
class ETFSharesOutstanding(PythonData):
_last_update_date: datetime.date = datetime(1,1,1).date()
@staticmethod
def get_last_update_date() -> datetime.date:
return ETFSharesOutstanding._last_update_date
def GetSource(self, config, date, isLiveMode):
return SubscriptionDataSource("data.quantpedia.com/backtesting_data/equity/etf_shares_outstanding.csv", SubscriptionTransportMedium.RemoteFile, FileFormat.Csv)
def Reader(self, config, line, date, isLiveMode):
data = ETFSharesOutstanding()
data.Symbol = config.Symbol
if not line[0].isdigit():
self.header_columns = line.split(';')
return None
split = line.split(';')
data.Time = datetime.strptime(split[0], "%Y-%m-%d") + timedelta(days=1)
data.Value = float(split[1])
for i, col in enumerate(self.header_columns[1:]): # skip date
data[col] = float(split[i+1])
# store last update date
if data.Time.date() > ETFSharesOutstanding._last_update_date:
ETFSharesOutstanding._last_update_date = data.Time.date()
return data
# Custom fee model.
class CustomFeeModel():
def GetOrderFee(self, parameters):
fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
return OrderFee(CashAmount(fee, "USD"))