社区所有版块导航
Python
python开源   Django   Python   DjangoApp   pycharm  
DATA
docker   Elasticsearch  
aigc
aigc   chatgpt  
WEB开发
linux   MongoDB   Redis   DATABASE   NGINX   其他Web框架   web工具   zookeeper   tornado   NoSql   Bootstrap   js   peewee   Git   bottle   IE   MQ   Jquery  
机器学习
机器学习算法  
Python88.com
反馈   公告   社区推广  
产品
短视频  
印度
印度  
Py学习  »  Python

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

中金点睛 • 3 年前 • 550 次点击  


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