Quant Buffet放轻松,别过度思虑

危机阿尔法投资组合

登录后收藏

学术论文

策略概要

该策略涉及五只期限从1年到20年不等的美国国债ETF。绝对动量是在九个月内评估的,不包括上个月的回报,使用1年期国债回报来计算超额回报。具有正动量的资产被纳入投资组合;否则,全部权重分配给1年期国债。相对动量侧重于基于风险调整后回报(超额回报除以方差)的资产排名。权重与排名成正比,投资组合每月重新平衡,确保由绝对和相对动量信号驱动的动态分配。

II. 策略合理性

由于信用风险低和流动性高,美国国债在市场危机期间被视为避险资产,即使股票表现不佳,也会推高其价格。这种避险地位在整个市场周期中持续存在,无论利率趋势如何。通过利用已被证明对政府债券有效的相对动量,并在市场风险上升时战术性地增持国债作为危机阿尔法生成器,投资者可以在金融动荡期间提高回报。该策略利用国债作为避险资产的持续表现,与市场动态保持一致,以减轻风险并改善不确定时期的投资组合结果。

回测表现

波动率5.04%
夏普比率0.32
索提诺比率-0.459
最大回撤-5.9%
胜率51%

完整 Python 代码

import numpy as np
from AlgorithmImports import *
class TheCrisisAlphaPortfolio(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2008, 1, 1)
self.SetCash(100000)
self.symbols = ['TLT', 'IEF', 'IEI', 'SHY']
self.treasuries_1year = 'BIL'

period = 10
self.SetWarmUp(period)

# Monthly etf price.
self.data = {}

for symbol in self.symbols + [self.treasuries_1year]:
    data = self.AddEquity(symbol, Resolution.Daily)
    self.data[symbol] = SymbolData(symbol, period)

self.last_month = -1
self.Schedule.On(self.DateRules.MonthStart(self.symbols[0]), self.TimeRules.At(0, 0), self.Rebalance)

def OnData(self, data):
if self.last_month == self.Time.month: return
self.last_month = self.Time.month
# Store monthly prices.
for symbol in self.symbols + [self.treasuries_1year]:
    symbol_obj = self.Symbol(symbol)
    if symbol_obj in data.Keys:
        if data[symbol_obj]:
            price = data[symbol_obj].Value
            if price != 0:
                self.data[symbol].update(price)

def Rebalance(self):
self.Liquidate()

# Etfs with positive momentum. - symbol -> (performance, volatility)
positive_mom = {x : (self.data[x].performance(), self.data[x].volatility()) for x in self.symbols if self.data[x].is_ready() and self.data[x].performance() > 0}

if not self.data[self.treasuries_1year].is_ready(): return
if len(positive_mom) > 0:
    bil_ret = self.data[self.treasuries_1year].performance()
    
    rank = {}
    for symbol, perf_volatility in positive_mom.items():
        excess_ret = perf_volatility[0] - bil_ret
        variance = perf_volatility[1] ** 2
        rank[symbol] = excess_ret / variance
    
    sorted_by_rank = [x[0] for x in sorted(rank.items(), key = lambda x: x[1], reverse = True)]
    
    rank_len = len(sorted_by_rank)
    total_score = sum(range(0, rank_len + 1))
    
    rank_index = rank_len
    for symbol in sorted_by_rank:
        weight = (1 / total_score) * rank_index
        self.SetHoldings(symbol, weight)
        rank_index -= 1
else:
    self.SetHoldings(self.treasuries_1year, 1)
    
class SymbolData():
def __init__(self, symbol, period):
self.Symbol = symbol
self.Price = RollingWindow[float](period)

def update(self, value):
self.Price.Add(value)

def is_ready(self) -> bool:
return self.Price.IsReady

def performance(self, values_to_skip = 0) -> float:
closes = [x for x in self.Price][values_to_skip:]
return (closes[0] / closes[-1] - 1)

def volatility(self):
prices = np.array([x for x in self.Price])
daily_returns = prices[:-1] / prices[1:] - 1
return np.std(daily_returns)