动量、景气、拥挤度：三个指标玩转行业轮动策略

创作时间:

作者:

@小白创作中心

动量、景气、拥挤度：三个指标玩转行业轮动策略

引用

来源

https://xueqiu.com/8185159194/290606752

在量化投资领域，行业轮动策略一直备受关注。通过分析行业动量、景气度和拥挤度这三个指标，投资者可以更好地把握市场机会。本文将详细介绍这三个指标的计算方法，并通过回测检验展示它们在实际应用中的效果。

行业动量

行业动量是衡量行业过去表现的重要指标，通常以行业过去12个月相对于行业等权指数的信息比率来计算。行业动量反映了市场投资者的投票结果，可以帮助识别市场一致预期。

主要代码如下：

def query_momentum_factor(industries: list, watch_date: str) -> pd.DataFrame:
    df_price = pd.DataFrame()
    trade_days = list(get_trade_days(end_date=watch_date, count=251))
    print(watch_date)
    print(trade_days)
    for ind in industries:
        df = finance.run_query(query(finance.SW1_DAILY_PRICE).filter(finance.SW1_DAILY_PRICE.code == ind,
                                                                     finance.SW1_DAILY_PRICE.date.in_(trade_days)).order_by(
            finance.SW1_DAILY_PRICE.date.desc()).limit(251))[['date', 'close']]
        df.set_index('date', inplace=True)
        df.rename(columns={"close": ind}, inplace=True)
        print(df)
        if len(df_price) == 0:
            df_price = df
        else:
            df_price = df_price.join(df)
    df_price = df_price.sort_index()
    df_pct = df_price.pct_change()[1:]
    df_return = df_price.iloc[-1] / df_price.iloc[0] - 1
    bench = df_pct.mean(axis=1)
    bench_return = (bench + 1).product() - 1
    track_error = df_pct.sub(bench, axis=0)
    std_te = np.std(track_error)
    IR = (df_return - bench_return) / (250 ** 0.5 * std_te)
    return IR

from tqdm import tqdm_notebook
from jqdata import *
from jqfactor import get_factor_values
import pandas as pd
from dateutil.parser import parse

START_DATE = '2015-01-01'
END_DATE = '2024-03-14'
today = datetime.date.today()
formatted_today = today.strftime('%y%m%d')
momentum_factors = get_ind_factor(query_momentum_factor, START_DATE, END_DATE)

回测检验

从2021年1月1日开始，在每周末将所有行业按动量指标从小到大排序，分别取动量最小的四个行业（front_ret），动量最大的四个行业（last_ret）和行业平均收益（mean_ret），走势如下：

可见行业动量最小和最大的组合的表现都优于行业平均，行业动量最小的组合反而表现最好，偏离预期。这种现象可能与市场情绪、交易行为等因素有关，需要进一步研究。

行业景气度

行业景气度反映了行业的基本面状况，通常通过分析行业营业收入增速、净利润增速、ROE-TTM和预期短期盈利增长率等指标来计算。国盛金工基于财务报表的信息和分析师预测构建了一个简单的行业景气度指标。

主要代码如下：

def query_growth_factor(industries: list, watch_date: str) -> pd.DataFrame:
    import warnings
    warnings.filterwarnings("ignore")
    df = pd.DataFrame(columns=['OPERATING_REVENUE_GROWTH_RATE', 'NET_PROFIT_GROWTH_RATE', 'ROE_TTM',
                               'SHORT_TERM_PREDICTED_EARNINGS_GROWTH'], index=industries)
    watch_date = watch_date.strftime('%Y-%m-%d')
    for ind in industries:
        stock_list = FilterStocks(ind, watch_date, N=3).Get_Stocks
        if len(stock_list) == 0:
            df = df.drop(index=ind)
            continue
        jq_factors = [
            'operating_revenue_growth_rate',
            'net_profit_growth_rate',
            'roe_ttm',
            'short_term_predicted_earnings_growth'
        ]
        factor_data = get_factor_values(stock_list, jq_factors, start_date=watch_date, end_date=watch_date)
        factor_data = dict2frame(factor_data)
        df.loc[ind] = factor_data.mean()
    return df

from tqdm import tqdm_notebook
from jqdata import *
from jqfactor import get_factor_values
import pandas as pd
from dateutil.parser import parse

START_DATE = '2015-01-01'
END_DATE = '2024-03-14'
today = datetime.date.today()
formatted_today = today.strftime('%y%m%d')
growth_factors = get_ind_factor(query_growth_factor, START_DATE, END_DATE)
ind_OPERATING_REVENUE_GROWTH_RATE = growth_factors['OPERATING_REVENUE_GROWTH_RATE'].unstack()
ind_NET_PROFIT_GROWTH_RATE = growth_factors['NET_PROFIT_GROWTH_RATE'].unstack()
ind_ROE_TTM = growth_factors['ROE_TTM'].unstack()
ind_SHORT_TERM_PREDICTED_EARNINGS_GROWTH = growth_factors['SHORT_TERM_PREDICTED_EARNINGS_GROWTH'].unstack()
ind_OPERATING_REVENUE_GROWTH_RATE_norm = roll(ind_OPERATING_REVENUE_GROWTH_RATE, 310).apply(normalization_zscore)
ind_NET_PROFIT_GROWTH_RATE = roll(ind_NET_PROFIT_GROWTH_RATE, 310).apply(normalization_zscore)
ind_ROE_TTM = roll(ind_ROE_TTM, 310).apply(normalization_zscore)
ind_SHORT_TERM_PREDICTED_EARNINGS_GROWTH = roll(ind_SHORT_TERM_PREDICTED_EARNINGS_GROWTH, 310).apply(normalization_zscore)
ind_growth_final = (ind_OPERATING_REVENUE_GROWTH_RATE_norm + ind_NET_PROFIT_GROWTH_RATE + ind_ROE_TTM + ind_SHORT_TERM_PREDICTED_EARNINGS_GROWTH) / 4

回测检验

从2021年1月1日开始，在每周末将所有行业按景气指标从小到大排序，分别取景气度最小的四个行业（front_ret），景气度最大的四个行业（last_ret）和行业平均收益（mean_ret），走势如下：

结果显示行业收益率表现与行业景气度呈反比，这可能与市场情绪、交易行为等因素有关，需要进一步研究。

行业拥挤度

行业拥挤度反映了市场的非有效性，通过识别市场的非理性行为来捕获市场真正的有效信息。行业拥挤度指标通常通过分析行业换手率、波动率和beta等指标来计算。

主要代码如下：

def query_crowd_factor(industries: list, end_date: str) -> pd.DataFrame:
    end_date = end_date.strftime('%Y-%m-%d')
    df = pd.DataFrame(columns=['ind_turnover', 'ind_std', 'ind_beta'], index=industries)
    for ind in industries:
        stock_list = FilterStocks(ind, end_date, N=3).Get_Stocks
        if len(stock_list) == 0:
            df = df.drop(index=ind)
            continue
        turnover = get_valuation(stock_list, end_date=end_date, count=63, fields=['turnover_ratio'])
        turnover = turnover.set_index(['day', 'code']).unstack()
        mean_turnover = np.mean(turnover)
        ind_turnover = np.mean(mean_turnover)
        stock_price = get_price(stock_list, count=63, end_date=end_date, frequency='daily', fields=['close'])['close']
        stock_chg = stock_price.pct_change()[1:]
        stock_std = np.std(stock_chg)
        ind_std = np.mean(stock_std)
        index = "000001.XSHG"
        mkt_close = get_price(index, count=63, end_date=end_date, frequency='daily', fields=['close'])['close']
        mkt_chg = mkt_close.pct_change()[1:]
        beta = []
        stockalpha = []
        for i in stock_list:
            X = mkt_chg
            Y = stock_chg[i]
            mask = ~np.isnan(X) & ~np.isnan(Y)
            temp_beta, temp_stockalpha = st.linregress(X[mask], Y[mask])[:2]
            beta.append(temp_beta)
        ind_beta = np.mean(beta)
        df.loc[ind] = pd.Series({'ind_turnover': ind_turnover, 'ind_std': ind_std, 'ind_beta': ind_beta})
    mean_df = np.mean(df)
    df1 = df.div(mean_df, axis=1)
    return df1

from tqdm import tqdm_notebook
crowd_factors = get_ind_factor(query_crowd_factor, START_DATE, END_DATE)
ind_turnover = crowd_factors['ind_turnover'].unstack()
ind_std = crowd_factors['ind_std'].unstack()
ind_beta = crowd_factors['ind_beta'].unstack()
ind_turnover_norm = roll(ind_turnover, 310).apply(normalization_zscore)
ind_std_norm = roll(ind_std, 310).apply(normalization_zscore)
ind_beta_norm = roll(ind_beta, 310).apply(normalization_zscore)
ind_crowd_final = (ind_turnover_norm + ind_std_norm + ind_beta_norm) / 3