“BAB策略涉及做多低贝塔股票,做空高贝塔股票,投资组合每月重新平衡。使用过去的收益来预测波动率,目标波动率为12%。”

I. 策略概要

BAB(反贝塔)策略涉及基于低贝塔和高贝塔股票构建投资组合,遵循Frazini和Pedersen的原则。买入低贝塔股票,卖空高贝塔股票。证券按其贝塔值排名,投资组合由贝塔值低于或高于中位数的股票组成。BAB投资组合是自筹资金的零贝塔投资组合,需要短期国库券的空头头寸来平衡投资组合。投资组合中的证券按其贝塔值加权。过去21天的已实现波动率用于预测未来波动率,并调整回报以维持12%的目标波动率。投资组合每月重新平衡,根据上个月的波动率调整其权重。该策略旨在通过瞄准稳定的波动率水平来利用风险回报权衡。

II. 策略合理性

将BAB的风险分解为市场风险和特定风险的学术论文发现,可预测的部分是特定的风险。此外,仅按特定风险调整的策略的业绩与使用总风险调整的策略非常相似。因此,学术论文得出结论,特定风险是风险管理收益的来源。

III. 来源论文

Managing the risk of the “betting-against-beta” anomaly: does it pay to bet against beta? [点击查看论文]

<摘要>

我们研究了反贝塔异常的风险动态。该策略显示出风险的强烈且可预测的时间变化,并且没有风险回报权衡。利用这一点的风险管理策略实现了1.28的年化夏普比率,相对于原始策略,信息比率高达0.94。市场、规模、价值、盈利能力和投资因素的类似策略平均仅实现了0.15的较低信息比率。风险调整带来的巨大经济效益与动量策略相似,并将这两个异常与其他股票因素区分开来。将风险分解为市场和特定成分,我们发现特定成分驱动了我们的结果。

IV. 回测表现

年化回报22.35%
波动率16.99%
β值0.782
夏普比率1.32
索提诺比率0.187
最大回撤-66.14%
胜率49%

V. 完整的 Python 代码

from AlgorithmImports import *
import numpy as np
from math import sqrt
import pandas as pd
from scipy import stats
from typing import Dict, List

class TimingBettingAgainstBetaAnomaly(QCAlgorithm):

    def Initialize(self):
        self.SetStartDate(2000, 1, 1)
        self.SetCash(100000)

        self.exchange_codes:List[str] = ['NYS']	

        # Daily price data.
        self.data:Dict[Symbol, RollingWindow] = {}
        self.period:int = 21

        self.leverage:int = 10
        self.min_share_price:int = 5

        # Warmup market daily data.
        self.symbol:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
        self.data[self.symbol] = RollingWindow[float](self.period)
        history:dataframe = self.History(self.symbol, self.period, Resolution.Daily)
        if history.empty:
            self.Log(f"Not enough data for {self.symbol} yet")
        else:
            closes:Series = history.loc[self.symbol].close
            for time, close in closes.items():
                self.data[self.symbol].Add(close)
        
        self.target_volatility:float = 0.12
        
        self.weight:Dict[Symbol, float] = {}
        
        self.fundamental_sorting_key = lambda x: x.DollarVolume
        self.fundamental_count:int = 250
        
        self.selection_flag = False
        self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
        self.UniverseSettings.Resolution = Resolution.Daily
        self.AddUniverse(self.FundamentalSelectionFunction)
        self.Schedule.On(self.DateRules.MonthEnd(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 the rolling window every day.
        for stock in fundamental:
            symbol:Symbol = stock.Symbol

            if symbol in self.data:
                # Store daily price.
                self.data[symbol].Add(stock.AdjustedPrice)
        
        # Selection once a month.
        if not self.selection_flag:
            return Universe.Unchanged
        
        # selected = [x.Symbol for x in fundamental if x.HasFundamentalData and x.Market == 'usa']
        selected:List[Fundamental] = [
            x for x in fundamental if x.HasFundamentalData and x.Price > self.min_share_price and x.Market == 'usa' \
            and x.MarketCap != 0 and x.SecurityReference.ExchangeId in self.exchange_codes
        ]
            
        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.data:
                self.data[symbol] = RollingWindow[float](self.period)
                history = self.History(symbol, self.period, Resolution.Daily)
                if history.empty:
                    self.Log(f"Not enough data for {symbol} yet")
                    continue
                closes = history.loc[symbol].close
                for time, close in closes.items():
                    self.data[symbol].Add(close)

        if not self.data[self.symbol].IsReady:
            return Universe.Unchanged
        
        market_returns:List[float] = []
        market_closes:np.ndarray = np.array([x for x in self.data[self.symbol]])
        market_returns = (market_closes[:-1] - market_closes[1:]) / market_closes[1:]
        
        beta:Dict[Symbol, float] = {}
        for stock in selected:
            symbol:Symbol = stock.Symbol
            
            # Data is ready.
            if self.data[symbol].IsReady and len(market_returns) != 0:
                stock_closes:np.ndarray = np.array([x for x in self.data[symbol]])
                stock_returns:np.ndarray = (stock_closes[:-1] - stock_closes[1:]) / stock_closes[1:]
                
                # cov = np.cov(market_returns, stock_returns)[0][1]
                # market_variance = np.std(market_returns) ** 2
                # beta[symbol] = cov / market_variance
                
                slope, intercept, r_value, p_value, std_err = stats.linregress(market_returns, stock_returns)
                beta[symbol] = slope

        # Beta diff calc.
        beta_median:float = np.median([x[1] for x in beta.items()])
        long_diff:List[Tuple[Symbol, float]] = [(x[0], x[1] - beta_median) for x in beta.items() if x[1] >= beta_median]
        short_diff:List[Tuple[Symbol, float]] = [(x[0], beta_median - x[1]) for x in beta.items() if x[1] < beta_median]

        # Beta diff weighting.
        for i, portfolio in enumerate([long_diff, short_diff]):
            total_diff:float = sum(list(map(lambda x: x[1], portfolio)))
            for symbol, diff in portfolio:
                self.weight[symbol] = ((-1)**i) * diff / total_diff
        
        return [x[0] for x in self.weight.items()]
        
    def OnData(self, data: Slice) -> None:
        # Market daily data is stored in fundamental.
        if not self.selection_flag:
            return
        self.selection_flag = False

        # Portfolio volatility calc.
        df:dataframe = pd.dataframe()
        weights:List[float] = []
        for symbol, w in self.weight.items():
            df[str(symbol)] = [x for x in self.data[symbol]]
            weights.append(w)
        
        weights = np.array(weights)
        
        daily_returns:dataframe = df.pct_change()
        portfolio_vol:flaot = np.sqrt(np.dot(weights.T, np.dot(daily_returns.cov() * 21, weights.T)))
        
        leverage:float = self.target_volatility / portfolio_vol
        
        # trade execution
        portfolio:List[PortfolioTarget] = [PortfolioTarget(symbol, w * leverage) 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

# Custom fee model.
class CustomFeeModel(FeeModel):
    def GetOrderFee(self, parameters):
        fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
        return OrderFee(CashAmount(fee, "USD"))

发表评论

了解 Quant Buffet 的更多信息

立即订阅以继续阅读并访问完整档案。

继续阅读