“该策略投资于MSCI全球指数中波动性最小的股票，在每个行业内和每个行业内的股票之间进行等权重分配，每月重新平衡以优先考虑低风险、多元化的回报。”
资产类别: 股票 | 地区: 全球 | 周期: 每月 | 市场: 股票 | 关键词: 波动率
I. 策略概要
该策略以MSCI全球指数股票为目标，计算每个行业的三年波动率。它做多每个行业内波动率最低的十分位股票，从而在十个行业中创建一个等权重的投资组合。每个行业内的股票也等权重。投资组合每月重新平衡，优先选择低波动率股票以获得持续的风险调整回报，同时保持行业多元化。
II. 策略合理性
作者指出了基本金融理论中导致低波动率异常的五个误解。首先，投资者面临杠杆和卖空限制，限制了他们套利定价异常的能力。其次，许多投资者优先考虑的目标并非最大化绝对回报或最小化波动率。第三，无交易成本、无税收、完美可分割性和流动性的假设是不现实的。第四，投资者在投资期限上存在差异，影响决策。最后，认知偏差，如代表性、过度自信和彩票偏好，影响着大多数投资者，导致偏离理性行为，并导致金融市场低波动率异常的持续存在。
III. 来源论文
The Low Volatility Anomaly in Equity Sectors – 10 Years Later! [点击查看论文]
- Benoit Bellone 和 Raul Leote de Carvalho。Quantcube Technology。法国巴黎银行资产管理公司
<摘要>
在发现股票表现中的低波动率异常是一种应在每个行业而非忽略行业的绝对基础上考虑的现象十年后，我们提供了证据表明这一观察结果经受住了考验，并且如果说有什么变化的话，那就是它变得更加有效。
IV. 回测表现
| 年化回报 | 6.96% |
| 波动率 | 12.55% |
| β值 | 0.657 |
| 夏普比率 | 0.55 |
| 索提诺比率 | 0.38 |
| 最大回撤 | N/A |
| 胜率 | 59% |
V. 完整的 Python 代码
from AlgorithmImports import *
from pandas.core.frame import dataframe
from itertools import chain
class TheLowVolatilityAnomalyInEquitySectors(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2000, 1, 1)
self.SetCash(100000)
self.data:Dict[Symbol, SymbolData] = {}
self.longs:List[List[Symbol]] = []
self.period:int = 3 * 12 * 21 # Three years of daily returns
self.quantile:int = 10
self.leverage:int = 5
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.Settings.MinimumOrderMarginPortfolioPercentage = 0.
self.AddUniverse(self.FundamentalSelectionFunction)
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())
security.SetLeverage(self.leverage)
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].update(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]]
sectors = {}
# Warmup price rolling windows.
for stock in selected:
symbol:Symbol = stock.Symbol
if symbol not in self.data:
self.data[symbol] = SymbolData(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].update(close)
if self.data[symbol].is_ready():
sector = stock.AssetClassification.MorningstarSectorCode
if sector not in sectors:
sectors[sector] = []
sectors[sector].append(symbol)
for sector, symbols in sectors.items():
symbols_volatility:Dict[Symbol, float] = { symbol : self.data[symbol].volatility() for symbol in symbols}
quantile:int = int(len(symbols_volatility) / self.quantile)
sorted_by_volatility:List[Symbol] = [x[0] for x in sorted(symbols_volatility.items(), key=lambda item: item[1])]
self.longs.append(sorted_by_volatility[:quantile]) # Least volatile decile of sector
return list(set(chain.from_iterable(self.longs)))
def OnData(self, data: Slice) -> None:
if not self.selection_flag:
return
self.selection_flag = False
# Trade execution
self.Liquidate()
total_sectors:int = len(self.longs)
for long in self.longs:
long_length:int = len(long)
for symbol in long:
if symbol in data and data[symbol]:
self.SetHoldings(symbol, 1. / long_length / total_sectors)
self.longs.clear()
def Selection(self) -> None:
self.selection_flag = True
class SymbolData():
def __init__(self, period):
self._closes:RollingWindow = RollingWindow[float](period)
def update(self, close: float) -> None:
self._closes.Add(close)
def is_ready(self) -> bool:
return self._closes.IsReady
def volatility(self) -> float:
closes:np.ndarray = np.array(list(self._closes))
returns:np.ndarray = (closes[:-1] - closes[1:]) / closes[1:]
return np.std(returns)
class CustomFeeModel(FeeModel):
def GetOrderFee(self, parameters):
fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
return OrderFee(CashAmount(fee, "USD"))