“该策略投资于法国的十大工业部门，使用动量（EMAR）优化子投资组合和回溯期，每月重新平衡。仅当动量超过现金时才建仓，否则投资于现金。”
资产类别: ETF | 地区: 美国 | 周期: 每月 | 市场: 股票 | 关键词: 动量
I. 策略概要
该策略投资于法国的十大等权重工业部门，使用行业ETF作为代理。每个子投资组合的动量通过月度回报的滞后指数平均值（EMAR）来衡量。该策略考虑长达18个月的回溯期，从而产生180种组合。在每个月末，EMAR用于优化子投资组合的数量和回溯期，以实现最佳夏普比率。如果子投资组合的动量不超过现金的动量，则不建仓，投资组合投资于现金。该策略每月重新平衡。
II. 策略合理性
该策略的功能植根于行为偏差，类似于经典的动量策略。优化回溯期可以改善结果，比固定回溯期表现更有效。积极管理基金组合或子投资组合有助于避免崩盘，尽管卖空低动量证券偶尔会失败。更好的解决方案是当动量未能超越现金回报时退出多头头寸，确保与基准相比更平稳的表现。这种方法本可以避免2007-2008年金融危机期间的重大损失，展示了其在动荡市场条件下的弹性。该策略的适应性和对动量驱动决策的关注导致更稳定的一致回报和更低的风险。
III. 来源论文
Fund and Subportfolio Momentum [点击查看论文]
- 迈克尔·C·奥康纳（Michael C. O’Connor），MO’C投资组合分析公司（MO’C Portfolio Analytics）。
<摘要>
简单持有高动量、高滞后回报股票的策略，因其获得了通常持怀疑态度的学者的支持而令人惊叹。但实践中也存在一些问题。未经对冲的纯动量策略对2007-2008年雷曼兄弟/次贷危机毫无帮助，在1929年也无济于事。本文将动量应用于股票基金组合或类似股票基金的子投资组合，而非股票。并且针对恐慌期间动量失效的情况，提出并测试了一种简单的“转向现金”疗法，使用一种新型“阿尔法”，尽管由此产生的投资组合回报具有混合分布特征，但其置信区间仍可计算。进行了前向回溯程序，这更像是一种模拟而非抽象的数理统计练习，它提供了一种动态优化的动量策略，能够适应市场的长期变化。结果表明，针对基金组合或子投资组合的动量度量的最佳形式与适用于股票的流行形式截然不同。并且发现，如果强调波动性抑制，并且在动量减弱时有转向现金的政策，那么持有的最佳目标基金或子投资组合数量是候选数量的很大一部分。
IV. 回测表现
| 年化回报 | 19.8% |
| 波动率 | 23.25% |
| β值 | 0.665 |
| 夏普比率 | 0.85 |
| 索提诺比率 | 0.338 |
| 最大回撤 | N/A |
| 胜率 | 77% |
V. 完整的 Python 代码
from AlgorithmImports import *
from itertools import combinations
from pandas.core.frame import dataframe
from dateutil.relativedelta import relativedelta
from scipy.optimize import minimize
import sys
# endregion
class OptimalizedSubportfolioMomentum(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2010, 1, 1)
self.SetCash(100_000)
industry_tickers: List[str] = [
'XLB', 'XLY', 'XLF', 'XLP', 'XLV', 'XLU', 'XLC', 'XLE', 'XLI', 'XLK'
]
self.universe_assets: List[Symbol] = [
self.AddEquity(ticker, Resolution.Daily).Symbol for ticker in industry_tickers
]
self.cash: Symbol = self.AddEquity('BIL', Resolution.Daily).Symbol
self.max_lookback_period: int = 18
self.period: int = 60
self.percentage_traded: float = .9
self.universe_returns: Dict[Symbol, RollingWindow] = {
symbol : RollingWindow[float](self.period) for symbol in self.universe_assets
}
self.recent_month: int = -1
self.settings.daily_precise_end_time = False
def OnData(self, slice: Slice) -> None:
# rebalance once a month
if self.recent_month == self.Time.month:
return
self.recent_month = self.Time.month
# price history
etf_history: dataframe = self.History(self.universe_assets + [self.cash], start=self.Time.date() - relativedelta(months=self.period + self.max_lookback_period), end=self.Time.date())['close'].unstack(level=0)
etf_history = etf_history.groupby(pd.Grouper(freq='MS')).last()
etf_history = etf_history.pct_change()
etf_history = etf_history.iloc[1:]
# data is ready
if any(np.isnan(x) for x in etf_history.iloc[0]) or len(etf_history) < self.period + self.max_lookback_period:
return
# EMAR calculation
ema_df: dataframe = pd.dataframe(index=etf_history.index)
for symbol in list(etf_history.columns):
for L in range(1, self.max_lookback_period + 1):
ema_df[f'{symbol}_{L}'] = etf_history[[symbol]].ewm(span=L, adjust=False).mean()
# remove first ema_df nan and adjust etf history as well
ema_df = ema_df.iloc[1:]
etf_history = etf_history.iloc[1:]
cash_cols: List[str] = [x for x in ema_df.columns if self.cash.Value in x]
etf_cols: List[str] = [x for x in ema_df.columns if x not in cash_cols]
etf_count: int = len(etf_cols) // self.max_lookback_period
signal: dataframe = pd.dataframe(index=ema_df.index)
for etf_range in range(0, etf_count):
etf_cols_from_range:List[str] = etf_cols[etf_range*self.max_lookback_period : etf_range*self.max_lookback_period + self.max_lookback_period]
# 0 -> cash, 1 -> sub-portfolio
signal[etf_cols_from_range] = np.where(ema_df[etf_cols_from_range].values > ema_df[cash_cols].values, 1, 0)
# all the possible combination of sub-portfolios
best_comb: Series = pd.Series()
best_comb_sharpe: float = sys.float_info.min
for period in range(1, self.max_lookback_period + 1):
relevant_portfolios: List[str] = [x for x in etf_cols if x[-len(f'_{period}'):] == f'_{period}']
for cnt in range(1, len(relevant_portfolios) + 1):
for combination in list(combinations(relevant_portfolios, cnt)):
comb_etfs: List[str] = [x.split('_')[0] for x in combination]
comb_arr: List[str] = list(combination)
# replace 0 with cash
comb_df: dataframe = etf_history[comb_etfs].values * signal[comb_arr].shift(1)
for col in comb_arr:
comb_df[col] = np.where(comb_df[col] == 0, etf_history[self.cash], comb_df[col])
# calculate sharpe ratio
comb_df_perf: dataframe = comb_df.mean(axis=1)
comb_df_perf = comb_df_perf.iloc[1:]
sr:float = comb_df_perf.mean() / comb_df_perf.std() * (12 ** 0.5)
if sr > best_comb_sharpe:
best_comb_sharpe = sr
best_comb = signal[comb_arr].iloc[-1]
# traded weight calculation
traded_weight:Dict[Symbol, float] = {}
for item in best_comb.index:
etf: Symbol = self.Symbol(item.split('_')[0])
signal: int = best_comb[item]
if signal == 1:
traded_weight[etf] = self.percentage_traded / best_comb.shape[0]
else:
if self.cash not in traded_weight:
traded_weight[self.cash] = 0
traded_weight[self.cash] += self.percentage_traded / best_comb.shape[0]
# trade execution
invested = [x.Key for x in self.Portfolio if x.Value.Invested]
for symbol in invested:
if symbol not in traded_weight:
self.Liquidate(symbol)
for symbol, w in traded_weight.items():
self.SetHoldings(symbol, w)