“通过特质波动率交易A股，做多风险最低的十分位，做空风险最高的十分位，使用价值加权、每月重新平衡的投资组合，排除CSMAR数据库中的微型股。”
资产类别: 股票 | 地区: 中国 | 周期: 每月 | 市场: 股票 | 关键词: 特质、波动率、中国
I. 策略概要
投资范围包括上海和深圳证券交易所的人民币计价A股，不包括微型股（股票中市值最小的30%）。数据来源于CSMAR数据库。特质波动率的计算方法是：使用过去250个交易日的数据，将每日超额股票回报对截距、价值加权市场因子、规模因子和价值因子进行回归，然后计算残差的标准差。每个月，股票根据特质波动率分为十分位。该策略做多风险最低的股票（最低十分位）和做空风险最高的股票（最高十分位），形成价值加权投资组合并每月重新平衡。
II. 策略合理性
在中国，散户投资者的投机行为导致他们偏好风险较高、类似彩票的股票，这些股票往往被高估。学术研究表明，低风险股票的表现优于高风险股票，因为低风险股票往往被低估。作者探讨了中国市场特有的因素是否会影响异常策略。他们发现，由于多头和空头回报同样强劲，套利限制无法解释这些异常现象。此外，国有企业普遍存在并不会影响策略的有效性，因为因子溢价在国有和非国有股票中都存在。尽管股市改革降低了回报，但许多异常现象仍然显著，证明了这些策略的稳健性。
III. 来源论文
Anomalies in the China A-share Market [点击查看论文]
- Jansen, Maarten 和 Swinkels, Laurens 和 Zhou, Weili。Robeco 量化投资。鹿特丹伊拉斯姆斯大学（EUR）；Robeco 资产管理。Robeco 机构资产管理
<摘要>
本文阐明了中国A股市场与其他市场异常现象存在的异同。为此，我们考察了2000-2019年期间中国A股市场32种异常现象的存在。我们发现价值、风险和交易异常现象在中国A股中也存在。规模、质量和过去回报类别中的异常现象证据明显较弱，但强劲的残差动量和反转效应除外。我们记录到大多数异常现象不能用行业构成来解释，并且存在于大、中、小市值股票中。我们是第一个研究中国A股市场残差反转、回报季节性以及关联公司动量存在的。我们发现前两者有强劲的样本外证据，但后者没有。我们更详细地考察了中国A股市场的具体特征，例如卖空限制、国有企业的普遍存在以及股市改革的影响。这些特征似乎不是我们实证发现的重要驱动因素。
IV. 回测表现
| 年化回报 | 14.02% |
| 波动率 | 23.52% |
| β值 | 0.015 |
| 夏普比率 | 0.59 |
| 索提诺比率 | 0.105 |
| 最大回撤 | N/A |
| 胜率 | 51% |
V. 完整的 Python 代码
from AlgorithmImports import *
import numpy as np
import statsmodels.api as sm
#endregion
class IdiosyncraticVolatilityInChina(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2010, 1, 1)
self.SetCash(100_000)
self.period: int = 251 # Regression needs n-1 daily returns
self.data: Dict[Symbol, SymbolData] = {}
self.long: List[Symbol] = []
self.short: List[Symbol] = []
self.quantile: int = 5
self.leverage: int = 10
self.min_share_price: float = 3.
self.market_cap_quantile: int = 3
self.traded_percentage: float = .2
market: Symbol = self.AddEquity("SPY", Resolution.Daily).Symbol
# factors
self.last_market_factor: Tuple[List[Symbol], List[Symbol]] = None # long only dict
self.last_size_factor: Tuple[List[Symbol], List[Symbol]] = None # long/short lists
self.last_value_factor: Tuple[List[Symbol], List[Symbol]] = None # long/short lists
self.selection_flag: bool = False
self.UniverseSettings.Resolution = Resolution.Daily
self.AddUniverse(self.FundamentalSelectionFunction)
self.settings.daily_precise_end_time = False
self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
self.Schedule.On(self.DateRules.MonthStart(market), self.TimeRules.AfterMarketOpen(market), self.Selection)
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)
# rebalace monthly
if not self.selection_flag:
return Universe.Unchanged
selected: List[Fundamental] = [
f for f in fundamental if f.HasFundamentalData
and f.MarketCap != 0
and not np.isnan(f.ValuationRatios.PBRatio) and f.ValuationRatios.PBRatio != 0
and f.Market == 'usa'
and f.CompanyReference.BusinessCountryID == 'CHN'
and f.Price >= self.min_share_price
]
# exclude 30% of lowest stocks by MarketCap
sorted_by_market_cap: List[Fundamental] = sorted(selected, key = lambda x: x.MarketCap)
selected = sorted_by_market_cap[int(len(sorted_by_market_cap) / self.market_cap_quantile):]
market_cap: Dict[Symbol, float] = {}
value_factor: Dict[Symbol, float] = {}
for stock in selected:
symbol: Symbol = stock.Symbol
# Get stock's volumes
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: Series = history.loc[symbol].close
for time, close in closes.items():
self.data[symbol].update(close)
if self.data[symbol].is_ready():
market_cap[symbol] = stock.MarketCap
value_factor[symbol] = stock.ValuationRatios.PBRatio
if len(market_cap) < self.market_cap_quantile:
return Universe.Unchanged
# sort cap and size portfolios
quantile: int = int(len(market_cap) / self.market_cap_quantile)
sorted_by_cap: List[Symbol] = [x[0] for x in sorted(market_cap.items(), key=lambda item: item[1])]
sorted_by_value: List[Symbol] = [x[0] for x in sorted(value_factor.items(), key=lambda item: item[1])]
# factors are ready
if self.last_market_factor and self.last_size_factor and self.last_value_factor:
# create regression x from daily returns of
# value weight market factor of all selected chinese stocks,
# size factor and value factor
regression_x: List = [
self.DailyPerformanceValueWeight(self.last_market_factor),
self.FactorDailyPerformance(self.last_size_factor[0], self.last_size_factor[1]),
self.FactorDailyPerformance(self.last_value_factor[0], self.last_value_factor[1])
]
idiosyncratic_volatility: Dict[Symbol, float] = {}
# calculate idiosyncratic volatility for each stock
for symbol in market_cap:
daily_returns: np.ndarray = self.data[symbol].daily_returns()
regression_model = self.MultipleLinearRegression(regression_x, daily_returns)
idiosyncratic_volatility[symbol] = np.std(regression_model.resid)
if len(idiosyncratic_volatility) < self.quantile:
return Universe.Unchanged
# selection based on idiosyncratic volatility
quantile: int = int(len(idiosyncratic_volatility) / self.quantile)
sorted_by_idio_vol: List[Symbol] = [x[0] for x in sorted(idiosyncratic_volatility.items(), key=lambda item: item[1])]
# Long the bottom decile and short the top decile
self.long = sorted_by_idio_vol[:quantile]
self.short = sorted_by_idio_vol[-quantile:]
# Store factor symbols for next month's calculations.
self.last_market_factor = market_cap
self.last_size_factor = (sorted_by_cap[:quantile], sorted_by_cap[-quantile:])
self.last_value_factor = (sorted_by_value[:quantile], sorted_by_value[-quantile:])
return self.long + self.short
def OnData(self, slice: Slice) -> None:
if not self.selection_flag:
return
self.selection_flag = False
# trade execution
targets:List[PortfolioTarget] = []
for i, portfolio in enumerate([self.long, self.short]):
for symbol in portfolio:
if symbol in slice and slice[symbol]:
targets.append(PortfolioTarget(symbol, (((-1) ** i) / len(portfolio)) * self.traded_percentage))
self.SetHoldings(targets, True)
self.long.clear()
self.short.clear()
def Selection(self) -> None:
self.selection_flag = True
def DailyPerformanceValueWeight(self, market_cap: Dict[Symbol, float]) -> np.ndarray:
# Create numpy array with zeros
total_daily_returns: np.ndarray = np.zeros(self.period - 1)
total_cap: float = sum([x[1] for x in market_cap.items()])
for symbol, cap in market_cap.items():
# calculate weight for current stock
weight: float = cap / total_cap
# get daily returns of current stock
daily_returns: np.ndarray = self.data[symbol].daily_returns()
# multiply each daily return by weight
daily_returns: np.ndarray = daily_returns * weight
# add daily returns of current stock to total_daily_returns of portfolio
total_daily_returns += daily_returns
return total_daily_returns
def FactorDailyPerformance(self, long: List[Symbol], short: List[Symbol]) -> np.ndarray:
# create numpy array with zeros
total_daily_returns: np.ndarray = np.zeros(self.period - 1)
# go through each long and short stock
# Add daily returns of long stocks and sub daily returns of short stocks.
for long_sym, short_sym in zip(long, short):
total_daily_returns += self.data[long_sym].daily_returns()
total_daily_returns -= self.data[short_sym].daily_returns()
return total_daily_returns
def MultipleLinearRegression(self, x, y):
x: np.ndarray = np.array(x).T
x = sm.add_constant(x)
result = sm.OLS(endog=y, exog=x).fit()
return result
class SymbolData():
def __init__(self, period: int) -> None:
self.daily_prices: RollingWindow = RollingWindow[float](period)
def update(self, close: float) -> None:
self.daily_prices.Add(close)
def is_ready(self) -> bool:
return self.daily_prices.IsReady
def daily_returns(self) -> np.ndarray:
daily_prices: np.ndarray = np.array([x for x in self.daily_prices])
return (daily_prices[:-1] - daily_prices[1:]) / daily_prices[1:]
# Custom fee model
class CustomFeeModel(FeeModel):
def GetOrderFee(self, parameters):
fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
return OrderFee(CashAmount(fee, "USD"))