中金固收 | 风格轮动：规律、策略与Python实现

def getDictPairs():
    return OrderedDict({"BigSmall": ["000016.SH", "000852.SH"],
            "ValueGrowth": ["399371.SZ", "399370.SZ"]})

def divergence(dictPairs, start, end, lstDay=None):
    if lstDay is None:
        lstDay = [40]
    
    lstCodes = []
    for v in dictPairs.values():
        lstCodes += v

    _, dfPct = w.wsd(",".join(lstCodes), "Close", dayOffset(start, -max(lstDay) - 1),
                     end, usedf=True)

    dfRet = pd.DataFrame(index=dfPct.index)

    for n in lstDay:
        for k, v in dictPairs.items():
            dfPctRolling = dfPct.rolling(n).apply(lambda x: 100


    
 * (x[-1] / x[0] - 1))
            dfRet[k + ":{_n} days".format(_n=n)] = dfPctRolling[v[0]] - dfPctRolling[v[1]]
            
    dfRet.index = map(lambda x: x.strftime("%Y/%m/%d"), pd.to_datetime(dfRet.index))
    
    return dfRet.dropna(how="any")

def rotation(obj, codes, date, tempCodes, dfAssetBook):
    # 获取市场风格分化情况.
    if not hasattr(obj, 'divergence'):
        obj.divergence = divergence(dictPairs, start, end)

    # 获取转债标的风格数据，得到dictStyle为字典，共有size和style两个字段，各保存一个表格以注明大小盘、成长价值风格
    if not hasattr(obj, 'dictStyle'):
        obj.dictStyle = getStyle(obj)

    sizeScore, styleScore = obj.divergence.loc[date, ["BigSmall:40 days"


    
,"ValueGrowth:40 days"]]

    # 获取阈值的相关数据，这里我们采取动态把控，而非定量的阈值
    drSize, drStyle = 1 if sizeScore > 0 else -1,1 if styleScore > 0 else -1
    idx = obj.divergence.index.get_loc(date)
    # theta代表该风格过去1年在40交易日内累计偏移度的95%分位数，epsilon则是这一偏移度绝对值的0.5倍标准差
    # 下面判断当前状态与阈值距离时，则以 score - theta的绝对值是否大于epsilon为准
    thetaSize, thetaStyle = obj.divergence.iloc[idx - 250:idx].abs().quantile(0.95)
    epsilonSize, epsilonStyle = obj.divergence.iloc[idx - 250:idx].abs().std() / 2.
    
    # 经过计算后，下面stylePick的choice将显示为所选风格的字符
    stylePicks = {'size': {'score': sizeScore, 'direction': drSize, 'theta': thetaSize, 'epsilon': epsilonSize,
                           'choice': 0, 'category': ['large', 'small']},
                  'style': {'score': styleScore, 'direction': drStyle, 'theta': thetaStyle, 'epsilon': epsilonStyle,
                            'choice': 0, 'category': ['value', 'growth']}}

    for k, v in stylePicks.items():



    
        # 判定：当前状态距离阈值是否较远
        if abs(v['score'] - v['theta'] * v['direction']) >= v['epsilon']:
            
            v["choice"] = v["direction"] if abs(v["score"]) < v["theta"] else -v["direction"]

        else:
            v['choice'] = -v['direction']
            
        v['choice'] = v['category'][0] if v['choice'] == 1 else v['category'][1]
    # 这里选择对应风格的品种，取交集
    srs = obj.dictStyle['size'].loc[date, tempCodes] == stylePicks['size']['choice']
    srs *= obj.dictStyle['style'].loc[date, tempCodes] == stylePicks['style']['choice']

    return srs[srs].index

文章来源