Py学习  »  Python

【中金固收·固收+】简单策略的“一加一大于二”及Python实现

中金固定收益研究 • 2 年前 • 671 次点击  


在此前的报告中我们已经分享了一部分简单的纯数量化和强化策略。但投资者也可能在现实约束面前难以抉择。最常见的,资金方可能对最大回撤有上限要求。因此这里我们要研究的是,如果我们有一些各有特点的储备策略,如何在特定的约束条件下选择出比较可行的策略组合。本期报告我们主要分享了一个简易实现策略组合的Python程序框架,供投资者参考。


正文


相比于单个策略达到的效果,投资者可能会面临更加复杂的约束——比如回撤,比如波动。《是时候,选出更好的策略了:转债策略库及测试》中我们分享了一部分简单的纯数量化策略及其回测结果,此后我们也在这基础上储备了一些强化策略,当然投资者不一定对其中的全部都熟悉。但即便投资者了解转债的各类策略,例如在《夹角余弦与转债及固收+基金策略》中提到的那些,投资者也可能在现实约束面前难以抉择。最常见的,资金方可能对最大回撤有上限要求。因此这里我们要研究的是,如果我们有一些各有特点的储备策略,如何在特定的约束条件下——例如收益、波动和回撤——选择出比较可行的策略组合。


图表:转债基金分年业绩评价

资料来源:万得资讯,中金公司研究部


我们的思路相对直观,即先测试基础策略是否符合组合评价要求,若没有基础策略符合要求,则通过两类方式去逼近要求:


1)通过策略分仓的方式,对比较接近要求的策略进行组合,简单举例,我们可以用50%的EasyBall搭配50%的低溢价策略去达到比转债指数更好的进攻性;


2)通过策略叠加的方式:将策略叠加后形成新的策略,以达到至少能启发思路的效果。例如我们发现正股高波动本身是有益于转债策略的因子,我们可以用高波动叠加双低,来达到更好的盈亏比。


图表:策略组合流程图

资料来源:中金公司研究部


策略评价体系构建

首先我们需要给目前储备的基础策略,在收益、波动、回撤的三维空间中给出定位。同时,为了此后更灵活地选择,我们也给出分年度计算收益、波动、回撤的计算方式,程序逻辑如下


净值年化评价


def  getAnnualiedReturn(srs):
    return 100* ((srs.pct_change().mean() + 1) ** 250 - 1.0)

def _getVol(srs):
    return 100*(srs.pct_change().std() * pd.np.sqrt(250.0))

def getMaxDraw(srs):
    srsMax = srs.rolling(len(srs), min_periods=1).max()
    return 100*(((srsMax - srs) / srs).max())

def strategyEvaluation(srs):
    ret = {'rt':getAnnualiedReturn(srs),'vol': _getVol(srs), 'md':getMaxDraw(srs)}
    ret['rt/vol'] = ret["rt"] / ret["vol"]
    ret['rt/md'] = ret["rt"] / ret["md"]

    return ret

资料来源:中金公司研究部


分年度净值评价体系


def strategyAnnualReview (srs, rf=0.025):

    # 分年的策略基本指标:年化回报、年化波动率、年度MDD、年度夏普比例
    Ret = srs.groupby(pd.DatetimeIndex(srs.index).year).apply(
        lambda x: (x[-1] / x[0] - 1.) * 100.)
    Vol = srs.groupby(pd.DatetimeIndex(srs.index).year).apply(
        lambda x: (x.pct_change().std() * np.sqrt(len(x)) * 100.))
    MDD = srs.groupby(pd.DatetimeIndex(srs.index).year).apply(
        lambda x: -(x / x.rolling(len(x), min_periods=1).max() - 1).min() * 100.)
    Sharpe = srs.groupby(pd.DatetimeIndex(srs.index).year).apply(
        lambda x: ((x[-1] / x[0] - 1.) - rf) / (
                (x.pct_change() - pow(1 + rf, 1 / len(x)) + 1.).std() * np.sqrt(len(x))))

    col = ['rt', 'vol', 'md', 'Sharpe']
    dfRet = pd.DataFrame([Ret, Vol, MDD, Sharpe]).transpose()
    dfRet.columns = col
    return dfRet.dropna()

资料来源:中金公司研究部


下图总结了我们核心基础策略库中28个策略的特征分布情况。


图表:核心基础策略库收益情况

资料来源:中金公司研究部,注:颜色深浅代表最大回撤,各策略净值数据截至2021年8月5日


调取策略历史净值情况

但仅策略表现不够,我们还需要将策略净值走势也保存在一个对象中,于是我们定义了如下的strategyClasses变量,其中strategies是由我们储备的策略所组成的字典(key为策略名称,对应的value为该策略函数),而ret则保存了各策略的净值。此外,为了避免策略反复调取而造成的损耗,我们暂时设定成,若本地已贮存则仅提取本地数据(readFromFile)。


策略储存对象


def plusPrepare(obj, start="2017/12/29"):
    end = obj.DB["Amt"].index[-1]

    obj.LR = getLR(obj, start, end)
    obj.LR_MACD = (obj.LR.rolling(20).mean() - obj.LR.rolling(120).mean()).diff(10).rolling(10).mean().shift(1)
    obj.LR_MACD_D = obj.LR_MACD.diff(1)


class strategyClasses(object):
    u"""以obj即数据库变量、addr:保存策略结果的位置(默认为strategyClasses)、
    start:默认2017/12/29、
    dictStrategy:策略字典,key为名称,value为函数名称,或者一个modify变量"
""

    def __init__(self, obj=None, start="2017/12/29", dictStrategy=None):

        self.ret = {}
        self.navs = None
        if obj is None:
            obj = cb.cb_data()
        self.obj = obj
        self.start = start
        self.end = self.obj.DB['Amt'].index[-1]
        self.obj.credit = getCredit(self.obj._excludeSpecial())
        self.dictStrategy = dictStrategy

    def  readFromFile(self, ftype=".xlsx"):
        self.ret = {}
        for k in self.dictStrategy.keys():
            try:
                self.ret[k] = pd.read_excel(self.addr + k + ftype, index_col=0)
            except IOError:
                print(k + " Failed, now run it")
                self.ret[k] = cb.frameStrategy(self.obj, self.start, roundMethod=21,
                                               selMethod=self.dictStrategy[k])

    def goInit(self):

        plusPrepare(self.obj, self.start)

        self.ret = {}
        for k, v in self.dictStrategy.items():
            print(k)
            _s = cb.frameStrategy(self.obj, self.start, selMethod=v, roundMethod=21)
            _s.to_excel(self.addr + k + ".xlsx")
            self.ret[k] = _s

    def display(self, plot=False):
        if self.ret == {}:
            strategyClasses.goInit(self)
        df = pd.concat(self .ret, axis=1)
        self.navs = df.loc[:, (slice(None), 'NAV')]
        if plot:
            return self.navs.plot()
        else:
            return self.navs

    def evaluate(self):
        if self.navs is None:
            strategyClasses.display(self)
        return pd.concat(strategyEvaluation(self.navs), axis=1)

资料来源:中金公司研究部


通过分仓搭建符合参数要求的策略组合

尽管理论上我们可以通过最优化来构造组合,但考虑可行性,我们仅考虑策略的两两组合。更实际地看,收益不是主要矛盾,关键在于回撤:两组合中由于收益可以直接线性组合,且只要回报要求不夸张,我们策略库中总有能够满足要求的——例如“趋势优先”。于是,核心问题则是如何牺牲部分收益来降波动/降回撤——即使有时候,从收益、风险比的角度来看并不经济,但有时这就是资金层面的要求,投资者不得不执行。


我们的思路是通过与第一策略波动相关性最低的策略来搜寻组合,从而降低回撤。当然,简单的两两遍历结合也能解决这个问题,但考虑效率,不考虑相关性时,我们的测试耗时30秒,而通过相关性筛选进行组合构建则仅需4.5秒。虽然30秒的时间看起来也可以接受,但投资者要考虑更大的数据量时,计算负担成倍增加的问题。


图表:基础策略净值波动相关性热力图

资料来源:万得资讯,中金公司研究部,注:各策略净值数据截至2021年8月5日


基础策略对照参数评价程序


def _comp(x):
    return x ** 2 if x <= 0 else 0


def evaluate(df, rt, vol, md):
    '''df是各策略的净值情况,rt, vol, md为约束条件'''

    # 将波动率与最大回撤取负值便于后续运算
    df.vol, df.md = -df.vol, -df.md
    matrix = dict(rt=rt, vol=-vol, md=-md)
    dfRet = pd.DataFrame(index=df.index, columns=['rt', 'vol', 'md'])

    for v in ['rt', 'vol', 'md']:
        if matrix[v] is not np.nan:
            for i in df.index:
                dfRet.loc[i, v] = _comp((df.loc[i, v] - matrix[v]))
    dfRet['distance'] = dfRet.sum(axis=1)
    return dfRet

资料来源:中金公司研究部


基于波动关联度而进行策略组合


def findPortfolio(rt=20, vol=np.nan, md=10):
    # 通过分仓来组合策略达成参数标准
    obj = strategyClasses()
    df = obj.evaluate()
    df.index = list(obj.dictStrategy.keys())

    dfRet = evaluate(df, rt, vol, md)

    if any(dfRet.distance == 0):
        return  df[dfRet.distance == 0]
    else:
        # 取净值波动关联度最小的进行组合
        dfNavs = obj.navs
        dfcorrmin = dfNavs.pct_change().corr().idxmin()

        dfTemp = pd.DataFrame(columns=['rt', 'vol', 'md'])
        for v1 in obj.dictStrategy.keys():
            v2 = dfcorrmin[v1][0][0]
            for k in range(5, 55, 5):
                k /= 100.
                test = dfNavs[v1] * k + dfNavs[v2] * (1 - k)
                test_str = v1 + str(k) + v2 + str(1 - k)
                dfTemp.loc[test_str] = list(strategyEvaluation(test.NAV).values())[:3]

        dfRet2 = evaluate(dfTemp, rt, vol, md)
        return dfTemp[dfRet2.distance == 0]

资料来源:中金公司研究部


倘若需要对每年净值业绩情况进行评价,则可以参照以下评价函数。当然此时约束特征值成倍增加,计算时间会拉长,且最后提取的组合有时并不能完全满足参数要求,需要将最后输出值调整为最接近的策略组合如dfRet.distance. nsmallest(10)。


基础策略分年度对照参数评价程序


def evaluateStrict(df, rt, vol, md):
    '''df是各策略的净值情况,rt, vol, md为约束条件'''
    
    # 考察过去每一年度是否均完成参数标准(后续主要参考这个选择策略)
    matrix = dict(rt=rt, vol=-vol, md=-md)
    dfRet = pd.DataFrame(index=df.columns, columns=['rt', 'vol', 'md'])
    for v in ['rt', 'vol', 'md']:
        if matrix[v] is not np.nan:
            for i in df.columns:
                score = 0
                dfTemp = strategyAnnualReview(df[i])
                dfTemp.vol, dfTemp.md = -dfTemp.vol, -dfTemp.md
                score += (dfTemp[v] - matrix[v]).apply(lambda x: _comp(x)).sum()
                dfRet.loc[i, v] = score
    dfRet['distance'] = dfRet.sum(axis=1)

    return dfRet

资料来源:中金公司研究部


通过策略叠加来创造符合参数要求的策略组合

策略叠加本质就是在原本策略的基础上用其他策略再做一次过滤——这样不可避免地,可能降低策略在经济意义上的直观性。当然实现起来的难度并不大。在此,我们简要分享一个方法,并不再深入探讨(其中,modifyObj是一个我们用于叠加策略的对象)。


基础策略分年度对照参数评价程序


def findPortfolioCombined(rt=20, vol=np.nan, md=1):

    # 通过两个策略叠加来完成参数要求

    sc = strategyClasses()
    sc.evaluate()
    dfTemp = pd.DataFrame(columns=['rt', 'vol', 'md'])
    matrix = dict(rt=rt, vol=-vol, md=-md)
    for v1 in sc.dictStrategy:
        for v2 in [key for key in sc.dictStrategy if key != v1]:
            m = modifyObj(sC.obj, preFunc=sC.dictStrategy[v1], subFunc=sC.dictStrategy[v2])
            m_str = v1 + v2
            try:
                test = cb.frameStrategy(sC.obj, sC.start, selMethod=m.func, roundMethod=21)
                dfTemp.loc[m_str] = list(strategyEvaluation(test.NAV).values())[:3]
            except Exception:

                # 由于部分策略两两叠加无法形成策略,因而进行剔除

                dfTemp.loc[m_str] = [np.nan, np.nan, np.nan]
    dfRet = evaluate(dfTemp, rt, vol, md)
    return dfTemp[dfRet.distance == 0]

资料来源:中金公司研究部



实测小结


在此我们试图回答两个问题:

1. 如何降低转债组合策略最大回撤?组合策略降回撤主要有三个思路,1)不同策略的组合效益;2)与大票AAA或者低价策略进行分仓;3)变动换仓频率。


图表:基础策略分年回撤结果(由高至低)

资料来源:万得资讯,中金公司研究部,注:2021年数据截至至2021年7月16日


具体来看,

1.1.   策略组合这个行为本身就具有降低波动和回撤的功能,尤其是低相关度的策略两两组合;


1.2.   尽管大票AAA转债在收益风险比、收益回撤比方面表现较差——主要是分子很小——但可以很显著地在不降名义仓位的情况下降低回撤。尽管,实际效果不及进攻型策略 + 空仓,但保持名义仓位也是部分客户的要求;


1.3.提高换仓频率对控制最大回撤有意义。


图表:基础策略回撤结果(由高至低)

资料来源:万得资讯,中金公司研究部,注:2021年数据截至至2021年7月16日


2.   从产品角度,这种程序当然不应局限于转债,我们还可以将这个思路拓宽至固收+产品更广泛的配置上。假定对于利率债或信用债,我们采取被动指数投资,则我们基于上述方法还可以构建符合参数标准的二级债基产品。在下图中,我们以典型的2:8分仓,来构建具备典型转债风格的二级债基产品。当然,用上述程序,我们可以搜寻到更多有实战价值的组合。


图表:模拟的二级债基收益情况

资料来源:万得资讯、中金公司研究部;注:利率债:中债-国债及政策性银行债财富(总值)指数;信用债:中债-信用债总财富(总值)指数;AAA(隐含)信用债:中债-市场隐含评级AAA信用债财富(总值)指数;短债:中债-新综合财富(1年以下)指数



推荐阅读


固收+基金风格的分步过滤测算及Python实现

转债基金的风格分解与Python实现

余弦与固收+基金策略识别

转债策略库及测试


文章来源

本文摘自:2021年8月6日已经发布的《简单策略的“一加一大于二”及Python实现》

罗 凡 SAC执业证书编号:S0080120070107

 冰 SAC执业证书编号:S0080515120002SFC CE Ref: BOM868

陈健恒 SAC执业证书编号:S0080511030011SFC CE Ref: BBM220



法律声明

向上滑动参见完整法律声明及二维码



Python社区是高质量的Python/Django开发社区
本文地址:http://www.python88.com/topic/118136
 
671 次点击