“该策略投资于CRSP股票，使用基于波动率的十等分投资组合和斜率信号，根据市场状况每月重新平衡，以动态地在高、低或中等波动率投资组合之间切换。”
资产类别: 股票 | 地区: 美国 | 周期: 每月 | 市场: 股票 | 关键词: 择时、波动率
I. 策略概要
该策略投资于CRSP普通股，首先选择市值最大的1000只股票。股票根据其过去一个月的已实现波动率分为十等分投资组合。斜率定义为高波动率和低波动率投资组合之间的回报差异，用于确定市场状况。显著为正的斜率（t检验，12个月窗口，0.1%显著性）表示有利条件，促使转向最高波动率投资组合。显著为负的斜率表示不利条件，转向最低波动率投资组合。否则，默认持有中等波动率投资组合。该策略每月重新平衡，利用波动率信号动态适应不断变化的市场状况。
II. 策略合理性
高波动率投资组合的长期回报表现不佳，而低波动率投资组合受益于复利，随着时间的推移实现有吸引力的风险调整后表现。波动率十等分投资组合表现出随时间变化的回报，其表现受市场状况的强烈影响。在良好市场条件下，高波动率投资组合提供最高回报，而在不良条件下，低波动率投资组合由于损失最小而表现优异。市场状况通过波动率十等分回报的斜率推断，该斜率根据普遍趋势而变化。负斜率表示回报与波动率之间存在向下关系，有利于低波动率投资组合。相反，正斜率表示强劲增长，有利于高波动率投资组合。当条件中性时，中等波动率投资组合提供平衡的回报-波动率权衡，使其成为波动率择时策略中的最佳默认选择。
III. 来源论文
Low-Volatility Strategy: Can We Time the Factor? [点击查看论文]
- 梁宝玲（Poh Ling Neo）和郑清文（Chyng Wen Tee），新跃社科大学（Singapore University of Social Sciences），新加坡管理大学李光前商学院（Singapore Management University – Lee Kong Chian School of Business）
<摘要>
我们表明，波动率十等分投资组合回报曲线的斜率包含有价值的信息，可用于在不同市场条件下进行波动率择时。在良好（不良）市场条件下，高（低）波动率投资组合产生最高回报。我们接着设计了一种基于波动率十等分投资组合回报曲线斜率统计检验的波动率择时策略。波动率择时是通过在强劲增长时期激进，而在市场低迷时期保守来实现的。获得了卓越的业绩，波动率择时策略额外获得了4.1%的回报，导致累积财富增加了五倍，同时Sortini比率和信息比率也得到了统计学上的显著改善。作者还证明，高波动率投资组合中的股票与低波动率投资组合中的股票相比，相关性更强。因此，波动率择时策略的盈利能力可以归因于在熊市中成功持有多元化投资组合，而在牛市中持有集中式增长投资组合。
IV. 回测表现
| 年化回报 | 17.2% |
| 波动率 | 17.7% |
| β值 | 0.977 |
| 夏普比率 | 0.72 |
| 索提诺比率 | 0.333 |
| 最大回撤 | N/A |
| 胜率 | 68% |
V. 完整的 Python 代码
import numpy as np
from AlgorithmImports import *
from pandas.core.frame import dataframe
class TimingHighLowVolatility(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2000, 1, 1)
self.SetCash(100000)
self.period:int = 21
self.quantile:int = 10
self.slope_std_threshold:float = 4.
self.long:List[Symbol] = []
self.slopes:RollingWindow = RollingWindow[float](12)
self.data:Dict[Symbol, RollingWindow] = {}
market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
self.fundamental_count:int = 500
self.fundamental_sorting_key = lambda x: x.DollarVolume
self.selection_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)
self.settings.daily_precise_end_time = False
def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
for security in changes.AddedSecurities:
security.SetFeeModel(CustomFeeModel())
def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
# Update the rolling window every day.
for stock in fundamental:
symbol:Symbol = stock.Symbol
# Store monthly price.
if symbol in self.data:
self.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']
if len(selected) > self.fundamental_count:
selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
volatility:Dict[Symbol, float] = {}
momentum:Dict[Symbol, float] = {}
# Warmup price rolling windows.
for stock in selected:
symbol:Symbol = stock.Symbol
if symbol not in self.data:
self.data[symbol] = RollingWindow[float](self.period)
history:dataframe = self.History(symbol, self.period, Resolution.Daily)
if history.empty:
self.Log(f"Not enough data for {symbol} yet.")
continue
closes:pd.Series = history.loc[symbol].close
for time, close in closes.items():
self.data[symbol].Add(close)
if self.data[symbol].IsReady:
closes:np.ndarray = np.array(list(self.data[symbol]))
momentum[symbol] = closes[0] / closes[-1] - 1
returns:np.ndarray = closes[:-1] / closes[1:] - 1
volatility[symbol] = np.std(returns)
if len(momentum) >= self.quantile:
# Volaility sorting
sorted_by_vol:List[Symbol] = sorted(volatility, key = volatility.get, reverse = True)
quantile:int = int(len(sorted_by_vol) / self.quantile)
high_by_vol:List[Symbol] = sorted_by_vol[:quantile]
low_by_vol:List[Symbol] = sorted_by_vol[-quantile:]
mid_by_vol:List[Symbol] = [x for x in sorted_by_vol if x not in high_by_vol and x not in low_by_vol]
# Slope calc
high_vol_return:float = sum([momentum[x] for x in high_by_vol if x in momentum])
low_vol_return:float = sum([momentum[x] for x in low_by_vol if x in momentum])
slope:float = high_vol_return - low_vol_return
if self.slopes.IsReady:
slopes:List[float] = list(self.slopes)
slopes_mean:float = np.mean(slopes)
slopes_std:float = np.std(slopes)
if slope > slopes_mean + self.slope_std_threshold * slopes_std:
self.long = high_by_vol
elif slope < slopes_mean - self.slope_std_threshold * slopes_std:
self.long = low_by_vol
else:
self.long = mid_by_vol
self.slopes.Add(slope)
return self.long
def OnData(self, data: Slice) -> None:
if not self.selection_flag:
return
self.selection_flag = False
# order execution
portfolio:List[PortfolioTarget] = [PortfolioTarget(symbol, 1. / len(self.long)) for symbol in self.long if symbol in data and data[symbol]]
self.SetHoldings(portfolio, True)
self.long.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"))