研报复制（七）：A股行业动量的精细结构

量化小白

发布于 2023-04-03 20:21:36

46200

代码可运行

文章被收录于专栏：量化小白上分记量化小白上分记

运行总次数：0

代码可运行

本文是对开源金工报告《A股行业动量的精细结构》的复制，欢迎指正！

背景

报告主要观点如下

其中关于动量效应和行业动量纵向切割的部分，已经在上一篇中复制过，本文复制报告关于行业动量的横向切割部分。

需要说明的几点是：

报告需要用到行业指数成分股的日度数据，鉴于手中没有数据，本文的研究都基于米筐平台，可阅读原文在平台上查看源码；
报告用的申万一级行业，尝试了若干平台后发现，由于申万行业指数调整过，但几乎所有平台都没有办法获得部分行业2014年之前的成分股数据，所以本文的测试基于中信一级行业指数，代码已经封装好函数，如果手中有对应的数据源，可以用申万行业测试一遍；
后台回复“行业牵引力因子”获取报告原文和源码。

因子定义

测试时直接取最佳参数lambda = 60%，不做参数寻优。并将牵引力因子和黄金律因子、传统行业动量因子做对比分析。

复制结果

回测区间：2006年1月-2019年12月
回测标的：中信一级行业指数
调仓频率：月频
牵引力因子累计IC

因子IC均值和年化ICIR

IC均值4%，ICIR较低，因子稳定性一般。

分层测试

与黄金律因子、传统动量因子的对比分析

其中，f为牵引力因子，其他四个因子沿用上一篇中的符号定义，M0表示日内动量因子、M1为隔夜反转因子、M为M0,M1打分和生成的黄金律因子。从上图可以看出，牵引力因子IC和ICIR都高于其他因子。

黄金律因子分层

传统动量因子分层

因子相关性和IC相关性（spearman）

整体来看，牵引力因子和传统动量因子、黄金律因子的相关性都很低，说明可能会有一定的信息增益。更细致的增益分析可以进行因子正交化和famamacbeth回归，偷懒不做啦。

另外动量因子一般在更短周期上会表现更好，所以如果放在周频上可能会有提升，有兴趣的童鞋可以自己测一下。

代码

参数设定

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import datetime

zxcode = ['CI005001.INDX','CI005002.INDX','CI005003.INDX','CI005004.INDX','CI005005.INDX',
           'CI005006.INDX','CI005007.INDX','CI005008.INDX','CI005009.INDX','CI005010.INDX',
           'CI005011.INDX','CI005012.INDX','CI005013.INDX','CI005014.INDX','CI005015.INDX',
           'CI005016.INDX','CI005017.INDX','CI005018.INDX','CI005019.INDX','CI005020.INDX',
           'CI005021.INDX','CI005022.INDX','CI005023.INDX','CI005024.INDX','CI005025.INDX',
           'CI005026.INDX','CI005027.INDX','CI005028.INDX','CI005029.INDX','CI005030.INDX']

swcode = ['801010.INDX','801020.INDX','801030.INDX','801040.INDX','801050.INDX','801080.INDX',
          '801110.INDX','801120.INDX','801130.INDX','801140.INDX','801150.INDX','801160.INDX',
          '801170.INDX','801180.INDX','801210.INDX','801230.INDX','801720.INDX','801730.INDX',
          '801740.INDX','801750.INDX','801760.INDX','801770.INDX','801780.INDX','801790.INDX',
          '801880.INDX','801890.INDX']

startdate = datetime.date(2005,12,25)
enddate = datetime.date(2019,12,31)

数据获取

zxclass = get_industry_mapping(source='citics', market='cn')
zxclass = zxclass.first_industry_name.unique().tolist()

indexname = all_instruments(type='INDX', market='cn')
zxname = pd.DataFrame(zxcode,columns = ['classcode'])
zxname = pd.merge(zxname,indexname,left_on = ['classcode'],right_on = ['order_book_id'],how = 'left')
zxname = zxname[['classcode','symbol']]
zxname['classname'] = zxname.symbol.apply(lambda x:x[3:])

pricezx = get_price(zxcode, start_date=startdate, end_date=enddate,fields = ['open','close'],expect_df = True)
pricezx = pricezx.reset_index()
pricezx.columns = ['classcode','tradedate','s_dq_open','s_dq_close']

# 中信行业价格
pricezx = pd.merge(pricezx,zxname,left_on = ['classcode'],right_on = ['classcode'])
pricezx = pricezx.drop(['classcode','symbol'],axis = 1)
pricezx.head()

# 月度交易日序列
alldates = get_trading_dates(startdate, enddate)
monthdate = pd.DataFrame(alldates,columns = ['tradedate'])
monthdate['ym'] = monthdate.tradedate.apply(lambda x:x.year*100 + x.month)
monthdate = monthdate.groupby('ym').last()
monthdate = monthdate.tradedate.tolist()
monthdate[:5]

牵引力因子定义

def getFactor(dateend,ind,lambdas,indtype = 'zx'):
    datestart = get_previous_trading_date(dateend,20,market='cn')

    # 获取行业成分股
    if indtype == 'zx':
        slist = get_industry(ind, source='citics', date = dateend, market='cn')
    else:
        slist = index_components(ind, date=dateend,market='cn')

    # 计算涨跌幅
    price_in = get_price(slist, start_date=datestart, end_date=datestart,fields = ['close'],expect_df = True)
    price_out = get_price(slist, start_date=dateend, end_date=dateend,fields = ['close'],expect_df = True)

    price = pd.concat([price_in,price_out],axis = 1)
    ret = price_out.reset_index(level = 1).close/price_in.reset_index(level = 1).close - 1
    ret = pd.DataFrame(ret)
    ret.columns = ['ret']



    # 获取成分股过去一个月的成交额
    samount = get_price(slist, start_date=datestart, end_date=datestart,fields = ['total_turnover'],expect_df = True)


    # 累加成交量
    tot_amt = samount.reset_index().groupby(['order_book_id']).sum().reset_index()
    tot_amt = tot_amt.set_index(['order_book_id'])
    tot_amt.columns = ['amt']

    # 判断龙头股和非龙头股
    tot_amt = tot_amt.sort_values(by = 'amt',ascending = False)
    tot_amt['percent'] =tot_amt.cumsum()/tot_amt.sum()
    tot_amt['stocktype'] = 'L'
    tot_amt.loc[tot_amt.percent > max(tot_amt.percent[0],lambdas),'stocktype'] = 'S'

    # 合并
    retv = pd.merge(ret,tot_amt,left_index = True,right_index = True)
    
    # 切割计算因子
    f = retv.ret.groupby(retv.stocktype).mean()
    fvalue = f['L'] - f['S']
    return [dateend,ind,fvalue]

因子计算（需要5-10分钟）

allfzx = []
lambdas = 0.6
for ind in zxclass:
    print(ind + ' 开始！')
    for i in range(1,len(monthdate)):
        dateend = monthdate[i]
        fzx = getFactor(dateend,ind,lambdas,indtype = 'zx')
        allfzx.append(fzx)
    print("{} 完成!".format(ind))

allfzx = pd.DataFrame(allfzx,columns = ['tradedate','classname','f'])
allfzx.head()

收益率计算和因子测试部分代码汇总（同前几篇）

def getRet(price,freq ='m',if_shift = True):
    price = price.copy()
   
    if freq == 'w':
        price['weeks'] = price['tradedate'].apply(lambda x:x.isocalendar()[0]*100 + x.isocalendar()[1])
        ret = price.groupby(['weeks','classname']).last().reset_index()
        del ret['weeks']
    
    elif freq =='m':
        price['ym'] = price.tradedate.apply(lambda x:x.year*100 + x.month)
        ret = price.groupby(['ym','classname']).last().reset_index()
        del ret['ym']
    
    ret = ret[['tradedate','classname','s_dq_close']]
    if if_shift:
        ret = ret.groupby('classname').apply(lambda x:x.set_index('tradedate').s_dq_close.pct_change(1).shift(-1))
    else:
        ret = ret.groupby('classname').apply(lambda x:x.set_index('tradedate').s_dq_close.pct_change(1))
    
    ret = ret.reset_index()
    ret['tradedate'] = ret['tradedate'].apply(lambda x:x.date())
    ret = ret.rename(columns = {ret.columns[2]:'ret'})
    return ret

def getICSeries(factors,ret,method):
    # method = 'spearman';factors = fall.copy();

    icall = pd.DataFrame()
    fall = pd.merge(factors,ret,left_on = ['tradedate','classname'],right_on = ['tradedate','classname'])
    icall = fall.groupby('tradedate').apply(lambda x:x.corr(method = method)['ret']).reset_index()
    icall = icall.drop(['ret'],axis = 1).set_index('tradedate')

    return icall

def GroupTestAllFactors(factors,ret,groups):
    """
    一次性测试多个因子
    """
    fnames = factors.columns
    fall = pd.merge(factors,ret,left_on = ['classname','tradedate'],right_on = ['classname','tradedate'])
    Groupret = []
    Groupic = []
    for f in fnames: # f= fnames[2]
        if ((f != 'classname')&(f != 'tradedate')):
            fuse = fall[['classname','tradedate','ret',f]]
        
            fuse['groups'] = fuse[f].groupby(fuse.tradedate).apply(lambda x:np.ceil(x.rank()/(len(x)/groups)))
            result = fuse.groupby(['tradedate','groups']).apply(lambda x:x.ret.mean())
            result = result.unstack().reset_index()
            result.insert(0,'factor',f)
            Groupret.append(result)
# print(f)
    Groupret = pd.concat(Groupret,axis = 0).reset_index(drop = True)
    Groupret = Groupret.fillna(0)
    Groupnav = Groupret.iloc[:,2:].groupby(Groupret.factor).apply(lambda x:(1 + x).cumprod())
    Groupnav = pd.concat([Groupret[['tradedate','factor']],Groupnav],axis = 1)

    return Groupnav