“通过季节性回报差异交易中国A股，做多最高十分位，做空最低十分位，使用价值加权、每月重新平衡的投资组合，排除CSMAR数据库中的微型股。”
资产类别: 股票 | 地区: 中国 | 周期: 每月 | 市场: 股票 | 关键词: 季节性、中国
I. 策略概要
投资范围包括在上海和深圳证券交易所上市的人民币计价A股，数据来自CSMAR数据库。微型股（市值最低的30%）被排除在外。季节性回报差异的计算方法是：同月平均回报减去其他月份平均回报，两者均基于横截面去均值的股票回报。历史平均值使用过去5到20年的数据计算。股票根据季节性回报差异分为十分位，做多最高十分位，做空最低十分位。投资组合按价值加权，每月重新平衡，利用股票回报的季节性模式。
II. 策略合理性
回报季节性在其他市场有充分的记录，表明个股回报因月份而异，并持续长达20年。同月高回报预示着正回报，而其他月份的回报则预示着负回报。这些异常现象是互补的，可以结合起来，两者都源于暂时的错误定价。作者还讨论了美国和中国市场之间潜在的差异。异常现象仍然稳健，因为套利限制不能解释回报，多头和空头回报相似。国有企业不影响策略功能，因为因子溢价存在于各种股票类型中。尽管股市改革降低了回报，但许多异常现象仍然显著。
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. 回测表现
| 年化回报 | 11.35% |
| 波动率 | 17.06% |
| β值 | 0.025 |
| 夏普比率 | 0.66 |
| 索提诺比率 | -0.578 |
| 最大回撤 | N/A |
| 胜率 | 59% |
V. 完整的 Python 代码
from AlgorithmImports import *
import numpy as np
from typing import List, Dict
from pandas.core.frame import dataframe
from pandas.core.series import Series
#endregion
class SeasonalDifferenceInChina(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2010, 1, 1)
self.SetCash(100_000)
self.period: int = 5 * 21 # storing n months of daily prices for each month
self.quantile: int = 10
self.leverage: int = 10
self.min_share_price: float = 1.
self.market_cap_quantile: int = 3
self.traded_percentage: float = .1
self.data: Dict[Symbol, SymbolData] = {}
self.weight: Dict[Symbol, float] = {}
market: Symbol = self.AddEquity("SPY", Resolution.Daily).Symbol
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]:
current_month: int = self.Time.month
# update the rolling window every day
for stock in fundamental:
symbol: Symbol = stock.Symbol
if symbol in self.data:
self.data[symbol].update(current_month, 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 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):]
seasonal_difference: Dict[Fundamental, float] = {}
# calculate seasonal return for this month
for stock in fundamental:
symbol: Symbol = stock.Symbol
# Get stock's closes
if symbol not in self.data:
self.data[symbol] = SymbolData(self.period)
# five years of daily prices
history: dataframe = self.History(symbol, self.period * 12, 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(time.month, close)
if self.data[symbol].is_ready():
# calculate average return for each month separately
months_averages: Dict[int, float] = self.data[symbol].months_averages()
# get seasonal momentum of the month
same_month_avg_return: Dict[int, float] = months_averages[current_month]
# sum averages of all other seasonal returns
other_months_avg_return: float = sum([avg_return for month_num, avg_return in months_averages.items() if month_num != current_month])
# calculate seasonal diffrence for current stock
seasonal_difference[stock] = same_month_avg_return - other_months_avg_return
if len(seasonal_difference) < self.quantile:
return Universe.Unchanged
# selection based on seasonal difference
quantile: int = int(len(seasonal_difference) / self.quantile)
sorted_by_seasonal_diff: List[Fundamental] = sorted(seasonal_difference, key=seasonal_difference.get)
# long the top decile and short the bottom decile
long: List[Fundamental] = sorted_by_seasonal_diff[-quantile:]
short: List[Fundamental] = sorted_by_seasonal_diff[:quantile]
# calculate weights
for i, portfolio in enumerate([long, short]):
mc_sum: float = sum(map(lambda x: x.MarketCap, portfolio))
for stock in portfolio:
self.weight[stock.Symbol] = (((-1) ** i) * stock.MarketCap / mc_sum) * self.traded_percentage
return list(self.weight.keys())
def OnData(self, slice: 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 slice.contains_key(symbol) and slice[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.daily_prices: Dict[int, RollingWindow] = {}
for i in range(1, 13):
# Storing daily prices under month number
self.daily_prices[i] = RollingWindow[float](period)
def update(self, month: int , close: float) -> None:
self.daily_prices[month].Add(close)
def is_ready(self) -> bool:
for _, rolling_window in self.daily_prices.items():
# Return False if atleast one rolling window in dictionary isn't ready
if not rolling_window.IsReady:
return False
# If all rolling windows are ready return True
return True
def months_averages(self) -> Dict[int, float]:
averages: Dict[int, float] = {}
for month_num, rolling_window in self.daily_prices.items():
month_performances: List[float] = []
closes: List[float] = list(rolling_window)
# calculate month performacnes
for i in range(0, len(closes), 21):
month_closes: List[float] = closes[i:i+21]
performance: float = (month_closes[0] - month_closes[-1]) / month_closes[-1]
month_performances.append(performance)
averages[month_num] = np.mean(month_performances)
return averages
# 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"))