“该策略投资于24个主要股指,计算广义风险调整动量。每月选择最佳参数,根据四分位排名确定头寸,并按目标波动率进行缩放以进行比较。”

I. 策略概要

该策略投资于来自发达市场和新兴市场的24个主要股指。对于每个指数,计算已实现波动率和动量。广义风险调整动量通过将动量除以已实现波动率的N次方来计算。每月通过最大化夏普比率来选择最佳N(范围从0到4),以平衡回报和波动率。指数分为四分位数,做多表现最佳的四分之一指数,做空表现最差的四分之一指数。为了与Barroso和Santa-Clara的时变波动率策略进行比较,该策略通过目标波动率进行缩放,并根据过去六个月的已实现波动率进行调整。

II. 策略合理性

该论文认为,动量策略中的高度不确定性是由个别资产的已实现波动率驱动的。高波动性资产更有可能被纳入动量投资组合,但此类投资组合会经历更高的超额波动性。动量策略在波动性较小的资产中被发现更有利可图,低波动性资产表现出更强的动量。专注于高波动性股票的传统动量策略在市场高度不确定时期往往表现不佳。该论文提出优化已实现波动率的幂次(N),这可以根据市场条件调整策略的激进程度,从而改善投资组合特征。将这种方法与Barroso和Santa-Clara的恒定波动率缩放进行比较,广义动量策略表现更优,产生了统计上显著的阿尔法。

III. 来源论文

Momentum and the Cross-Section of Stock Volatility [点击查看论文]

<摘要>

最近的文献表明,动量策略在某些时期表现出显著的下行风险,称为“动量崩溃”。我们发现动量策略回报的高度不确定性来源于个股的横截面波动率。在形成期内具有高已实现波动率的股票往往会失去动量效应。我们提出了一种新方法,即广义风险调整动量(GRJMOM),以减轻高动量特定风险的负面影响。GRJMOM 被证明在包括英国股票、商品、全球股指和固定收益市场在内的多个资产类别中,比现有动量排名方法更有利可图且风险更低。

IV. 回测表现

年化回报12.7%
波动率19.9%
β值-0.348
夏普比率0.64
索提诺比率-0.1
最大回撤-53.6%
胜率48%

V. 完整的 Python 代码

from collections import deque
from AlgorithmImports import *
from math import sqrt
import operator
import numpy as np
class GeneralisedRiskAdjustedMomentuminEquityIndexes(QCAlgorithm):
    def Initialize(self):
        self.SetStartDate(2008, 4, 1)   # ACWI data starts in 2008.
        self.SetCash(100000)
        
        self.symbols = [
                        "EWA",  # iShares MSCI Australia Index ETF
                        "EWO",  # iShares MSCI Austria Investable Mkt Index ETF
                        "EWK",  # iShares MSCI Belgium Investable Market Index ETF
                        "EWZ",  # iShares MSCI Brazil Index ETF
                        "EWC",  # iShares MSCI Canada Index ETF
                        "FXI",  # iShares China Large-Cap ETF
                        "EWQ",  # iShares MSCI France Index ETF
                        "EWG",  # iShares MSCI Germany ETF 
                        "EWH",  # iShares MSCI Hong Kong Index ETF
                        "EWI",  # iShares MSCI Italy Index ETF
                        "EWJ",  # iShares MSCI Japan Index ETF
                        "EWM",  # iShares MSCI Malaysia Index ETF
                        "EWW",  # iShares MSCI Mexico Inv. Mt. Idx
                        "EWN",  # iShares MSCI Netherlands Index ETF
                        "EWS",  # iShares MSCI Singapore Index ETF
                        "EZA",  # iShares MSCI South Africe Index ETF
                        "EWY",  # iShares MSCI South Korea ETF
                        "EWP",  # iShares MSCI Spain Index ETF
                        "EWD",  # iShares MSCI Sweden Index ETF
                        "EWL",  # iShares MSCI Switzerland Index ETF
                        "EWT",  # iShares MSCI Taiwan Index ETF
                        "THD",  # iShares MSCI Thailand Index ETF
                        "EWU",  # iShares MSCI United Kingdom Index ETF
                        "SPY",  # SPDR S&P 500 ETF
                        ]
        # Daily price data.
        self.data = {}
        self.index = self.AddEquity('ACWI', Resolution.Daily).Symbol
        self.data[self.index] = deque(maxlen = 21)
        self.leverage_cap = 3
        # Volatility factor.
        self.volatility_factor_symbols = []                  # Symbols
        self.volatility_factor_vector = deque(maxlen = 6)    # Monthly volatility.
        
        # Minimal data count needed for optimalization.
        self.period = 12 * 21
        self.quantile = 4
        
        for symbol in self.symbols:
            data = self.AddEquity(symbol, Resolution.Daily)
            data.SetLeverage(50)
            
            self.data[symbol] = deque()
        self.Schedule.On(self.DateRules.MonthStart(self.symbols[0]), self.TimeRules.AfterMarketOpen(self.symbols[0]), self.Rebalance)
        
    def OnData(self, data):
        # Store daily data.
        for symbol in self.data:
            if symbol in data and data[symbol]:
                price = data[symbol].Value
                self.data[symbol].append(price)
                        
    def Rebalance(self):
        # Sharpe ratio indexed by n.
        sharpe_ratio_data = {}
        
        n = 0
        while n <= 4:
            performance_volatility = {}
            generalized_momentum = {}
            
            for symbol in self.symbols:
                if self.Securities[symbol].GetLastData() and (self.Time.date() - self.Securities[symbol].GetLastData().Time.date()).days < 5:
                    # At least year of data is ready.
                    if len(self.data[symbol]) >= self.period:
                        # Realized volatility calc.
                        perf = Return(self.data[symbol])
                        vol = Volatility(self.data[symbol])
                        performance_volatility[symbol] = (perf, vol)
                        
                        daily_prices = np.array([x for x in self.data[symbol]])
                        daily_returns = (daily_prices[1:] - daily_prices[:-1]) / daily_prices[:-1]
                        realized_volatility = sqrt(sum(daily_returns**2) / self.period)
                        
                        generalized_momentum[symbol] = perf / (realized_volatility ** n)
                    
            if len(generalized_momentum) >= self.quantile:
                sorted_by_momentum = sorted(generalized_momentum.items(), key = lambda x: x[1], reverse = True)
                quantile = int(len(sorted_by_momentum) / self.quantile)
                long = [x[0] for x in sorted_by_momentum[:quantile]]
                short = [x[0] for x in sorted_by_momentum[-quantile:]]
                
                # Sharpe ratio calc.
                symbol_count = len(long + short)
                total_performance = sum([performance_volatility[x][0] / symbol_count for x in long])
                total_performance += sum([((-1) * performance_volatility[x][0]) / symbol_count for x in short])
                
                total_volatility = sum([performance_volatility[x][1] / symbol_count for x in long + short])
                portfolio_sharpe_ratio = total_performance / total_volatility
                
                portfolio_symbols = [long, short]
                sharpe_ratio_data[str(n)] = (portfolio_sharpe_ratio, portfolio_symbols)
                
            n += 0.1
        
        if len(sharpe_ratio_data) != 0:
            max_by_sharpe_ratio = max(sharpe_ratio_data.items(), key = operator.itemgetter(1))
            long = max_by_sharpe_ratio[1][1][0]
            short = max_by_sharpe_ratio[1][1][1]
            # Calculate last month's volatility.
            if len(self.volatility_factor_symbols) != 0:
                monthly_volatility = self.CalculateFactorVolatility(self.data, self.volatility_factor_symbols)
                if monthly_volatility != 0:
                    self.volatility_factor_vector.append(monthly_volatility)
            # Store new factor symbols.
            self.volatility_factor_symbols = long + short
            
            # Volatility factor is ready.
            if len(self.volatility_factor_vector) == self.volatility_factor_vector.maxlen:
                # Volatility targetting.
                if len(self.data[self.index]) == self.data[self.index].maxlen:
                    annual_index_volatility = Volatility(self.data[self.index]) * sqrt(12*21)
                    realized_strategy_volatility = np.mean(self.volatility_factor_vector)
                    # Cap leverage if needed.
                    target_leverage = min(self.leverage_cap, (annual_index_volatility / realized_strategy_volatility))
                    
                    # Trade execution.
                    symbols_invested = [x.Key.Value for x in self.Portfolio if x.Value.Invested]
                    for symbol in symbols_invested:
                        if symbol not in long + short:
                            self.Liquidate(symbol)
                    
                    count = len(long + short)
                    for symbol in long:
                        self.SetHoldings(symbol, target_leverage / len(long))
                    for symbol in short:
                        self.SetHoldings(symbol, -target_leverage / len(short))
    def CalculateFactorVolatility(self, data, factor_symbols):
        monthly_volatility = 0
        if len(factor_symbols) != 0:
            for symbol in factor_symbols:
                if symbol in data and len(data[symbol]) >= 21:
                    monthly_volatility += (Volatility([x for x in data[symbol]][-21:]) / len(factor_symbols))
        return monthly_volatility
def Return(values):
    return (values[-1] - values[0]) / values[0]
    
def Volatility(values):
    values = np.array(values)
    returns = (values[1:] - values[:-1]) / values[:-1]
    return np.std(returns)  

发表评论

了解 Quant Buffet 的更多信息

立即订阅以继续阅读并访问完整档案。

继续阅读