“该策略根据12个月动量和净派息收益率选择前100只低波动性美国股票，每季度重新平衡并等权重，旨在利用强大的股东收益率和动量。”
资产类别: 股票 | 地区: 美国 | 周期: 季度 | 市场: 股票 | 关键词: 保守
I. 策略概要
投资范围包括美国市值最大的1,000只股票。首先根据36个月的股票回报波动率将股票分为两组。其中500只波动率最低的股票，再根据12-1个月的价格动量和净派息收益率（NPY，包括股息收益率和流通股变化）进行排名。动量和NPY排名取平均值，选择排名前100的股票进行多头配置。该策略每季度重新平衡并等权重，旨在利用具有强大股东收益率和动量的低波动性股票。
II. 策略合理性
保守公式结合了动量、波动性和净派息收益率，为投资者提供了对主要因子溢价的有效敞口。它依赖于简单的价格和股息数据，使其自1929年以来具有稳健性和可回溯性。这种简单性降低了“p值操纵”和因子捕捞的风险。该策略应用于美国最大的1,000只股票，通过每季度重新平衡，确保了经济相关性，并减少了周转率和交易成本。该公式已经过高级资产定价模型的测试，在数据输入较少的情况下表现相似或更好。其回报在不同时间和国际市场中保持一致，证明了其稳健性和适应性。
III. 来源论文
The Conservative Formula: Quantitative Investing Made Easy [点击查看论文]
- 范·弗利特（Pim van Vliet）、布利茨（David Blitz），罗贝科量化投资公司（Robeco Quantitative Investments）
<摘要>
我们提出了一种保守的投资公式，该公式根据三个标准选择100只股票：低回报波动率、高净派息收益率和强劲的价格动量。我们表明，这个简单的公式为投资者提供了对最重要的因子溢价的充分且有效的敞口，从而有效地将半个世纪的实证资产定价研究总结为一种易于实施的投资策略。自1929年以来，保守公式的复合年回报率为15.1%，大幅跑赢市场。它降低了下行风险，并在每个十年都显示出正回报。该公式在欧洲、日本和新兴股票市场也表现强劲，并击败了基于规模、价值、质量和动量组合的各种其他策略。该公式旨在成为广泛投资者实际有用的工具，并通过使用三个简单标准（甚至不需要会计数据）来解决学术界对“p值操纵”的担忧。
IV. 回测表现
| 年化回报 | 15.1% |
| 波动率 | 16.5% |
| β值 | 0.675 |
| 夏普比率 | 0.67 |
| 索提诺比率 | 0.459 |
| 最大回撤 | N/A |
| 胜率 | 78% |
V. 完整的 Python 代码
import numpy as np
from AlgorithmImports import *
from typing import List, Dict
from numpy import isnan
class TheConservativeFormula(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2000, 1, 1)
self.SetCash(100000)
self.period:int = 36 * 21
self.leverage:int = 10
self.quantile:int = 5
self.long:List[Symbol] = []
self.data:Dict[Symbol, SymbolData] = {}
market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
self.fundamental_count:int = 500
self.fundamental_sorting_key = lambda x: x.DollarVolume
self.selection_flag:bool = False
self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
self.UniverseSettings.Resolution = Resolution.Daily
self.AddUniverse(self.FundamentalSelectionFunction)
self.Schedule.On(self.DateRules.MonthStart(market), self.TimeRules.AfterMarketOpen(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] = [x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and \
not isnan(x.ValuationRatios.TotalYield) and (x.ValuationRatios.TotalYield > 0)]
# Warmup price rolling windows.
for stock in selected:
symbol:Symbol = stock.Symbol
if symbol in self.data:
continue
self.data[symbol] = SymbolData(self.period)
history = self.History(symbol, self.period, Resolution.Daily)
if history.empty:
self.Log(f"Not enough data for {symbol} yet.")
continue
closes = history.loc[symbol].close
for time, close in closes.items():
self.data[symbol].update(close)
yield_performance_volatility:Dict[Symbol, (Tuple[float])] = { x.Symbol : (x.ValuationRatios.TotalYield , self.data[x.Symbol].performance(), self.data[x.Symbol].volatility()) for x in selected if self.data[x.Symbol].is_ready()}
# Volatility sorting.
if len(yield_performance_volatility) < 2:
return Universe.Unchanged
sorted_by_ret_vol:List[Tuple[Symbol, float]] = sorted(yield_performance_volatility.items(), key = lambda x: x[1][2], reverse = True)
half:int = int(len(sorted_by_ret_vol) / 2)
low_by_ret_vol:List[Tuple[Symbol, float]] = [x for x in sorted_by_ret_vol[-half:]]
# Scoring
rank:Dict[Symbol, int] = {}
for symbol, _ in low_by_ret_vol:
rank[symbol] = 0
sorted_by_mom:List[Tuple[Symbol, float]] = sorted(low_by_ret_vol, key = lambda x: x[1][1], reverse = True)
score:int = len(sorted_by_mom)
for symbol, _ in sorted_by_mom:
rank[symbol] += score
score -= 1
sorted_by_yield:List[Tuple[Symbol, float]] = sorted(low_by_ret_vol, key = lambda x: x[1][0], reverse = True)
score:int = len(sorted_by_yield)
for symbol, _ in sorted_by_yield:
rank[symbol] += score
score -= 1
if len(rank) >= self.quantile:
sorted_by_rank:List[Tuple[Symbol, float]] = sorted(rank.items(), key = lambda x: x[1], reverse = True)
quintile:int = int(len(sorted_by_rank) / self.quantile)
self.long:List[Symbol] = [x[0] for x in sorted_by_rank[:quintile]]
return self.long
def OnData(self, data: Slice) -> None:
if not self.selection_flag:
return
self.selection_flag = False
# Trade execution
portfolio:List[PortfolioTarget] = [PortfolioTarget(symbol, 1 / len(self.long)) for symbol in self.long if symbol in data and data[symbol]]
self.SetHoldings(portfolio, True)
self.long.clear()
def Selection(self) -> None:
if self.Time.month % 3 == 0:
self.selection_flag = True
class SymbolData():
def __init__(self, period:int):
self._price:RollingWindow = RollingWindow[float](period)
def update(self, price:float) -> None:
self._price.Add(price)
def is_ready(self) -> bool:
return self._price.IsReady
def performance(self, values_to_skip = 0) -> float:
closes:List[float] = [x for x in self._price][:12*21][values_to_skip:]
return (closes[0] / closes[-1] - 1)
def volatility(self) -> float:
closes:np.ndarray = np.array([x for x in self._price])
returns:np.ndarray = (closes[:-1] / closes[1:]) - 1
return np.std(returns)
# Custom fee model.
class CustomFeeModel(FeeModel):
def GetOrderFee(self, parameters):
fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
return OrderFee(CashAmount(fee, "USD"))