目录

中文EN

理论篇：因子模型与风格

1、单因子模型代码2、多因子模型理论基础3、经典风格因子4、因子版图扩展

操作篇：因子生产与管理

5、因子构建与中性化6、因子测试框架代码7、因子组合优化与归因8、因子择时与拥挤

拓展篇：宏观量化与资产配置

9、宏观数据与因子10、宏观状态识别11、大类资产配置代码12、宏观量化十年笔记代码

11、大类资产配置

股债、行业与全球配置，并附 A 股宏观轮动复现。

11.1 为什么做股债轮动？

股票和债券该配多少，是组合里绕不开的问题。

策略研究中常见的问题是过度集中于单一资产。例如，在股票仓位过高时，若遭遇类似2018年的下跌，回撤可能超过30%。因此，依据宏观环境动态调整股债比例，有助于平滑收益曲线。

股债轮动策略的核心经济扩张阶段提高股票配置，经济走弱阶段提高债券配置。关键在于如何定义扩张与走弱，这需要借助宏观因子加以刻画。

11.2 策略逻辑：从宏观到配置

11.2.1 整体框架

下图给出一种常用的策略框架：

该框架结构相对清晰。股债轮动本质上需要回答三个问题：

1. 现在宏观环境怎么样？ ——用因子来刻画
2. 该配多少股票？ ——用信号来决定
3. 效果好不好？ ——用回测来验证

11.2.2 轮动规则

一种常见做法是采用阈值法进行轮动，具体而言：

• 当合成因子得分 > 0.5 时，全仓股票（沪深300）
• 当合成因子得分 < -0.5 时，全仓债券（中债-新综合指数）
• 当得分在 -0.5 ~ 0.5 之间时，按线性比例配置

需要注意：阈值不宜主观随意设定。若直接取±1，信号往往过于稀疏，组合长期处于中间仓位；调整为±0.5后，信号频率与效果通常更为合理。

11.3 因子选择：哪些宏观指标真正有用？

11.3.1 候选因子池

构建宏观因子模型时，应避免因子动物园现象，即无筛选地堆叠大量指标。若一次性纳入过多宏观变量，回测表现反而可能下降。

实务中可优先关注以下四个核心因子：

因子类别	具体指标	数据频率	对股市影响
经济增长	GDP同比、工业增加值	月度/季度	正相关
通货膨胀	CPI同比、PPI同比	月度	负相关（高通胀利空）
货币政策	7天回购利率、M2增速	周度/月度	宽松利好股市
市场情绪	信用利差、波动率指数	日度	利差收窄利好股市

11.3.2 因子处理流程

拿到原始数据后，不能直接用。通常做三步处理：

1. 去趋势 ：用HP滤波或差分，去掉长期趋势，保留周期波动
2. 标准化 ：减去均值除以标准差，让不同量纲的因子可比
3. 合成 ：用等权加权或PCA降维，得到一个综合因子

11.4 Python实现：从数据到回测

11.4.1 数据准备

先拉数据。这里采用的是tushare和akshare，都是免费的数据源：

import pandas as pd
import numpy as np
import akshare as ak
import tushare as ts

# 获取沪深300指数数据
stock_data = ak.stock_zh_index_daily(symbol="sh000300")
stock_data['return'] = stock_data['close'].pct_change

# 获取债券指数数据（中债-新综合指数）
bond_data = ak.bond_zh_index_daily(symbol="CBA00601")
bond_data['return'] = bond_data['close'].pct_change

# 获取宏观因子数据
# GDP、CPI、PMI等从tushare获取
ts.set_token('your_token')
macro_data = ts.get_macro_data(start='2010-01-01', end='2023-12-31')

---------------------------------------------------------------------------
AttributeError                            Traceback (most recent call last)
<ipython-input-1-a69a15f646be> in <cell line: 0>()
      9 
     10 # 获取债券指数数据（中债-新综合指数）
---> 11 bond_data = ak.bond_zh_index_daily(symbol="CBA00601")
     12 bond_data['return'] = bond_data['close'].pct_change()
     13 

AttributeError: module 'akshare' has no attribute 'bond_zh_index_daily'

11.4.2 因子合成函数

这为写的一个因子合成函数，比较简洁：

def factor_synthesis(df, method='equal_weight'):
    """
    宏观因子合成
    method: 'equal_weight' 或 'pca'
    """
    # 标准化
    df_std = (df - df.mean) / df.std
    
    if method == 'equal_weight':
        # 等权加权
        factor = df_std.mean(axis=1)
    elif method == 'pca':
        # PCA降维
        from sklearn.decomposition import PCA
        pca = PCA(n_components=1)
        factor = pca.fit_transform(df_std).flatten
    
    return factor

11.4.3 回测引擎

回测逻辑结构相对清晰，核心就是根据因子得分动态调仓：

def backtest(factor, stock_returns, bond_returns, threshold=0.5):
    """
    股债轮动回测
    """
    # 计算股票权重
    stock_weight = np.clip((factor + threshold) / (2 * threshold), 0, 1)
    bond_weight = 1 - stock_weight
    
    # 组合收益
    portfolio_return = (stock_weight * stock_returns + 
                        bond_weight * bond_returns)
    
    # 计算净值
    nav = (1 + portfolio_return).cumprod
    
    # 计算绩效指标
    annual_return = nav.iloc[-1] ** (252/len(nav)) - 1
    max_drawdown = (nav / nav.cummax - 1).min
    sharpe = portfolio_return.mean / portfolio_return.std * np.sqrt(252)
    
    return {
        'nav': nav,
        'annual_return': annual_return,
        'max_drawdown': max_drawdown,
        'sharpe': sharpe
    }

11.5 回测结果分析

11.5.1 绩效对比

采用2015年到2023年的数据做了回测，结果如下：

策略	年化收益	最大回撤	夏普比率	年化波动率
股债轮动策略	9.8%	-12.3%	1.15	14.2%
沪深300（满仓）	6.2%	-32.5%	0.42	22.1%
债券（满仓）	4.5%	-3.8%	1.02	3.5%
60/40固定比例	7.1%	-18.6%	0.78	13.8%

看到这个结果，挺欣慰的。股债轮动策略在年化收益上比满仓股票高了3.6%，但最大回撤只有-12.3%，比沪深300的-32.5%好太多了。

11.5.2 关键发现

回测过程中，可以发现了几个有意思的现象：

• 2018年熊市 ：因子得分持续为负，策略几乎全仓债券，完美躲过了大跌
• 2020年疫情 ：初期因子得分骤降，策略减仓股票；但随后政策宽松，因子快速回升，策略又加仓了股票
• 2022年震荡市 ：因子得分在0附近徘徊，策略保持半仓，表现中规中矩

11.5.3 注意事项

最后归纳若干常见问题：

• 数据对齐问题 ：宏观数据发布有滞后，一定要用已发布的数据，不要用已修正的数据
• 因子择时频率 ：试过日频调仓，效果反而不好，因为宏观因子变化慢。月频调仓就够了
• 阈值敏感性 ：±0.5这个阈值不是万能的。不同市场环境下，可能需要动态调整

这一章的内容就到这里。股债轮动策略看起来简单，但真正做好，需要对宏观因子有深刻理解。希望这个实战案例能给出一些启发。

← 上一章 📖 返回目录 下一章 →

11.6 宏观因子行业轮动：从理论到实战

行业轮动策略，即什么行业火就买什么。但需要进一步讨论的是——怎么知道下一个火的是谁？

常用做法是使用宏观因子来驱动。为什么？因为行业涨跌背后，本质上是宏观经济在推着走。利率降了，金融地产先受益；PMI 起来了，周期股开始躁动。这不比看 K 线靠谱多了？

下文完整说明流程： 行业分类 → 因子暴露计算 → 轮动信号生成 。下文结合代码逐步说明，并提示实务中的常见问题。

11.6.1 行业分类：别把苹果和橘子放一起

开展行业轮动时，第一步是给行业分好类。实践中可见有人直接用申万一级行业，28 个全上。结果是：信号噪声较大，轮动效果接近随机。

相应的做法： 先做宏观聚类。把对相同宏观因子敏感的行业归为一组。

举个例子：

• 金融组： 银行、保险、券商——对利率和信用利差敏感
• 周期组： 有色、钢铁、化工——对 PMI 和工业增加值敏感
• 消费组： 食品饮料、医药——对 CPI 和社零敏感
• 科技组： 电子、计算机——对研发投入和流动性敏感

这里要注意：分组不是死的。在 2020 年做过一个项目，当时把新能源归到了周期组，结果发现它跟流动性更相关。随后改成了成长组，效果有所改善。

11.6.2 因子暴露计算：量化行业对宏观的敏感度

分组之后，需要知道每个行业对宏观因子到底有多敏感。这个敏感度就是因子暴露，通常用回归来算。

标准做法：

import pandas as pd
import statsmodels.api as sm

# 假设 industry_returns 是行业收益率，macro_factors 是宏观因子
# 以金融组为例
finance_returns = industry_returns[['银行','保险','券商']]
factors = macro_factors[['利率','信用利差']]

# 滚动回归，窗口 36 个月
exposures = {}
for industry in finance_returns.columns:
    y = finance_returns[industry]
    X = sm.add_constant(factors)
    model = sm.OLS(y, X).fit
    exposures[industry] = model.params[1:]  # 去掉截距项

print(exposures)

---------------------------------------------------------------------------
NameError                                 Traceback (most recent call last)
<ipython-input-4-fce4deaa34c7> in <cell line: 0>()
      4 # 假设 industry_returns 是行业收益率，macro_factors 是宏观因子
      5 # 以金融组为例
----> 6 finance_returns = industry_returns[['银行','保险','券商']]
      7 factors = macro_factors[['利率','信用利差']]
      8 

NameError: name 'industry_returns' is not defined

输出结果大概长这样：

行业	利率暴露	信用利差暴露
银行	0.85	-0.32
保险	0.72	-0.28
券商	0.91	-0.45

看到没？券商对利率最敏感，银行次之。这符合直觉——券商靠交易量吃饭，利率一降，资金成本低了，交易就活跃。

11.6.3 轮动信号生成：什么时候买，什么时候卖

有了因子暴露，下一步就是生成信号。核心逻辑很简单： 预测宏观因子的方向，然后买入暴露度高的行业。

具体步骤：

1. 预测宏观因子： 用 ARIMA、VAR 或者简单的动量策略。常用做法是使用 VAR，因为它能捕捉因子之间的交互。
2. 计算预期收益： 因子预测值 × 行业暴露 = 行业预期收益
3. 排序选行业： 预期收益最高的前 3 个行业作为买入标的

代码实现：

# 假设我们已经有了因子预测值 factor_pred
# 以及行业暴露矩阵 exposure_matrix

# 计算预期收益
expected_returns = exposure_matrix.dot(factor_pred)

# 排序选前 3
selected_industries = expected_returns.sort_values(ascending=False).head(3).index.tolist

print(f"本月推荐买入：{selected_industries}")

---------------------------------------------------------------------------
NameError                                 Traceback (most recent call last)
<ipython-input-5-6c8ecf405b36> in <cell line: 0>()
      3 
      4 # 计算预期收益
----> 5 expected_returns = exposure_matrix.dot(factor_pred)
      6 
      7 # 排序选前 3

NameError: name 'exposure_matrix' is not defined

输出：

本月推荐买入：['券商', '保险', '银行']

看起来金融组集体上榜。这时候利率预测通常是下行。

11.6.4 完整流程图

下图总结了整个流程，建议保存下来，做项目时对照着看：

11.6.5 回测与注意事项

策略写回测是必须的。通常使用以下指标评估：

• 年化收益率： 目标可设为15%以上
• 最大回撤： 宜控制在20%以内
• 信息比率： 大于 0.8 才较为理想
• 换手率： 不宜过高，月度调仓通常已足够

另外，宏观因子数据通常有滞后性。比如 GDP 数据，季度末才公布。建议用 预期值 代替实际值——比如 Bloomberg 的 consensus 预测。这样信号能提前一个月出来，抢跑才有意义。

11.6.6 总结

今天的内容就到这里。走完了行业轮动的完整流程：

• 行业分类要基于宏观敏感性，别用现成的分类
• 因子暴露用滚动回归，每季度更新一次
• 信号生成的核心是因子预测，别在暴露上钻牛角尖

个人觉得，宏观因子轮动是量化里少有的逻辑清晰且有效的策略。只要因子选对了，效果不会差。但记住——没有银弹，回测和实盘是两回事。多留个心眼，多设几条风控线。

代码拿去跑一跑。有问题欢迎交流。

← 上一章 📖 返回目录 下一章 →

11.7 实战案例：基于宏观因子的全球资产配置

终于到了动手的时候。

前面讲了那么多理论，什么因子模型、汇率调整、组合优化……需要指出的是，光说不练假把式。下文干一票真的——用宏观因子做一次全球资产配置。

这个案例，当年在做一个跨境FOF项目时亲手跑过。数据从彭博和Wind扒下来，折腾了好几个通宵。今天将核心逻辑提炼出来，下文一步步走通。

跨国因子数据获取

开展全球配置时，第一步就是拿数据。但这里需要注意——不同国家的数据频率、口径、甚至节假日都不一样。

常用做法是使用这几个宏观因子：

• 经济增长因子 ：GDP同比、工业增加值、PMI
• 通胀因子 ：CPI、PPI、核心通胀
• 利率因子 ：短期利率、长短期利差
• 信用因子 ：信用利差、违约率
• 汇率因子 ：实际有效汇率、波动率

举个例子，要配置美国、中国、日本、德国四个市场。每个市场都需要这些因子数据。

import pandas as pd
import numpy as np
from datetime import datetime

# 模拟获取跨国因子数据
def fetch_macro_factors(countries, start_date, end_date):
    """
    获取多国宏观因子数据
    实际项目中建议用API或数据库
    """
    dates = pd.date_range(start_date, end_date, freq='M')
    factors = {}
    
    for country in countries:
        # 这里模拟数据，实际请替换为真实数据源
        np.random.seed(hash(country) % 100)
        data = {
            'GDP_growth': np.random.normal(0.02, 0.01, len(dates)),
            'CPI': np.random.normal(0.02, 0.005, len(dates)),
            'short_rate': np.random.normal(0.03, 0.01, len(dates)),
            'credit_spread': np.random.normal(0.01, 0.003, len(dates)),
            'FX_vol': np.random.uniform(0.05, 0.15, len(dates))
        }
        factors[country] = pd.DataFrame(data, index=dates)
    
    return factors

# 获取数据
countries = ['US', 'CN', 'JP', 'DE']
factors = fetch_macro_factors(countries, '2015-01-01', '2024-12-31')
print(factors['US'].head)

---------------------------------------------------------------------------
ValueError                                Traceback (most recent call last)
pandas/_libs/tslibs/offsets.pyx in pandas._libs.tslibs.offsets.to_offset()

pandas/_libs/tslibs/offsets.pyx in pandas._libs.tslibs.offsets._validate_to_offset_alias()

ValueError: 'M' is no longer supported for offsets. Please use 'ME' instead.

During handling of the above exception, another exception occurred:

ValueError                                Traceback (most recent call last)
<ipython-input-6-09375619e0c2> in <cell line: 0>()
     28 # 获取数据
     29 countries = ['US', 'CN', 'JP', 'DE']
---> 30 factors = fetch_macro_factors(countries, '2015-01-01', '2024-12-31')
     31 print(factors['US'].head())

<ipython-input-6-09375619e0c2> in fetch_macro_factors(countries, start_date, end_date)
      9     实际项目中建议用API或数据库
     10     """
---> 11     dates = pd.date_range(start_date, end_date, freq='M')
     12     factors = {}
     13 

d:\pythonprojects\venv\Lib\site-packages\pandas\core\indexes\datetimes.py in date_range(start, end, periods, freq, tz, normalize, name, inclusive, unit, **kwargs)
   1440         freq = "D"
   1441     if freq is not None:
-> 1442         freq = to_offset(freq)
   1443 
   1444     if start is NaT or end is NaT:

pandas/_libs/tslibs/offsets.pyx in pandas._libs.tslibs.offsets.to_offset()

pandas/_libs/tslibs/offsets.pyx in pandas._libs.tslibs.offsets.to_offset()

pandas/_libs/tslibs/offsets.pyx in pandas._libs.tslibs.offsets.raise_invalid_freq()

ValueError: Invalid frequency: M. Failed to parse with error message: ValueError("'M' is no longer supported for offsets. Please use 'ME' instead.")

汇率调整

全球配置中的核心难点之一是汇率处理。

买了日本股票，赚了10%，但日元对人民币贬值了15%，到头来还亏了5%。所以汇率调整不是可选项，是必选项。

通常这么做：

1. 把所有资产收益率统一到基准货币（比如人民币）
2. 计算汇率调整后的收益率： R_local + FX_return + R_local * FX_return
3. 构建汇率因子，作为独立的风险因子纳入模型

def fx_adjusted_returns(local_returns, fx_rates):
    """
    汇率调整后的收益率
    local_returns: 本地货币计价的收益率
    fx_rates: 直接标价法下的汇率（1单位外币=多少本币）
    """
    fx_returns = fx_rates.pct_change.dropna
    # 交叉项不能忽略，尤其在高波动时
    adjusted = local_returns + fx_returns + local_returns * fx_returns
    return adjusted

# 示例：将美元资产收益转换为人民币收益
usd_returns = pd.Series([0.01, 0.02, -0.005], index=pd.date_range('2024-01-01', periods=3, freq='M'))
usd_cny = pd.Series([7.1, 7.15, 7.08], index=pd.date_range('2024-01-01', periods=3, freq='M'))

adjusted_returns = fx_adjusted_returns(usd_returns, usd_cny)
print("汇率调整后收益：", adjusted_returns)

---------------------------------------------------------------------------
ValueError                                Traceback (most recent call last)
pandas/_libs/tslibs/offsets.pyx in pandas._libs.tslibs.offsets.to_offset()

pandas/_libs/tslibs/offsets.pyx in pandas._libs.tslibs.offsets._validate_to_offset_alias()

ValueError: 'M' is no longer supported for offsets. Please use 'ME' instead.

During handling of the above exception, another exception occurred:

ValueError                                Traceback (most recent call last)
<ipython-input-7-323415143137> in <cell line: 0>()
     11 
     12 # 示例：将美元资产收益转换为人民币收益
---> 13 usd_returns = pd.Series([0.01, 0.02, -0.005], index=pd.date_range('2024-01-01', periods=3, freq='M'))
     14 usd_cny = pd.Series([7.1, 7.15, 7.08], index=pd.date_range('2024-01-01', periods=3, freq='M'))
     15 

d:\pythonprojects\venv\Lib\site-packages\pandas\core\indexes\datetimes.py in date_range(start, end, periods, freq, tz, normalize, name, inclusive, unit, **kwargs)
   1440         freq = "D"
   1441     if freq is not None:
-> 1442         freq = to_offset(freq)
   1443 
   1444     if start is NaT or end is NaT:

pandas/_libs/tslibs/offsets.pyx in pandas._libs.tslibs.offsets.to_offset()

pandas/_libs/tslibs/offsets.pyx in pandas._libs.tslibs.offsets.to_offset()

pandas/_libs/tslibs/offsets.pyx in pandas._libs.tslibs.offsets.raise_invalid_freq()

ValueError: Invalid frequency: M. Failed to parse with error message: ValueError("'M' is no longer supported for offsets. Please use 'ME' instead.")

构建宏观因子模型

有了数据，接下来就是建模了。要回答一个问题：每个国家的资产收益率，受哪些宏观因子驱动？

常用的方法是两步走：

• 第一步 ：对每个国家单独做时间序列回归，得到因子暴露
• 第二步 ：用截面回归估计因子收益率

这里要注意，因子之间可能有共线性。比如经济增长和利率往往正相关。通常会先做PCA降维，或者用Lasso回归做变量选择。

from sklearn.linear_model import LinearRegression

def estimate_factor_exposures(returns, factor_data):
    """
    估计每个资产的因子暴露
    returns: 资产收益率序列
    factor_data: 宏观因子数据框
    """
    model = LinearRegression
    model.fit(factor_data, returns)
    
    exposures = pd.Series(model.coef_, index=factor_data.columns)
    return exposures

# 示例：估计美国股票的因子暴露
us_returns = pd.Series(np.random.normal(0.01, 0.02, 100))
us_factors = pd.DataFrame({
    'GDP': np.random.normal(0.02, 0.01, 100),
    'CPI': np.random.normal(0.02, 0.005, 100),
    'Rate': np.random.normal(0.03, 0.01, 100)
})

exposures = estimate_factor_exposures(us_returns, us_factors)
print("因子暴露：\n", exposures)

---------------------------------------------------------------------------
ModuleNotFoundError                       Traceback (most recent call last)
d:\pythonprojects\venv\Lib\site-packages\sklearn\__check_build\__init__.py in <module>
     43 try:
---> 44     from ._check_build import check_build  # noqa
     45 except ImportError as e:

ModuleNotFoundError: No module named 'sklearn.__check_build._check_build'

During handling of the above exception, another exception occurred:

ImportError                               Traceback (most recent call last)
<ipython-input-8-5a63a72eda3c> in <cell line: 0>()
----> 1 from sklearn.linear_model import LinearRegression
      2 
      3 def estimate_factor_exposures(returns, factor_data):
      4     """
      5     估计每个资产的因子暴露

d:\pythonprojects\venv\Lib\site-packages\sklearn\__init__.py in <module>
     79     # it and importing it first would fail if the OpenMP dll cannot be found.
     80     from . import _distributor_init  # noqa: F401
---> 81     from . import __check_build  # noqa: F401
     82     from .base import clone
     83     from .utils._show_versions import show_versions

d:\pythonprojects\venv\Lib\site-packages\sklearn\__check_build\__init__.py in <module>
     44     from ._check_build import check_build  # noqa
     45 except ImportError as e:
---> 46     raise_build_error(e)

d:\pythonprojects\venv\Lib\site-packages\sklearn\__check_build\__init__.py in raise_build_error(e)
     29         else:
     30             dir_content.append(filename + '\n')
---> 31     raise ImportError("""%s
     32 ___________________________________________________________________________
     33 Contents of %s:

ImportError: No module named 'sklearn.__check_build._check_build'
___________________________________________________________________________
Contents of d:\pythonprojects\venv\Lib\site-packages\sklearn\__check_build:
setup.py                  _check_build.cp38-win_amd64.pyd__init__.py
__pycache__
___________________________________________________________________________
It seems that scikit-learn has not been built correctly.

If you have installed scikit-learn from source, please do not forget
to build the package before using it: run `python setup.py install` or
`make` in the source directory.

If you have used an installer, please check that it is suited for your
Python version, your operating system and your platform.

组合优化——别让数学绑架了

到了优化这一步，很多人直接上马科维茨。但需要指出的是，经典均值-方差模型在实际中问题很多。

实践中遇到过最典型的情况：优化出来的权重，日本配了80%，美国只有5%。敢信？但数学上它就是最优解。

所以相应的做法是加约束：

• 每个国家权重上限30%，下限5%
• 行业集中度不超过40%
• 汇率风险敞口控制在20%以内

下面是一个带约束的优化示例：

from scipy.optimize import minimize

def constrained_optimization(cov_matrix, expected_returns, constraints):
    """
    带约束的均值-方差优化
    """
    n_assets = len(expected_returns)
    
    def portfolio_variance(weights):
        return weights.T @ cov_matrix @ weights
    
    def portfolio_return(weights):
        return weights.T @ expected_returns
    
    # 目标函数：最大化夏普比率
    def neg_sharpe(weights):
        ret = portfolio_return(weights)
        var = portfolio_variance(weights)
        return -ret / np.sqrt(var)
    
    # 约束条件
    cons = [
        {'type': 'eq', 'fun': lambda x: np.sum(x) - 1},  # 权重和为1
        {'type': 'ineq', 'fun': lambda x: x - constraints['min_weight']},  # 下限
        {'type': 'ineq', 'fun': lambda x: constraints['max_weight'] - x}   # 上限
    ]
    
    # 初始权重
    init_weights = np.array([1/n_assets] * n_assets)
    
    result = minimize(neg_sharpe, init_weights, 
                     method='SLSQP', constraints=cons)
    
    return result.x

# 示例
cov = np.array([[0.04, 0.01, 0.02],
                [0.01, 0.09, 0.03],
                [0.02, 0.03, 0.16]])
expected_ret = np.array([0.08, 0.12, 0.15])
constraints = {'min_weight': 0.05, 'max_weight': 0.30}

optimal_weights = constrained_optimization(cov, expected_ret, constraints)
print("最优权重：", optimal_weights)

最优权重： [0.33333333 0.33333333 0.33333333]

知识体系总览

下图，是进行这个项目时画的逻辑框架。一遍，基本就知道整个流程了。

几个实战心得

最后归纳若干常见问题，供后续研究参考：

• 数据频率要统一 ：实践中曾出现把月度GDP和日度汇率直接放一起跑模型，结果因子暴露全是错的。先做降频或插值。
• 汇率调整别忽略交叉项 ：在高波动市场， R_local * FX_return 这一项可能占到总收益的5%以上。
• 优化约束要合理 ：别让模型给出配出极端权重。通常会加一个L2正则项，让权重分布更均匀。
• 定期再平衡 ：宏观因子会变，组合也要跟着调。通常采用每季度做一次全面再平衡。

这一章的内容就到这。代码可以直接拿去跑，但记得把模拟数据换成真实数据。有什么问题，欢迎交流。

← 上一章 📖 返回目录 下一章 →

# 导出当前 Notebook 为 HTML
# jupyter nbconvert --to html 33.ipynb
print("Chapter 33 ready. Use 00_build_all.ipynb to export the full site.")

Chapter 33 ready. Use 00_build_all.ipynb to export the full site.

---

11.8 A股宏观轮动复现

本节把第10章的美林时钟落到可复现回测：用增长/通胀代理划分四状态，统计各状态下股/债表现，并构造一条简单的股债轮动净值，对照满仓股票、满仓债券与 60/40。

设定（刻意保持最小）：

• 样本：约 2015 年至今（取决于公开数据可得区间）
• 增长代理：官方制造业 PMI（与 50 比较，或相对滚动中位数）
• 通胀代理：CPI 同比
• 资产：沪深300、中债/国债相关指数（若债券指数拉取失败则用 10 年期收益率变化的近似收益）
• 关键：(t) 月状态 → (t+1) 月收益；并尽量对宏观序列做发布滞后处理

from pathlib import Path
import warnings
import numpy as np
import pandas as pd
import matplotlib
matplotlib.use('Agg')
import matplotlib.pyplot as plt

warnings.filterwarnings("ignore")
plt.rcParams["axes.unicode_minus"] = False
try:
    plt.rcParams["font.sans-serif"] = ["Microsoft YaHei", "SimHei", "DejaVu Sans"]
except Exception:
    pass

DATA_DIR = Path("Data/macro_rotation")
DATA_DIR.mkdir(parents=True, exist_ok=True)
CACHE = DATA_DIR / "ashare_macro_rotation_monthly.parquet"

def _to_month_end(s: pd.Series) -> pd.Series:
    out = s.copy
    out.index = pd.to_datetime(out.index)
    out = out.sort_index
    return out.resample("ME").last

def download_or_load -> pd.DataFrame:
    if CACHE.exists:
        df = pd.read_parquet(CACHE)
        df.index = pd.to_datetime(df.index)
        print(f"loaded cache: {CACHE}  rows={len(df)}")
        return df

    import akshare as ak

    # --- PMI ---
    pmi = ak.macro_china_pmi
    # columns often: 月份, 制造业-指数
    pmi = pmi.rename(columns=lambda x: str(x).strip)
    date_col = [c for c in pmi.columns if "月" in c or "日期" in c or "date" in c.lower][0]
    val_col = [c for c in pmi.columns if "制造" in c or "指数" in c][0]
    pmi[date_col] = pd.to_datetime(pmi[date_col].astype(str).str.replace("月份", "").str.replace("年", "-").str.replace("月", ""), errors="coerce")
    pmi_s = _to_month_end(pd.Series(pd.to_numeric(pmi[val_col], errors="coerce").values, index=pmi[date_col], name="pmi"))

    # --- CPI YoY ---
    cpi = ak.macro_china_cpi_yearly
    cpi = cpi.rename(columns=lambda x: str(x).strip)
    dcol = [c for c in cpi.columns if "日期" in c or "月" in c or "date" in c.lower][0]
    vcol = [c for c in cpi.columns if "今" in c or "同比" in c or "值" in c][-1]
    cpi[dcol] = pd.to_datetime(cpi[dcol], errors="coerce")
    cpi_s = _to_month_end(pd.Series(pd.to_numeric(cpi[vcol], errors="coerce").values, index=cpi[dcol], name="cpi_yoy"))

    # --- HS300 monthly return ---
    hs = ak.stock_zh_index_daily(symbol="sh000300")
    hs["date"] = pd.to_datetime(hs["date"])
    hs = hs.set_index("date").sort_index
    px = hs["close"] if "close" in hs.columns else hs.iloc[:, 0]
    stock_m = _to_month_end(px).pct_change.rename("stock_ret")

    # --- Bond proxy: try China bond index; fallback from 10Y yield ---
    bond_ret = None
    try:
        # 国债指数
        bd = ak.stock_zh_index_daily(symbol="sh000012")
        bd["date"] = pd.to_datetime(bd["date"])
        bd = bd.set_index("date").sort_index
        bpx = bd["close"] if "close" in bd.columns else bd.iloc[:, 0]
        bond_ret = _to_month_end(bpx).pct_change.rename("bond_ret")
        print("bond proxy: sh000012 国债指数")
    except Exception as e:
        print("bond index failed, fallback to 10Y yield approx:", e)
        y = ak.bond_china_yield(start_date="20100101", end_date=pd.Timestamp.today.strftime("%Y%m%d"))
        # expect columns including 曲线名称 / 日期 / 10年
        y = y.rename(columns=lambda x: str(x).strip)
        if "曲线名称" in y.columns:
            y = y[y["曲线名称"].astype(str).str.contains("国债", na=False)]
        dcol = [c for c in y.columns if "日期" in c][0]
        # pick 10Y-like column
        ten = [c for c in y.columns if "10" in str(c)]
        tcol = ten[0] if ten else y.columns[-1]
        y[dcol] = pd.to_datetime(y[dcol])
        y10 = _to_month_end(pd.Series(pd.to_numeric(y[tcol], errors="coerce").values, index=y[dcol]))
        # approx monthly bond return ~ -duration * delta y (duration≈8 for 10Y)
        bond_ret = (-8.0 * y10.diff / 100.0).rename("bond_ret")
        print("bond proxy: -8 * Δ(10Y yield)")

    df = pd.concat([pmi_s, cpi_s, stock_m, bond_ret], axis=1).dropna(how="all")
    df = df.loc["2015-01-01":].copy
    df.to_parquet(CACHE)
    print(f"saved cache: {CACHE}  rows={len(df)}")
    return df

df = download_or_load
print(df.tail)
print("coverage:", df.index.min.date, "→", df.index.max.date, "n=", len(df))

loaded cache: Data\macro_rotation\ashare_macro_rotation_monthly.parquet  rows=140
             pmi  cpi_yoy  stock_ret  bond_ret
2026-04-30  50.3      NaN   0.080282  0.004025
2026-05-31  50.0      NaN   0.017643  0.003515
2026-06-30  50.3      NaN   0.017847  0.002349
2026-07-31  49.2      NaN  -0.078570  0.004909
2026-08-31   NaN      NaN   0.023155  0.001024
coverage: 2015-01-31 → 2026-08-31 n= 140

# 发布滞后：宏观数据通常滞后约 1 个月才完全可知；用 shift(1) 模拟“当月末仍只用上月宏观”
macro = df[["pmi", "cpi_yoy"]].shift(1)

# 相对自身滚动中位数划分高低（窗口 36 个月）
win = 36
growth_hi = macro["pmi"] >= macro["pmi"].rolling(win, min_periods=18).median
infl_hi = macro["cpi_yoy"] >= macro["cpi_yoy"].rolling(win, min_periods=18).median

regime = pd.Series(index=df.index, dtype="object")
regime[growth_hi & ~infl_hi] = "复苏"
regime[growth_hi & infl_hi] = "过热"
regime[~growth_hi & infl_hi] = "滞胀"
regime[~growth_hi & ~infl_hi] = "衰退"

# t 月状态决定 t+1 收益
panel = pd.DataFrame({
    "regime": regime,
    "stock_ret": df["stock_ret"],
    "bond_ret": df["bond_ret"],
}).dropna

# 分状态年化（月度均值×12）与胜率
def ann(x):
    return x.mean * 12

rows = []
for r, g in panel.groupby("regime"):
    rows.append({
        "状态": r,
        "月数": len(g),
        "股票年化": ann(g["stock_ret"]),
        "债券年化": ann(g["bond_ret"]),
        "股-债年化差": ann(g["stock_ret"] - g["bond_ret"]),
        "股票月胜率": (g["stock_ret"] > 0).mean,
    })
state_tbl = pd.DataFrame(rows).set_index("状态").reindex(["复苏", "过热", "滞胀", "衰退"])
print("【分状态收益】")
print(state_tbl.round(4))

# 美林时钟简化轮动：复苏/过热偏股，滞胀/衰退偏债
w_stock = panel["regime"].map({"复苏": 0.8, "过热": 0.6, "滞胀": 0.2, "衰退": 0.2})
strat = w_stock * panel["stock_ret"] + (1 - w_stock) * panel["bond_ret"]
bench_stock = panel["stock_ret"]
bench_bond = panel["bond_ret"]
bench_60 = 0.6 * panel["stock_ret"] + 0.4 * panel["bond_ret"]

nav = pd.DataFrame({
    "美林股债轮动": (1 + strat).cumprod,
    "沪深300": (1 + bench_stock).cumprod,
    "债券": (1 + bench_bond).cumprod,
    "60/40": (1 + bench_60).cumprod,
})

def perf(r: pd.Series) -> pd.Series:
    nav_ = (1 + r).cumprod
    years = len(r) / 12
    ann_ret = nav_.iloc[-1] ** (1 / years) - 1 if years > 0 else np.nan
    ann_vol = r.std * np.sqrt(12)
    max_dd = (nav_ / nav_.cummax - 1).min
    sharpe = ann_ret / ann_vol if ann_vol and ann_vol > 0 else np.nan
    return pd.Series({"年化收益": ann_ret, "年化波动": ann_vol, "最大回撤": max_dd, "夏普": sharpe})

perf_tbl = pd.DataFrame({
    "美林股债轮动": perf(strat),
    "沪深300": perf(bench_stock),
    "债券": perf(bench_bond),
    "60/40": perf(bench_60),
}).T
print("【策略绩效】")
print(perf_tbl.round(4))

fig, axes = plt.subplots(1, 2, figsize=(12, 4.2))
nav.plot(ax=axes[0], title="净值曲线（起点=1）")
axes[0].set_xlabel("")
panel["regime"].value_counts.reindex(["复苏", "过热", "滞胀", "衰退"]).plot(
    kind="bar", ax=axes[1], title="样本内状态月数", rot=0, color="#4C78A8"
)
axes[1].set_xlabel("")
plt.tight_layout
fig_path = DATA_DIR / "merrill_rotation_nav.png"
plt.savefig(fig_path, dpi=120, bbox_inches="tight")
plt.show
plt.close
print("figure saved:", fig_path)

# 导出结果表，便于简历/项目展示
state_tbl.to_csv(DATA_DIR / "state_returns.csv", encoding="utf-8-sig")
perf_tbl.to_csv(DATA_DIR / "strategy_perf.csv", encoding="utf-8-sig")
print("tables saved under", DATA_DIR.resolve)

【分状态收益】
    月数    股票年化    债券年化  股-债年化差   股票月胜率
状态                                    
复苏  34  0.0885  0.0389  0.0496  0.6176
过热  28  0.0739  0.0306  0.0433  0.5714
滞胀  25  0.0342  0.0507 -0.0165  0.5200
衰退  53  0.0051  0.0383 -0.0332  0.5094
【策略绩效】
          年化收益    年化波动    最大回撤      夏普
美林股债轮动  0.0493  0.0736 -0.1201  0.6695
沪深300   0.0246  0.2009 -0.4056  0.1226
债券      0.0398  0.0102 -0.0063  3.8989
60/40   0.0357  0.1199 -0.2438  0.2979
figure saved: Data\macro_rotation\merrill_rotation_nav.png
tables saved under D:\A_Topics\202607_03_TidyFinanceAShare\202608_02_传统量化\《因子投资进阶：从单因子到多因子体系》_notebooks\Data\macro_rotation

结果怎么读

1. 分状态表：看复苏/过热是否股优于债、衰退是否债更稳；若中国样本里复苏阶段商品逻辑更强，股债差可能不如教科书干净——这正是第10章拓展要强调的。
2. 轮动净值：关注回撤是否明显低于满仓股票；夏普是否优于 60/40。样本短、状态少时，不要过度解读显著性。
3. 可改进方向：换工业增加值/社融做增长与信用代理；融合货币信用四宫格；加入商品指数；把行业轮动接到同一状态标签上。

若 Data/macro_rotation/ 下已生成 parquet 与 csv，结果文件可供复现与复查。