股票 · 中国 · 股票 · 每月 · 中国 Free

中国市场中的特质波动率

登录后收藏

学术论文

Anomalies in the China A-share market

作者Anomalies in the China A-share Market [点击查看论文]

机构

?Robeco Asset Management
NLErasmus University Rotterdam
?Erasmus University Rotterdam (EUR)
?Robeco Quantitative Investments

论文链接https://papers.ssrn.com/sol3/papers.cfm?abstract_id=3810114

策略概要

投资范围包括上海和深圳证券交易所的人民币计价A股，不包括微型股（股票中市值最小的30%）。数据来源于CSMAR数据库。特质波动率的计算方法是：使用过去250个交易日的数据，将每日超额股票回报对截距、价值加权市场因子、规模因子和价值因子进行回归，然后计算残差的标准差。每个月，股票根据特质波动率分为十分位。该策略做多风险最低的股票（最低十分位）和做空风险最高的股票（最高十分位），形成价值加权投资组合并每月重新平衡。

II. 策略合理性

在中国，散户投资者的投机行为导致他们偏好风险较高、类似彩票的股票，这些股票往往被高估。学术研究表明，低风险股票的表现优于高风险股票，因为低风险股票往往被低估。作者探讨了中国市场特有的因素是否会影响异常策略。他们发现，由于多头和空头回报同样强劲，套利限制无法解释这些异常现象。此外，国有企业普遍存在并不会影响策略的有效性，因为因子溢价在国有和非国有股票中都存在。尽管股市改革降低了回报，但许多异常现象仍然显著，证明了这些策略的稳健性。

回测表现

波动率23.52%

夏普比率0.59

索提诺比率0.105

胜率51%

完整 Python 代码

1from AlgorithmImports import *
2import numpy as np
3import statsmodels.api as sm
4#endregion
5class IdiosyncraticVolatilityInChina(QCAlgorithm):
6def Initialize(self):
7self.SetStartDate(2010, 1, 1)
8self.SetCash(100_000)
9
10self.period: int = 251 # Regression needs n-1 daily returns
11
12self.data: Dict[Symbol, SymbolData] = {}
13self.long: List[Symbol] = []
14self.short: List[Symbol] = []
15self.quantile: int = 5
16self.leverage: int = 10
17self.min_share_price: float = 3.
18self.market_cap_quantile: int = 3
19self.traded_percentage: float = .2
20market: Symbol = self.AddEquity("SPY", Resolution.Daily).Symbol
21
22# factors
23self.last_market_factor: Tuple[List[Symbol], List[Symbol]] = None  # long only dict
24self.last_size_factor: Tuple[List[Symbol], List[Symbol]] = None    # long/short lists
25self.last_value_factor: Tuple[List[Symbol], List[Symbol]] = None   # long/short lists
26
27self.selection_flag: bool = False
28self.UniverseSettings.Resolution = Resolution.Daily
29self.AddUniverse(self.FundamentalSelectionFunction)
30self.settings.daily_precise_end_time = False
31self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
32self.Schedule.On(self.DateRules.MonthStart(market), self.TimeRules.AfterMarketOpen(market), self.Selection)
33def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
34for security in changes.AddedSecurities:
35    security.SetFeeModel(CustomFeeModel())
36    security.SetLeverage(self.leverage)
37    
38def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
39# update the rolling window every day
40for stock in fundamental:
41    symbol: Symbol = stock.Symbol
42    
43    # Store monthly price.
44    if symbol in self.data:
45        self.data[symbol].update(stock.AdjustedPrice)
46
47# rebalace monthly
48if not self.selection_flag:
49    return Universe.Unchanged
50selected: List[Fundamental] = [
51    f for f in fundamental if f.HasFundamentalData 
52    and f.MarketCap != 0 
53    and not np.isnan(f.ValuationRatios.PBRatio) and f.ValuationRatios.PBRatio != 0 
54    and f.Market == 'usa' 
55    and f.CompanyReference.BusinessCountryID == 'CHN' 
56    and f.Price >= self.min_share_price
57]
58    
59# exclude 30% of lowest stocks by MarketCap
60sorted_by_market_cap: List[Fundamental] = sorted(selected, key = lambda x: x.MarketCap)
61selected = sorted_by_market_cap[int(len(sorted_by_market_cap) / self.market_cap_quantile):]
62
63market_cap: Dict[Symbol, float] = {}
64value_factor: Dict[Symbol, float] = {}
65
66for stock in selected:
67    symbol: Symbol = stock.Symbol
68    
69    # Get stock's volumes
70    if symbol not in self.data:
71        self.data[symbol] = SymbolData(self.period)
72        
73        history: DataFrame = self.History(symbol, self.period, Resolution.Daily)
74        
75        if history.empty:
76            self.Log(f"Not enough data for {symbol} yet")
77            continue
78        
79        closes: Series = history.loc[symbol].close
80        for time, close in closes.items():
81            self.data[symbol].update(close)
82    
83    if self.data[symbol].is_ready():
84        market_cap[symbol] = stock.MarketCap
85        value_factor[symbol] = stock.ValuationRatios.PBRatio
86
87if len(market_cap) < self.market_cap_quantile:
88    return Universe.Unchanged
89
90# sort cap and size portfolios
91quantile: int = int(len(market_cap) / self.market_cap_quantile)
92sorted_by_cap: List[Symbol] = [x[0] for x in sorted(market_cap.items(), key=lambda item: item[1])]
93sorted_by_value: List[Symbol] = [x[0] for x in sorted(value_factor.items(), key=lambda item: item[1])]
94    
95# factors are ready
96if self.last_market_factor and self.last_size_factor and self.last_value_factor:
97    # create regression x from daily returns of
98    # value weight market factor of all selected chinese stocks,
99    # size factor and value factor
100    regression_x: List = [
101        self.DailyPerformanceValueWeight(self.last_market_factor),
102        self.FactorDailyPerformance(self.last_size_factor[0], self.last_size_factor[1]),
103        self.FactorDailyPerformance(self.last_value_factor[0], self.last_value_factor[1])
104    ]
105    
106    idiosyncratic_volatility: Dict[Symbol, float] = {}
107    
108    # calculate idiosyncratic volatility for each stock
109    for symbol in market_cap:
110        daily_returns: np.ndarray = self.data[symbol].daily_returns()
111        regression_model = self.MultipleLinearRegression(regression_x, daily_returns)
112        idiosyncratic_volatility[symbol] = np.std(regression_model.resid)
113        
114    if len(idiosyncratic_volatility) < self.quantile:
115        return Universe.Unchanged
116        
117    # selection based on idiosyncratic volatility
118    quantile: int = int(len(idiosyncratic_volatility) / self.quantile)
119    sorted_by_idio_vol: List[Symbol] = [x[0] for x in sorted(idiosyncratic_volatility.items(), key=lambda item: item[1])]
120    
121    # Long the bottom decile and short the top decile
122    self.long = sorted_by_idio_vol[:quantile]
123    self.short = sorted_by_idio_vol[-quantile:]
124# Store factor symbols for next month's calculations.
125self.last_market_factor = market_cap
126self.last_size_factor = (sorted_by_cap[:quantile], sorted_by_cap[-quantile:])
127self.last_value_factor = (sorted_by_value[:quantile], sorted_by_value[-quantile:])
128    
129return self.long + self.short
130def OnData(self, slice: Slice) -> None:
131if not self.selection_flag:
132    return
133self.selection_flag = False
134
135# trade execution
136targets:List[PortfolioTarget] = []
137for i, portfolio in enumerate([self.long, self.short]):
138    for symbol in portfolio:
139        if symbol in slice and slice[symbol]:
140            targets.append(PortfolioTarget(symbol, (((-1) ** i) / len(portfolio)) * self.traded_percentage))
141self.SetHoldings(targets, True)
142self.long.clear()
143self.short.clear()
144
145def Selection(self) -> None:
146self.selection_flag = True
147
148def DailyPerformanceValueWeight(self, market_cap: Dict[Symbol, float]) -> np.ndarray:
149# Create numpy array with zeros
150total_daily_returns: np.ndarray = np.zeros(self.period - 1)
151
152total_cap: float = sum([x[1] for x in market_cap.items()])
153for symbol, cap in market_cap.items():
154    # calculate weight for current stock
155    weight: float = cap / total_cap
156    # get daily returns of current stock
157    daily_returns: np.ndarray = self.data[symbol].daily_returns()
158    # multiply each daily return by weight
159    daily_returns: np.ndarray = daily_returns * weight
160    # add daily returns of current stock to total_daily_returns of portfolio
161    total_daily_returns += daily_returns
162
163return total_daily_returns
164
165def FactorDailyPerformance(self, long: List[Symbol], short: List[Symbol]) -> np.ndarray:
166# create numpy array with zeros
167total_daily_returns: np.ndarray = np.zeros(self.period - 1)
168
169# go through each long and short stock
170# Add daily returns of long stocks and sub daily returns of short stocks.
171for long_sym, short_sym in zip(long, short):
172    total_daily_returns += self.data[long_sym].daily_returns()
173    total_daily_returns -= self.data[short_sym].daily_returns()
174    
175return total_daily_returns
176
177def MultipleLinearRegression(self, x, y):
178x: np.ndarray = np.array(x).T
179x = sm.add_constant(x)
180result = sm.OLS(endog=y, exog=x).fit()
181return result
182class SymbolData():
183def __init__(self, period: int) -> None:
184self.daily_prices: RollingWindow = RollingWindow[float](period)
185
186def update(self, close: float) -> None:
187self.daily_prices.Add(close)
188
189def is_ready(self) -> bool:
190return self.daily_prices.IsReady
191
192def daily_returns(self) -> np.ndarray:
193daily_prices: np.ndarray = np.array([x for x in self.daily_prices])
194return (daily_prices[:-1] - daily_prices[1:]) / daily_prices[1:]
195
196# Custom fee model
197class CustomFeeModel(FeeModel):
198def GetOrderFee(self, parameters):
199fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
200return OrderFee(CashAmount(fee, "USD"))