“该策略按波动率对美国股票进行排名，并将其分配到五分位数。它在每个波动率五分位数内做多低相关性股票，做空高相关性股票，每月重新平衡以优化绩效。”
资产类别: 股票 | 地区: 美国 | 周期: 每月 | 市场: 股票 | 关键词: 相关性效应
I. 策略概要
该策略使用美国股票，按上个月估计的波动率排名，并分为五个五分位数。每个五分位数进一步分为低市场相关性投资组合和高市场相关性投资组合。低相关性股票在低相关性投资组合中被赋予更大的权重，而高相关性股票在高相关性投资组合中被赋予更大的权重。该策略在每个波动率五分位数内做多低相关性股票，做空高相关性股票。五分位数等权重，投资组合每月重新平衡，以捕捉波动率和相关性动态。
II. 策略合理性
低风险效应可以用两种理论解释：杠杆约束和行为效应。该策略侧重于杠杆约束，将贝塔除以波动率和市场相关性。通过使用因子BAC（押注反对相关性）中和波动率，该策略在美国和国际上均表现良好，支持了杠杆约束理论。当保证金债务较低时，BAB和BAC显示出更高的回报，表明当杠杆约束较高时，投资者偏爱低风险股票。使用LMAX和SMAX因子测试的行为理论在解释该效应方面的重要性较低，杠杆约束是低风险回报更稳健的解释。
III. 来源论文
Betting Against Correlation: Testing Theories of the Low-Risk Effect [点击查看论文]
- 阿斯内斯（Asness）、弗拉齐尼（Frazzini）、戈姆森（Gormsen）、彼得森（Pedersen），AQR资本管理公司（AQR Capital Management, LLC），芝加哥大学布斯商学院（University of Chicago – Booth School of Business），AQR资本管理公司；哥本哈根商学院金融系（Copenhagen Business School – Department of Finance）；经济政策研究中心（Centre for Economic Policy Research, CEPR）
<摘要>
我们测试低风险效应是否由（a）杠杆约束驱动，因此风险应使用贝塔衡量，还是由（b）行为效应驱动，因此风险应由特质风险衡量。贝塔取决于波动率和相关性，其中只有波动率与特质风险相关。我们引入了一个新的押注反对相关性（BAC）因子，该因子特别适合区分杠杆约束与彩票解释。BAC在美国和国际上均产生了强劲的表现，支持了杠杆约束理论。同样，我们构建了新的因子SMAX来隔离彩票需求，该因子也产生了正回报。与杠杆和彩票理论共同促成低风险效应一致，我们发现BAC与保证金债务相关，而特质风险因子与情绪相关。
IV. 回测表现
| 年化回报 | 12.28% |
| 波动率 | 13.2% |
| β值 | 1.616 |
| 夏普比率 | 0.93 |
| 索提诺比率 | 0.297 |
| 最大回撤 | N/A |
| 胜率 | 53% |
V. 完整的 Python 代码
import numpy as np
from AlgorithmImports import *
from typing import Dict, List
class BettingAgainstCorrelationEffect(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2000, 1, 1)
self.SetCash(100000)
self.leverage:int = 5
self.quantile:int = 5
self.min_share_price:int = 5
self.fundamental_sorting_key = lambda x: x.DollarVolume
self.fundamental_count:int = 500
self.quintile_internal_count:int = 10 # Pick n low corraleted stocks and n high corraleted stocks in each quintile.
self.data:Dict[Symbol, RollingWindow] = {} # Storing daily prices
self.long:List[Symbol] = []
self.short:List[Symbol] = []
self.symbol:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
self.period:int = 21
self.SetWarmUp(self.period)
self.market_prices:RollingWindow = RollingWindow[float](self.period)
self.selection_flag:bool = False
self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
self.UniverseSettings.Resolution = Resolution.Daily
self.AddUniverse(self.FundamentalSelectionFunction)
self.Schedule.On(self.DateRules.MonthStart(self.symbol), self.TimeRules.AfterMarketOpen(self.symbol), 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 Prices in RollingWindow
for stock in fundamental:
symbol:Symbol = stock.Symbol
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.Price > self.min_share_price 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]]
# Warmup price rolling windows.
for stock in selected:
symbol:Symbol = stock.Symbol
if symbol in self.data:
continue
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:Series = history.loc[symbol].close
for time, close in closes.items():
self.data[symbol].Add(close)
if not self.market_prices.IsReady:
return Universe.Unchanged
market_prices:List[float] = [x for x in self.market_prices]
correlation:Dict[Symbol, float] = {}
volatility:Dict[Symbol, float] = {}
for stock in selected:
symbol:Symbol = stock.Symbol
if not self.data[symbol].IsReady:
continue
closes:List[float] = [x for x in self.data[symbol]]
volatility[symbol] = self.Volatility(closes)
correlation[symbol] = np.correlate(closes, market_prices)
# Volatility sorting
sorted_by_vol:List[Symbol] = [x[0] for x in sorted(volatility.items(), key=lambda item: item[1], reverse=True)]
quantile:int = int(len(sorted_by_vol)/self.quantile)
# Group splitting
volatility_groups:List[List[Symbol]] = [sorted_by_vol[x:x+quantile] for x in range(0, len(sorted_by_vol),quantile)]
group_count:int = len(volatility_groups)
# Long = array of arrays = groups by 10 stocks to long
# Short = array of arrays = groups by 10 stocks to short
for vol_group in volatility_groups:
symbols:List[Symbol] = [x for x in vol_group]
sorted_by_correlation:List[Symbol] = sorted(symbols, key = lambda x: correlation[x], reverse = True)
# Go long stocks with highest correlations and short stocks with lowest correlation
self.long.append([x for x in sorted_by_correlation][:self.quintile_internal_count])
self.short.append([x for x in sorted_by_correlation][-self.quintile_internal_count:])
long_short:List[Symbol] = []
for group in self.long + self.short:
for symbol in group:
long_short.append(symbol)
return long_short
def OnData(self, data: Slice) -> None:
# store daily market price
if self.symbol in data and data[self.symbol]:
price:float = data[self.symbol].Value
self.market_prices.Add(price)
if not self.selection_flag:
return
self.selection_flag = False
# Trade execution
self.Liquidate()
if len(self.long) == 0 or len(self.short) == 0:
self.Liquidate()
return
long_length:int = len(self.long)
short_length:int = len(self.short)
equity_per_long_group:float = float(1 / long_length)
equity_per_short_group:float = float(1 / short_length)
for group in self.long:
count:int = int(len(group))
weight_index:float = 1 / sum([x for x in range(count+1)])
for symbol in group:
if symbol in data and data[symbol]:
self.SetHoldings(symbol, equity_per_long_group * (count*weight_index))
count -= 1
for group in self.short:
count = int(len(group))
weight_index = 1 / sum([x for x in range(count+1)])
count = 1
for symbol in group:
if symbol in data and data[symbol]:
self.SetHoldings(symbol, -equity_per_short_group * (count*weight_index))
count -= 1
self.long.clear()
self.short.clear()
def Selection(self) -> None:
# Monthly rebalance
self.selection_flag = True
def Volatility(self, values) -> float:
values = np.array(values)
returns = (values[:-1] - values[1:]) / values[1:]
return np.std(returns)
# Custom fee model
class CustomFeeModel(FeeModel):
def GetOrderFee(self, parameters):
fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
return OrderFee(CashAmount(fee, "USD"))