“该策略涉及对低贝塔股票(加杠杆)进行多头头寸,对高贝塔股票(去杠杆)进行空头头寸,使用贝塔排名,并在中国A股市场每月重新平衡。”

I. 策略概要

该投资范围包括中国A股。BAB投资组合的创建分为三个步骤:根据股票的事前市场贝塔进行排名,根据贝塔排名分配权重,并使用贝塔平价方法重新调整投资组合。低贝塔投资组合做多并加杠杆,而高贝塔投资组合做空并去杠杆。投资组合每月重新平衡,股票按排名加权。该策略涉及对低贝塔股票采取杠杆头寸,对高贝塔股票采取去杠杆头寸,旨在从两者之间的风险敞口差异中获利。

II. 策略合理性

中国股市拥有庞大的散户投资者基础,容易受到过度自信等偏差的影响,尤其是来自缺乏经验的交易者。过度自信导致过度交易和对高风险股票的更高需求,从而产生“逆贝塔”效应。由于不成熟投资者的主导地位和政策不确定性,新兴市场尤其容易受到此类偏差的影响。这项研究表明,投资者过度自信解释了中国证券市场线(SML)的负斜率,特别是当自归因偏差(即先前的成功进一步增强过度自信)放大时。低贝塔异常主要由股票换手率驱动,因为高贝塔股票交易更频繁,风险调整后的回报更低。在考虑换手率后,低贝塔异常消失。

III. 来源论文

Investor Overconfidence and the Security Market Line: New Evidence from China [点击查看论文]

<摘要>

本文记录了中国证券市场线(SML)的显著向下倾斜,这比美国典型的“扁平化”SML更令人费解,并且与现有的低贝塔异常理论不符。我们表明,投资者过度自信为解决中国的这一难题提供了一些希望:在时间序列维度上,当投资者变得更加过度自信时,SML的斜率变得更加“倒置”。这种动态过度自信效应因有偏的自我归因而加剧。作为横截面上过度自信的一个普遍症状,高贝塔股票也是交易最活跃的股票。在考虑交易量后,在公司和投资组合层面都不再存在低贝塔异常。共同基金的证据强化了这样一种观点,即机构投资者通过避开高贝塔股票并押注低贝塔股票以获得卓越表现,从而积极利用向下倾斜的SML的投资组合含义。

IV. 回测表现

年化回报12.54%
波动率32.67%
β值-0.041
夏普比率0.99
索提诺比率0.041
最大回撤N/A
胜率57%

V. 完整的 Python 代码

import numpy as np
from scipy import stats
from AlgorithmImports import *
from typing import List, Dict
from pandas.core.frame import dataframe
from pandas.core.series import Series
import data_tools
class BettingAgainstBetaInChina(QCAlgorithm):
    def Initialize(self) -> None:
        self.SetStartDate(2010, 1, 1)
        self.SetCash(100_000)
        
        self.leverage: float = 5
        self.period: int = 12 * 21 * 5 # 5 years of daily data
        self.traded_portion: float = 0.2
        self.data: Dict[Symbol, data_tools.SymbolData] = {}
        self.weight: Dict[Symbol, float] = {}
    
        symbol: Symbol = self.AddEquity("SPY", Resolution.Daily).Symbol
        
        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(symbol), self.TimeRules.AfterMarketOpen(symbol), self.Selection)
        
    def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
        for security in changes.AddedSecurities:
            security.SetFeeModel(data_tools.CustomFeeModel())
            security.SetLeverage(self.leverage)
            
    def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
        for stock in fundamental:
            symbol: Symbol = stock.Symbol
            
            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.CompanyReference.BusinessCountryID == 'CHN'
        ]
        
        for stock in selected:
            symbol: Symbol = stock.Symbol
            if symbol in self.data:
                continue
            self.data[symbol] = data_tools.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: Series = history.loc[symbol].close
            for time, close in closes.items():
                self.data[symbol].update(close)
        daily_returns: Dict[Symbol, np.ndarray] = {
            x.Symbol : self.data[x.Symbol].daily_returns() for x in selected if self.data[x.Symbol].is_ready()
        }
        
        if len(daily_returns) == 0:
            return Universe.Unchanged
            
        daily_returns_for_mean: np.ndarray = np.array([x[1] for x in daily_returns.items()])
        mean_daily_returns: List[float] = [np.mean(daily_returns_for_mean[:, i]) for i in range(len(daily_returns_for_mean[0]))]
        
        betas: Dict[Symbol, float] = {}
        for symbol, returns in daily_returns.items():
            regression_result = stats.linregress(mean_daily_returns, returns)
            betas[symbol] = regression_result[0]
            
        beta_median: float = np.median([x[1] for x in betas.items()])
        sorted_by_beta: List[Symbol] = [x[0] for x in sorted(betas.items(), key = lambda item: item[1], reverse = True)]
        
        # portfolios
        long: List[Symbol] = []
        short: List[Symbol] = []
        
        total_long_rank: float = 0.
        total_short_rank: float = 0.
        
        for symbol in sorted_by_beta:
            if betas[symbol] < beta_median:
                total_long_rank += betas[symbol]
                long.append(symbol)
                    
            elif betas[symbol] > beta_median:
                total_short_rank += betas[symbol]
                short.append(symbol)
        
        # Lowest beta has highest weight.
        # long and short lists are sorted ascending according to beta value.
        long_weights: List[float] = reversed([(betas[x] / total_long_rank) for x in long])
        short_weights: List[float] = reversed([(betas[x] / total_short_rank) for x in short])
        
        for i, portfolio_weights in enumerate([zip(long, long_weights), zip(short, short_weights)]):
            for symbol, weight in portfolio_weights:
                self.weight[symbol] = ((-1) ** i) * weight * self.traded_portion
        
        return long + short
        
    def OnData(self, data: Slice) -> None:
        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
class SymbolData():
    def __init__(self, period: int) -> None:
        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 daily_returns(self) -> np.ndarray:
        closes: np.ndarray = np.array([x for x in self.closes])
        returns: np.ndarray = (closes[:-1] - closes[1:]) / closes[1:]
        return returns
        
# 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"))
		

发表评论

了解 Quant Buffet 的更多信息

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

继续阅读