“该策略根据特质性非对称性（IE）对美国股票进行排序，对低IE的股票进行多头交易，对高IE的股票进行空头交易。投资组合按价值加权配置，并每月进行再平衡。”
资产类别：股票 | 地区：美国 | 频率：每月 | 市场：股票 | 关键词：特质、非对称性
I. 策略概述
该策略以CRSP数据库（NYSE、AMEX、NASDAQ）中股价高于5美元的美国股票为目标。利用过去12个月的数据，将每只股票的超额收益回归于市场超额收益和市场超额收益的平方。特质性非对称性（IE）定义为大幅收益和大幅亏损的累计概率差异，这些概率以均值为中心、标准差为界限计算得出。根据IE值将股票分为十分位，对低IE（风险更高）的股票做多，对高IE（风险较低）的股票做空。投资组合按价值加权配置，并每月进行再平衡，以捕捉非对称性趋势。
II. 策略合理性
该策略利用投资者的行为偏差，即投资者更倾向于选择具有更高极端收益可能性的股票（正IE），而避免可能产生严重亏损的股票。这种偏好导致高IE股票的价格被高估，低IE股票的价格被低估，从而使高IE股票的预期回报更低。与偏度（skewness）不同，新的非对称性测度捕捉了更高阶的非对称性，从而更有效地解释股票的横截面回报。研究表明，即使偏度为零，高上行非对称性仍与低回报强相关。基于十分位的分析验证了偏度本身无法可靠预测回报，而新的非对称性测度提供了关键的股票表现洞察。
III. 论文来源
Stock Return Asymmetry: Beyond Skewness [点击浏览原文]
- 作者：雷江、可武、郭富周、朱一峰
- 机构：清华大学、中国人民大学、圣路易斯华盛顿大学、埃默里大学、中央财经大学
<摘要>
本文提出了两种用于股票回报的非对称性测度。与传统的偏度测度不同，我们的测度基于数据的分布函数，而不仅仅是第三中心矩。通过实证研究表明，上行非对称性越大，回报越低，且这种关系在偏度为零时依然显著。我们的研究为股票表现提供了新的视角，同时验证了更高阶非对称性的重要性。
IV. 回测表现
| 年化收益率 | 2.25% |
| 波动率 | 6.03% |
| Beta | 0.005 |
| 夏普比率 | 0.37 |
| 索提诺比率 | -0.149 |
| 最大回撤 | N/A |
| 胜率 | 53% |
V. 完整python代码
import statsmodels.api as sm
from AlgorithmImports import *
class IdiosyncraticAsymmetryInUSStocks(QCAlgorithm):
def Initialize(self) -> None:
self.SetStartDate(2000, 1, 1)
self.SetCash(100000)
self.data:Dict[Symbol, SymbolData] = {}
self.weight:Dict[Symbol, float] = {}
self.regression_period:int = 12 * 21 + 1 # need n daily prices
self.selection_size:int = 10 # 10 = decile selection, 5 = quintile selection, ...
self.leverage:int = 5
self.min_share_price:float = 5.
self.exchange_codes:List[str] = ['NYS', 'NAS', 'ASE']
self.symbol:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
# warm up SPY prices
self.PerformHistory(self.symbol)
self.fundamental_count:int = 3000
self.fundamental_sorting_key = lambda x: x.MarketCap
self.selection_flag:bool = False
self.UniverseSettings.Resolution = Resolution.Daily
self.AddUniverse(self.FundamentalSelectionFunction)
self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
self.Schedule.On(self.DateRules.MonthStart(self.symbol), self.TimeRules.BeforeMarketClose(self.symbol, 0), 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 daily closes
for stock in fundamental:
symbol = stock.Symbol
if symbol in self.data:
# update daily closes
self.data[symbol].update_closes(stock.AdjustedPrice)
# monthly rebalance
if not self.selection_flag:
return Universe.Unchanged
selected:List[Fundamental] = [
x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and x.Price > self.min_share_price and \
x.SecurityReference.ExchangeId in self.exchange_codes and x.MarketCap != 0 and x.Symbol != self.symbol and x.CompanyReference.BusinessCountryID == 'USA' and \
not np.isnan(x.FinancialStatements.IncomeStatement.ResearchAndDevelopment.TwelveMonths) and x.FinancialStatements.IncomeStatement.ResearchAndDevelopment.TwelveMonths != 0
]
if len(selected) > self.fundamental_count:
selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
# can't perform selection, when market closes aren't ready
if not self.data[self.symbol].are_closes_ready():
return Universe.Unchanged
regression_x:List = [
self.data[self.symbol].daily_returns(),
self.data[self.symbol].daily_returns_squared()
]
IE:Dict[Symbol, float] = {} # storing stocks IE values keyed by stocks symbols
for stock in selected:
symbol:Symbol = stock.Symbol
# warm up stock prices
if symbol not in self.data:
self.PerformHistory(symbol)
# check if closes data are ready
if not self.data[symbol].are_closes_ready():
continue
# perform regression
regression_y = self.data[symbol].daily_returns()
regression_model = self.MultipleLinearRegression(regression_x, regression_y)
# retrieve all residuals from regression
daily_residuals = regression_model.resid
# calcualte IE value from stock daily residuals
IE_value:float = self.data[symbol].calculate_IE(daily_residuals)
# store stock's IE value keyed by stock's symbol
IE[stock] = IE_value
# make sure, there are enough stocks for selection
if len(IE) < (self.selection_size * 2):
return Universe.Unchanged
# perform selection
selection_num:int = int(len(IE) / self.selection_size)
sorted_by_IE:List[Fundamental] = [x[0] for x in sorted(IE.items(), key=lambda item: item[1])]
# long stocks with low IE values
long:List[Fundamental] = sorted_by_IE[:selection_num]
# short stocks with high IE values
short:List[Fundamental] = sorted_by_IE[-selection_num:]
# 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
return list(self.weight.keys())
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 PerformHistory(self, symbol):
''' warm up stock prices from History object based on symbol parameter '''
self.data[symbol] = SymbolData(self.regression_period)
history = self.History(symbol, self.regression_period, Resolution.Daily)
# make sure history isn't empty
if history.empty:
return
closes = history.loc[symbol].close
for time, close in closes.items():
self.data[symbol].update_closes(close)
def MultipleLinearRegression(self, x, y):
''' perform multiple regression and return regression model '''
x = np.array(x).T
x = sm.add_constant(x)
result = sm.OLS(endog=y, exog=x).fit()
return result
def Selection(self) -> None:
self.selection_flag = True
class SymbolData():
def __init__(self, regression_period: int):
self._closes:RollingWindow = RollingWindow[float](regression_period)
def update_closes(self, close: float) -> None:
self._closes.Add(close)
def are_closes_ready(self) -> bool:
return self._closes.IsReady
def daily_returns(self) -> np.ndarray:
# calculate daily returns for period t-6 to t-1 months
closes:np.ndarray = np.array([x for x in self._closes])[21:]
daily_returns:np.ndarray = (closes[:-1] - closes[1:]) / closes[1:]
return daily_returns
def daily_returns_squared(self) -> List[float]:
# calculate daily returns for period t-6 to t-1 months
closes:np.ndarray = np.array([x for x in self._closes])[21:]
daily_returns:np.ndarray = (closes[:-1] - closes[1:]) / closes[1:]
daily_returns_squared:List[float] = [x*x for x in daily_returns]
return daily_returns_squared
def calculate_IE(self, residuals) -> int:
average_residuals:float = np.average(residuals)
two_residuals_std:float = 2 * np.std(residuals)
avg_plus_two_std:float = average_residuals + two_residuals_std
avg_minus_two_std:float = average_residuals - two_residuals_std
over_avg_plus_two_std:int = 0 # counting number of residuals, which were over avg_plus_two_std
under_avg_minus_two_std:int = 0 # counting number of residuals, which were under avg_minus_two_std
for residual in residuals:
if residual > avg_plus_two_std:
over_avg_plus_two_std += 1
elif residual < avg_minus_two_std:
under_avg_minus_two_std += 1
IE_value:int = over_avg_plus_two_std - under_avg_minus_two_std
return IE_value
# Custom fee model
class CustomFeeModel(FeeModel):
def GetOrderFee(self, parameters):
fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
return OrderFee(CashAmount(fee, "USD"))
