“按季度再平衡的投资组合使用 Fama-French 回归计算特质方差，预测超额市场回报，并根据预期回报、市场方差和投资者风险厌恶程度确定股票权重。”
资产类别：差价合约（CFDs）、交易所交易基金（ETFs）、基金、期货 | 区域：全球 | 频率：季度 | 市场：股票 | 关键词：市场择时、特质、波动率
策略概述
市场方差以前一季度每日超额股票市场回报的平方值计算。S&P 500 股票的特质方差通过以 Fama 和 French 因子（市场、规模、市净率）为自变量的回归获得。残差按照市值加权以汇总特质方差。使用市场方差和特质方差预测下一季度的预期超额股票市场回报。根据预期超额回报、市场方差和投资者的风险厌恶系数，确定投资组合的股票权重。投资组合每季度重新平衡，以反映更新后的估计并优化风险回报匹配。
经济基础
学术界假设，高水平的特质波动率与市场参与者意见的高度分歧相关。因此，特质波动率与未来股票回报之间可能存在负相关关系。
论文来源
Market Timing with Aggregate and Idiosyncratic Stock Volatilities [点击浏览原文]
- Hui Guo 和 Jason Higbee
- 辛辛那提大学金融与房地产系；圣路易斯联邦储备银行研究部
<摘要>
Guo 和 Savickas [2005] 表明，整体股票市场波动率和平均特质股票波动率可以共同预测股票回报。在本文中，我们从投资组合管理者的角度量化其结果的经济意义。具体而言，我们评估了一位基于这些变量进行市场择时的均值-方差管理者的表现，例如 Sharpe 比率和 Jensen α。我们发现，在 1968-2004 年期间，与买入持有策略相比，相关的市场择时策略表现优异，这一差异在统计上和经济上都具有显著意义。
回测表现
| 年化收益率 | 20.07% |
| 波动率 | 37.8% |
| Beta | -0.125 |
| 夏普比率 | 0.43 |
| 索提诺比率 | -0.471 |
| 最大回撤 | N/A |
| 胜率 | 55% |
完整python代码
import numpy as np
from AlgorithmImports import *
import statsmodels.api as sm
import numpy as np
from typing import Dict, List, Tuple
from pandas.core.frame import dataframe
from pandas.core.series import Series
class MarketTimingwithAggregateandIdiosyncraticStockVolatilities(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2000, 1, 1)
self.SetCash(100000)
self.data:Dict[Symbol, SymbolData] = {}
self.weight:Dict[Symbol, float] = {}
self.period:int = 3*21
self.fundamental_count:int = 500
self.leverage:int = 5
self.quantile:int = 10
self.exchange_codes:List[str] = ['NYS', 'NAS', 'ASE']
self.last_size_long:List[Symbol] = []
self.last_size_short:List[Symbol] = []
self.last_book_long:List[Symbol] = []
self.last_book_short:List[Symbol] = []
self.market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
self.data[self.market] = SymbolData(self.period)
self.bonds:Symbol = self.AddEquity('SHY', Resolution.Daily).Symbol
self.regression_data:List[Tuple[float, float, float]] = []
self.regression_data_min_period:int = 12
self.selection_flag:bool = False
self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
self.UniverseSettings.Resolution = Resolution.Daily
self.AddUniverse(self.FundamentalSelectionFunction)
self.Schedule.On(self.DateRules.MonthEnd(self.market), self.TimeRules.AfterMarketOpen(self.market), 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 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)
if not self.selection_flag:
return Universe.Unchanged
selected: List[Fundamental] = sorted([x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and x.MarketCap != 0 \
and not np.isnan(x.ValuationRatios.PBRatio) and x.ValuationRatios.PBRatio != 0
and x.SecurityReference.ExchangeId in self.exchange_codes],
key=lambda x: x.DollarVolume, reverse=True)[:self.fundamental_count]
selected_dict: Dict[Symbol, Fundamental] = {x.Symbol : x for x in selected}
market_cap:Dict[Symbol, float] = {}
book_to_market:Dict[Symbol, float] = {}
# Warmup price rolling windows.
for symbol in list(selected_dict.keys()) + [self.market]:
if symbol in self.data:
continue
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] = selected_dict[symbol].MarketCap
book_to_market[symbol] = 1 / selected_dict[symbol].ValuationRatios.PBRatio # book-to-market
# Return, if after fundamental filtration we don't have enough stocks to sort them into deciles.
# In next month we will use size and book factors from t-3 month, where t is current month.
if len(market_cap) < self.quantile:
return Universe.Unchanged
quantile:int = int(len(market_cap) / self.quantile)
sorted_by_market_cap:List[Symbol] = [x[0] for x in sorted(market_cap.items(), key=lambda item: item[1])]
sorted_book_to_market:List[Symbol] = [x[0] for x in sorted(book_to_market.items(), key=lambda item: item[1])]
# Size Factor
current_size_long:List[Symbol] = sorted_by_market_cap[:quantile]
current_size_short:List[Symbol] = sorted_by_market_cap[-quantile:]
# Book to market Factor
current_book_long:List[Symbol] = sorted_book_to_market[:quantile]
current_book_short:List[Symbol] = sorted_book_to_market[-quantile:]
# Check if factors are ready
if (len(self.last_size_long) > 0 and len(self.last_size_short) > 0 and
len(self.last_book_long) > 0 and len(self.last_book_short) > 0):
# Check if market prices are ready
if self.data[self.market].is_ready():
# Calculate factors daily returns
book_factor_daily_returns:List[float] = self.CalculateFactorDailyReturns(self.last_book_long, self.last_book_short)
size_factor_daily_returns:List[float] = self.CalculateFactorDailyReturns(self.last_size_long, self.last_size_short)
market_returns:np.ndarray = self.data[self.market].daily_performances()
market_variance:np.ndarray = (np.std(market_returns) * np.sqrt(252)) ** 2
market_return:float = self.data[self.market].performance()
# stock residuals
idiosyncratic_variance:List[Tuple[float, float]] = []
for stock in selected:
symbol: Symbol = stock.Symbol
if symbol not in self.data:
continue
if not self.data[symbol].is_ready():
continue
Y:np.ndarray = self.data[symbol].daily_performances()
X:np.ndarray = [
market_returns,
size_factor_daily_returns,
book_factor_daily_returns
]
regression_model = self.MultipleLinearRegression(X, Y)
idiosyncratic_variance.append((regression_model.resid[-1], selected_dict[symbol].MarketCap))
# summary idiosyncratic variance calculation
summary_market_cap:float = sum([x[1] for x in idiosyncratic_variance])
summary_idiosyncratic_variance:float = sum([x[0] * (x[1] / summary_market_cap) for x in idiosyncratic_variance])
self.regression_data.append((market_return, market_variance, summary_idiosyncratic_variance))
self.last_size_long = current_size_long
self.last_size_short = current_size_short
self.last_book_long = current_book_long
self.last_book_short = current_book_short
return Universe.Unchanged
def OnData(self, data:Slice) -> None:
if not self.selection_flag:
return
self.selection_flag = False
if self.bonds not in data or not data[self.bonds]:
return
# regression data is ready
if len(self.regression_data) >= self.regression_data_min_period:
Y = [x[0] for x in self.regression_data][:-1]
X = [
[x[1] for x in self.regression_data][1:],
[x[2] for x in self.regression_data][1:]
]
# expected market return calculation
regression_model = self.MultipleLinearRegression(X, Y)
alpha:float = regression_model.params[0]
beta1:float = regression_model.params[1]
beta2:float = regression_model.params[2]
expected_market_return:float = alpha + self.regression_data[0][1]*beta1 + self.regression_data[0][2]*beta2
expected_market_variance:float = np.mean([x[1] for x in self.regression_data])
risk_aversion:int = 5
# equity and tbill weight
equity_weight:float = expected_market_return / (risk_aversion * expected_market_variance)
tbill_weight:float = 1 - equity_weight
# trade execution
if self.market in data and data[self.market] and self.bonds in data and data[self.bonds]:
self.SetHoldings(self.market, equity_weight)
self.SetHoldings(self.bonds, tbill_weight)
def CalculateFactorDailyReturns(self, factor_long:List[Symbol], factor_short:List[Symbol]) -> List[float]:
long_daily_returns:List[Symbol] = self.ListOfDailyReturns(factor_long, True)
short_daily_returns:List[Symbol] = self.ListOfDailyReturns(factor_short, False)
stocks_daily_returns:np.ndarray = np.array(long_daily_returns + short_daily_returns)
factor_daily_returns:List[float] = []
for i in range(len(stocks_daily_returns[0])):
factor_daily_returns.append(np.mean(stocks_daily_returns[:, i])) # Takes column of 2d array
return factor_daily_returns
def ListOfDailyReturns(self, symbols:List[Symbol], long_flag:bool) -> List[Symbol]:
symbol_list = []
for symbol in symbols:
# Those aren't symbols which were return from coarse
if symbol in self.data:
if self.data[symbol].is_ready():
if long_flag:
symbol_list.append(self.data[symbol].daily_performances())
else:
symbol_list.append(self.data[symbol].short_daily_performances())
return symbol_list
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
def Selection(self):
if self.Time.month % 3 == 0:
self.selection_flag = True
class SymbolData():
def __init__(self, period:int) -> None:
self.closes:RollingWindow[float] = RollingWindow[float](period)
self.period:int = period
def update(self, close:float) -> None:
self.closes.Add(close)
def is_ready(self) -> bool:
return self.closes.IsReady
def performance(self) -> float:
return self.closes[0] / self.closes[self.period - 1] - 1
def daily_performances(self) -> np.ndarray:
closes = np.array([x for x in self.closes])
return (closes[:-1] - closes[1:]) / closes[1:]
def short_daily_performances(self) -> np.ndarray:
closes = np.array([x for x in self.closes])
return ( -(closes[:-1] - closes[1:]) ) / closes[1:] # We change mark of daily performance for short stocks.
# Custom fee model
class CustomFeeModel(FeeModel):
def GetOrderFee(self, parameters):
fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
return OrderFee(CashAmount(fee, "USD"))