主题切换
19.3 行业轮动模型
A 股市场的一个显著特征是行业轮动（Sector Rotation）——不同行业在不同宏观环境和市场阶段表现出极大的收益分化。2020-2021 年的新能源与消费赛道牛市，与同期地产和金融板块的低迷，形成了鲜明对比。行业轮动策略的目标是系统性地捕捉这种分化，将资金动态配置到最具上涨潜力的行业。
A股行业轮动的宏观驱动因素
A 股的行业轮动受到多重因素的驱动，理解这些驱动因素是构建有效轮动模型的前提。
宏观经济周期
中国经济周期的不同阶段对不同行业的盈利有差异化影响。经典的美林时钟框架（根据经济增长和通胀两个维度划分四个阶段）可以在 A 股市场找到对应的映射：
| 经济阶段 | 经济增速 | 通胀水平 | 占优行业（A股映射） |
|---|---|---|---|
| 复苏（Recovery） | 上行 | 下行 | 可选消费、科技（政策刺激敏感） |
| 过热（Overheat） | 上行 | 上行 | 周期（有色、煤炭、钢铁）、原材料 |
| 滞胀（Stagflation） | 下行 | 上行 | 必选消费（食品饮料、医药）、公用事业 |
| 衰退（Reflation） | 下行 | 下行 | 金融（银行、保险）、稳定红利 |
美林时钟在中国的应用中需要注意：中国经济周期受政策影响更大（尤其是货币政策和财政政策），周期的切换速度可能快于发达市场。
流动性环境
流动性是 A 股行业轮动的最重要变量之一。流动性宽松时期，资金成本低，高估值成长行业（如 TMT、新能源）往往表现突出；流动性收紧时期，低估值价值行业（如银行、地产）获得相对优势。
核心跟踪指标：
- 社会融资规模增速：宽信用的先行指标
- M1-M2 剪刀差：M1 增速高于 M2 增速通常预示资金活化程度高，利好股市
- DR007（7天期存款类机构质押式回购利率）：反映银行间市场流动性松紧
- 两融余额变化：杠杆资金的行业偏好
产业政策与监管
A 股受政策影响的程度远超成熟市场。"政策市"的特点意味着行业轮动模型中必须纳入政策因子：
- 五年规划：每个五年规划期初，重点扶持行业往往有超额表现。如"十四五"规划中的新能源、半导体、数字经济等。
- 产业政策力度：补贴力度、税收优惠、政府采购的行业偏向。
- 监管收紧：反垄断、数据安全、教育双减等政策对特定行业的冲击。
盈利周期
行业盈利增速的差异是轮动的根本驱动力。关键指标包括：
- 行业净利润增速（单季度同比和环比）
- 行业ROE水平及变化趋势
- 分析师盈利预测的修正方向
动量轮动 vs 均值回归轮动
动量轮动
动量轮动的核心假设是：过去表现好的行业在未来一段时间内继续表现好（"强者恒强"）。A 股行业动量效应的实证特征：
- 短期反转（1周以内）：追涨行业可能面临均值回归风险。
- 中期动量（1-12个月）：过去 3-6 个月表现最好的行业，未来 1-3 个月往往继续表现较好。这是行业ETF轮动策略最常见的信号频率。
- 长期反转（3-5年）：均值回归力量占据主导。
均值回归轮动
均值回归轮动认为，行业估值会在长期中回归合理中枢。估值极端的行业应该被反向操作：
- 高估值卖出：相对估值（如 PE 分位数）处于历史高位的行业未来收益偏低。
- 低估值买入：跌破历史估值底部的行业可能是逆向布局的机会。
混合策略
实践中，将动量信号和估值信号结合起来往往能获得更稳健的表现：
$$\text{Signal}_i = \alpha \cdot Z(\text{Momentum}_i) + (1-\alpha) \cdot Z(\text{Value}_i^{-1})$$
其中
此处有展示代码展开 ▼
python
import numpy as np
import pandas as pd
from scipy import stats
class SectorRotationStrategy:
"""A股行业轮动策略"""
# 申万一级行业分类（示例）
SW_SECTORS = [
'农林牧渔', '采掘', '化工', '钢铁', '有色金属',
'电子', '家用电器', '食品饮料', '纺织服装', '轻工制造',
'医药生物', '公用事业', '交通运输', '房地产', '商业贸易',
'休闲服务', '综合', '建筑材料', '建筑装饰', '电气设备',
'国防军工', '计算机', '传媒', '通信', '银行',
'非银金融', '汽车', '机械设备'
]
def __init__(self, sector_returns: pd.DataFrame,
sector_valuations: pd.DataFrame = None):
"""
Parameters
----------
sector_returns : pd.DataFrame
行业日收益率，index=日期，columns=行业
sector_valuations : pd.DataFrame
行业估值数据（PE/PB），index=日期，columns=行业
"""
self.returns = sector_returns
self.valuations = sector_valuations
def compute_momentum_signal(self, lookback: int = 60,
skip: int = 5) -> pd.DataFrame:
"""
计算动量信号
Parameters
----------
lookback : int
回溯窗口（交易日）
skip : int
跳过最近N天（避免短期反转效应）
"""
momentum = self.returns.rolling(lookback).apply(
lambda x: (1 + x[:-skip]).prod() - 1 if len(x) >= lookback else np.nan
)
return momentum
def compute_value_signal(self) -> pd.DataFrame:
"""
计算估值信号（估值分位数越低，信号越强）
"""
if self.valuations is None:
return None
# 滚动估值分位数（5年窗口）
value_percentile = self.valuations.rolling(252 * 5).apply(
lambda x: stats.percentileofscore(x, x.iloc[-1]) / 100
if len(x) >= 252 * 3 else np.nan
)
# 信号反转：低分位数 -> 高信号
value_signal = 1 - value_percentile
return value_signal
def compute_crowding_indicator(self, window: int = 20) -> pd.DataFrame:
"""
计算拥挤度指标
拥挤度高意味着该行业交易过热，前期收益可能来自追涨资金的集中涌入，
而非基本面的支撑，后续反转风险较大。
"""
# 成交量拥挤度
if 'volume' not in self.returns.columns.name.lower():
# 如果没有成交量数据，用收益率的波动率变化作为代理
vol = self.returns.rolling(window).std()
vol_ma = vol.rolling(252).mean()
vol_ratio = vol / vol_ma # 当前波动率/历史波动率
# 高波动率时期通常伴随拥挤交易
crowding = vol_ratio
else:
# 使用成交量指标
pass
return crowding
def compute_composite_signal(self, momentum_weight: float = 0.6,
value_weight: float = 0.3,
crowding_weight: float = 0.1) -> pd.DataFrame:
"""计算综合轮动信号"""
# 动量信号
mom_signal = self.compute_momentum_signal(lookback=60)
# Z-score标准化
mom_z = mom_signal.sub(mom_signal.mean(axis=1), axis=0)\
.div(mom_signal.std(axis=1), axis=0)
composite = mom_z * momentum_weight
# 估值信号
if self.valuations is not None:
val_signal = self.compute_value_signal()
val_z = val_signal.sub(val_signal.mean(axis=1), axis=0)\
.div(val_signal.std(axis=1), axis=0)
composite += val_z * value_weight
# 拥挤度惩罚
crowding = self.compute_crowding_indicator()
crowd_z = crowding.sub(crowding.mean(axis=1), axis=0)\
.div(crowding.std(axis=1), axis=0)
composite -= crowd_z * crowding_weight # 高拥挤度惩罚
return composite
def generate_allocation(self, signal: pd.Series,
n_select: int = 5,
equal_weight: bool = True) -> pd.Series:
"""
根据信号生成行业配置权重
Parameters
----------
signal : pd.Series
某日的行业信号值
n_select : int
选出的行业数量
equal_weight : bool
True=等权，False=信号加权
"""
# 选出信号最强的n个行业
top_sectors = signal.dropna().nlargest(n_select)
if equal_weight:
weights = pd.Series(1.0 / n_select, index=top_sectors.index)
else:
# 信号归一化为权重
positive_signals = top_sectors.clip(lower=0)
if positive_signals.sum() > 0:
weights = positive_signals / positive_signals.sum()
else:
weights = pd.Series(1.0 / n_select, index=top_sectors.index)
return weights
def backtest_rotation(self, start_date: str, end_date: str,
n_select: int = 5, rebalance_freq: str = 'M') -> pd.DataFrame:
"""行业轮动策略回测"""
signals = self.compute_composite_signal()
# 生成再平衡日期
all_dates = self.returns.loc[start_date:end_date].index
if rebalance_freq == 'M':
rebalance_dates = pd.Series(all_dates).dt.to_period('M').drop_duplicates()
rebalance_dates = all_dates[all_dates.isin(
all_dates.groupby(all_dates.to_period('M')).apply(lambda x: x[-1])
)]
elif rebalance_freq == 'W':
rebalance_dates = all_dates[::5] # 约每周
else:
rebalance_dates = all_dates
results = []
current_weights = None
for date in all_dates:
if date in rebalance_dates:
signal_t = signals.loc[date]
current_weights = self.generate_allocation(signal_t, n_select=n_select)
if current_weights is not None and len(current_weights) > 0:
daily_ret = self.returns.loc[date, current_weights.index]
portfolio_ret = (daily_ret * current_weights).sum()
else:
portfolio_ret = 0
results.append({
'date': date,
'return': portfolio_ret,
'n_stocks': len(current_weights) if current_weights is not None else 0
})
result_df = pd.DataFrame(results).set_index('date')
result_df['cum_return'] = (1 + result_df['return']).cumprod()
return result_df
# 行业ETF轮动（实战简化版）
class SectorETFRotation:
"""行业ETF轮动策略"""
# 代表性行业ETF（示例映射）
ETF_MAPPING = {
'银行': '512800',
'证券': '512880',
'医药': '512010',
'消费': '159928',
'科技': '515000',
'新能源': '516160',
'军工': '512660',
'半导体': '512480',
'白酒': '512690',
'房地产': '512200',
}
def __init__(self, etf_data: pd.DataFrame, risk_free_rate: float = 0.02):
self.data = etf_data
self.rf = risk_free_rate
def dual_momentum_signal(self, lookback_short: int = 20,
lookback_long: int = 60) -> pd.Series:
"""
双动量信号
逻辑：短期动量和长期动量的平均值，同时过滤绝对动量为负的ETF
"""
returns = self.data.pct_change()
mom_short = returns.rolling(lookback_short).apply(
lambda x: (1 + x).prod() - 1
)
mom_long = returns.rolling(lookback_long).apply(
lambda x: (1 + x).prod() - 1
)
# 绝对动量过滤：长期动量为负的ETF不参与
mom_filter = mom_long > 0
# 综合动量 = (短期 + 长期) / 2
composite = (mom_short + mom_long) / 2
composite = composite.where(mom_filter)
return composite.iloc[-1] # 返回最新一期的信号
def select_top_etfs(self, signal: pd.Series, n: int = 3) -> list:
"""选择信号最强的n只ETF"""
top = signal.dropna().nlargest(n)
return list(zip(top.index, top.values))
def risk_parity_weights(self, selected_etfs: list,
lookback: int = 60) -> dict:
"""
计算风险平价权重
Parameters
----------
selected_etfs : list
[(etf_code, signal_value), ...]
lookback : int
用于计算波动率的回溯窗口
"""
returns = self.data[list(zip(*selected_etfs))[0]].pct_change().iloc[-lookback:]
# 逆波动率权重
vols = returns.std()
inv_vols = 1 / vols
weights = inv_vols / inv_vols.sum()
return weights.to_dict()点击展开可浏览运行结果
📘 本段代码定义了 2 个函数/类：类 `SectorRotationStrategy`（A股行业轮动策略）、类 `SectorETFRotation`（行业ETF轮动策略）。该片段为教学展示（未包含独立运行的输入数据），可在实战练习中结合真实数据调用。
拥挤度指标与轮动信号
拥挤度（Crowding）是指某个行业中"过热"交易的程度，高拥挤度预示着未来反转风险上升。在 A 股市场，拥挤度检测尤为重要——A 股投资者结构的散户化和羊群效应特征使得拥挤度信号具有较高的预测价值。
拥挤度的多维指标体系
交易拥挤度：
- 行业成交额占全市场成交额比例的历史分位数
- 行业换手率的历史分位数
- 行业融资买入额的变化率
资金拥挤度：
- 公募基金在该行业的超配比例（相对基准权重）
- 北向资金在该行业的持仓集中度变化
- ETF 资金流的行业集中度
关注度拥挤度：
- 分析师覆盖数量变化
- 财经媒体提及频率
- 社交媒体讨论热度
估值拥挤度：
- 行业PE/PB相对全市场的历史分位数
- PEG（市盈率相对盈利增长比率）评估
此处有展示代码展开 ▼
python
def compute_crowding_composite(sector_data: pd.DataFrame) -> pd.DataFrame:
"""
计算行业拥挤度综合指标
返回拥挤度分数：0-1之间，1表示极度拥挤
"""
crowding = pd.DataFrame(index=sector_data.index, columns=sector_data.columns)
for sector in sector_data.columns:
sector_df = sector_data[sector]
# 1. 成交额占比拥挤度
if 'amount_pct' in sector_df:
amount_percentile = sector_df['amount_pct'].rolling(252).apply(
lambda x: stats.percentileofscore(x, x.iloc[-1]) / 100
)
crowding[sector] = amount_percentile * 0.3
# 2. 换手率拥挤度
if 'turnover' in sector_df:
turnover_ma = sector_df['turnover'].rolling(20).mean()
turnover_hist_ma = sector_df['turnover'].rolling(252).mean()
# 当换手率显著高于历史平均时，拥挤度高
turnover_ratio = turnover_ma / turnover_hist_ma
crowding[sector] = crowding[sector].fillna(0) + \
turnover_ratio.clip(0, 3) / 3 * 0.25
# 3. 收益率拥挤度（连续阳线天数）
if 'return' in sector_df:
consecutive_up = (sector_df['return'] > 0).astype(int)
streak = consecutive_up * (consecutive_up.groupby(
(consecutive_up != consecutive_up.shift()).cumsum()
).cumcount() + 1)
# 连续上涨超过5天，拥挤度上升
streak_score = (streak / 10).clip(0, 1)
crowding[sector] = crowding[sector].fillna(0) + streak_score * 0.20
# 4. 波动率拥挤度
if 'return' in sector_df:
vol = sector_df['return'].rolling(20).std()
vol_hist = sector_df['return'].rolling(252).std()
vol_ratio = vol / vol_hist
crowding[sector] = crowding[sector].fillna(0) + \
vol_ratio.clip(0, 3) / 3 * 0.25
return crowding.clip(0, 1) # 确保在 0-1 之间点击展开可浏览运行结果
📘 本段代码定义了 1 个函数/类：函数 `compute_crowding_composite`（计算行业拥挤度综合指标）。该片段为教学展示（未包含独立运行的输入数据），可在实战练习中结合真实数据调用。
拥挤度信号在轮动中的应用
拥挤度信号可以以两种方式集成到行业轮动模型中：
- 过滤法：排除拥挤度过高（如 > 80% 分位数）的行业，即使它们的动量信号很强。
- 惩罚法：在综合信号中减去拥挤度的惩罚项，拥挤度越高惩罚越大。
行业轮动是 A 股量化策略中最具实战价值的模块之一。成功的关键不在于模型复杂度，而在于对宏观经济和产业政策的深刻理解、合理的信号频率选择、以及审慎的拥挤度管理。在 A 股这个政策影响突出、行业分化剧烈的市场中，行业选择往往比个股选择对组合收益的贡献更大。