“该策略涉及使用标普500股指期货期权构建跳跃风险模拟投资组合，根据股票的跳跃风险贝塔将股票分为五等份，并做多低贝塔股票，做空高贝塔股票。”
资产类别: 股票 | 地区: 美国 | 周期: 每月 | 市场: 股票 | 关键词: 股票
I. 策略概要
该投资范围包括价格高于1美元的CRSP股票。跳跃风险模拟投资组合是使用标普500股指期货期权构建的，重点关注市场中性、vega中性和gamma正策略。该策略涉及做多到期日为T1的平值跨式期权，做空到期日为T2的跨式期权，以确保整体投资组合的中性。跨式期权回报每日计算，通过选择未来两个月到期的平值看涨和看跌期权。股票每月根据其跳跃风险贝塔分为五等份，做多最低五等份的股票，做空最高五等份的股票。投资组合每月重新平衡，并采用价值加权。
II. 策略合理性
该跳跃风险异常现象的解释是，对市场跳跃风险高度敏感的股票为风险规避型投资者提供了对冲机会。由于风险的市场价格为负，这些股票的回报较低。这与经济直觉相符，因为投资者愿意为跳跃风险保护支付费用。研究发现了一个统计学和经济学上显著的异常现象，随着对市场跳跃风险敏感度的增加，业绩会下降，在五分位数组中呈现出稳健的趋势。这表明对跳跃风险敏感度高的股票会获得较低的回报，为跳跃风险补偿理论提供了证据。
III. 来源论文
Aggregate Jump and Volatility Risk in the Cross-Section of Stock Returns [点击查看论文]
- MARTIJN CREMERS, MICHAEL HALLING, and DAVID WEINBAUM。圣母大学；ECGI，卢森堡大学，雪城大学
<摘要>
我们通过构建可投资的期权交易策略来检验股票回报截面中总跳跃风险和波动率风险的定价，这些策略加载一个因子但与另一个因子正交。总跳跃风险和波动率风险都有助于解释预期回报的变化。与理论一致，对跳跃风险和波动率风险敏感度高的股票具有较低的预期回报。两者可以单独衡量，并且在经济上都很重要，跳跃（波动率）因子载荷增加两个标准差与预期年股票回报下降3.5%至5.1%（2.7%至2.9%）相关。
IV. 回测表现
| 年化回报 | 8.9% |
| 波动率 | 17.21% |
| β值 | -0.051 |
| 夏普比率 | 0.52 |
| 索提诺比率 | -0.143 |
| 最大回撤 | N/A |
| 胜率 | 50% |
V. 完整的 Python 代码
from AlgorithmImports import *
from pandas.tseries.offsets import BDay
from typing import List, Dict
import statsmodels.api as sm
# endregion
class JumpRiskinStocks(QCAlgorithm):
def Initialize(self) -> None:
self.SetStartDate(2010, 1, 1)
self.SetCash(100000)
self.required_exchanges: List[str] = ['NYS', 'NAS', 'ASE']
self.tickers_to_ignore: List[str] = ['KELYB']
market: Symbol = self.AddEquity("SPY", Resolution.Daily).Symbol
self.period: int = 12 * 21
self.min_share_price: int = 1
self.leverage: int = 5
self.quantile: int = 5
self.price_data: Dict[Symbol, RollingWindow] = {}
self.weight: Dict[Symbol, float] = {}
self.underlying_options_strategy: Symbol = self.AddData(QuantpediaEquity, '530_NYSE_MAPPED', Resolution.Daily).Symbol
self.price_data[self.underlying_options_strategy] = RollingWindow[float](self.period)
self.fundamental_count: int = 3000
self.fundamental_sorting_key = lambda x: x.MarketCap
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(market), self.TimeRules.BeforeMarketClose(market, 0), 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]:
QP_equity_last_update_date: Dict[Symbol, datetime.date] = QuantpediaEquity.get_last_update_date()
# check if custom data is still coming
if self.Securities[self.underlying_options_strategy].GetLastData() and self.Time.date() > QP_equity_last_update_date[self.underlying_options_strategy]:
self.Liquidate()
return Universe.Unchanged
# 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].Add(stock.AdjustedPrice)
if not self.selection_flag:
return Universe.Unchanged
selected: List[Fundamental] = [
x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and x.Price > self.min_share_price and \
x.MarketCap != 0 and x.SecurityReference.ExchangeId in self.required_exchanges and x.Symbol.Value not in self.tickers_to_ignore
]
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.price_data:
self.price_data[symbol] = RollingWindow[float](self.period)
if not self.price_data[self.underlying_options_strategy].IsReady:
return Universe.Unchanged
x_prices: np.ndarray = np.array(list(self.price_data[self.underlying_options_strategy])[:-1])
x: np.ndarray = x_prices[:-1] / x_prices[1:] - 1
price_data: Dict[Symbol, List[float]] = {
stock.Symbol: np.array(list(self.price_data[stock.Symbol])[1:])
for stock in selected
if self.price_data[stock.Symbol].IsReady
and stock.Symbol != self.underlying_options_strategy
}
returns: Dict[Symbol, float] = {symbol : prices[:-1] / prices[1:] - 1 for symbol, prices in price_data.items()}
y: np.ndarray = np.array(list(zip(*[[i for i in x] for x in returns.values()])))
model: RegressionResultWrapper = self.multiple_linear_regression(x, y)
beta_values: np.ndarray = model.params[1]
equity_beta: Dict[Symbol, float] = { symbol: beta_values[i] for i, symbol in enumerate(price_data.keys()) }
if len(equity_beta) <= self.quantile:
return Universe.Unchanged
# sort by beta
sorted_by_beta: List[Tuple[Symbol, float]] = sorted(equity_beta, key=equity_beta.get, reverse=True)
quantile: int = int(len(sorted_by_beta) / self.quantile)
long: List[Symbol] = sorted_by_beta[-quantile:]
short: List[Symbol] = sorted_by_beta[:quantile]
# calculate weights
for i, portfolio in enumerate([long, short]):
for symbol in portfolio:
self.weight[symbol] = ((-1) ** i) / len(portfolio)
return list(self.weight.keys())
def OnData(self, data: Slice) -> None:
# store underlying strategy price - causing one day lag behind FundamentalSelectionFunction
if self.underlying_options_strategy in data and data[self.underlying_options_strategy]:
self.price_data[self.underlying_options_strategy].Add(data[self.underlying_options_strategy].Value)
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:
self.selection_flag = True
def multiple_linear_regression(self, x: np.ndarray, y: np.ndarray):
x: np.ndarray = sm.add_constant(x, has_constant='add')
result: RegressionResultWrapper = sm.OLS(endog=y, exog=x).fit()
return result
# NOTE: IMPORTANT: Data order must be ascending (datewise)
class QuantpediaEquity(PythonData):
_last_update_date:Dict[Symbol, datetime.date] = {}
@staticmethod
def get_last_update_date() -> Dict[Symbol, datetime.date]:
return QuantpediaEquity._last_update_date
def GetSource(self, config: SubscriptionDataConfig, date: datetime, isLiveMode: bool) -> SubscriptionDataSource:
return SubscriptionDataSource("data.quantpedia.com/backtesting_data/equity/quantpedia_strategies/{0}.csv".format(config.Symbol.Value), SubscriptionTransportMedium.RemoteFile, FileFormat.Csv)
def Reader(self, config: SubscriptionDataConfig, line: str, date: datetime, isLiveMode: bool) -> BaseData:
data = QuantpediaEquity()
data.Symbol = config.Symbol
if not line[0].isdigit(): return None
split: str = line.split(';')
data.Time = datetime.strptime(split[0], "%Y-%m-%d") + timedelta(days = 1)
data['price'] = float(split[1])
data.Value = float(split[1])
if config.Symbol not in QuantpediaEquity._last_update_date:
QuantpediaEquity._last_update_date[config.Symbol] = datetime(1,1,1).date()
if data.Time.date() > QuantpediaEquity._last_update_date[config.Symbol]:
QuantpediaEquity._last_update_date[config.Symbol] = data.Time.date()
return data
# Custom fee model
class CustomFeeModel(FeeModel):
def GetOrderFee(self, parameters: OrderFeeParameters) -> OrderFee:
fee: float = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
return OrderFee(CashAmount(fee, "USD"))