过去约十年,A 股与债券的涨跌背后,最常被提到的宏观力量仍是:经济增长、通胀、利率(必要时加上信用)。本章做一份可复现的十年笔记。
因子定义(增长、通胀、利率):
| 因子 | 代理变量 | 月度因子取值 |
|---|---|---|
| 增长 | 官方制造业 PMI | $\Delta$ PMI |
| 通胀 | CPI 同比 | $\Delta$ CPI同比(百分点) |
| 利率 | 10 年期国债收益率 | $\Delta$ 收益率(百分点);上升通常压制估值 |
| 信用 | 社融存量同比(若可得) | $\Delta$ 社融同比 |
贡献分解:
对每个日历年 $y$,用该年(或滚动窗口)的月度回归
$$ r_t = \alpha_y + \sum_f \beta_{y,f} F_{f,t-1} + \varepsilon_t $$其中 $F$ 为标准化后的因子(便于比较),并用 滞后一期 减轻同期反向因果。年贡献为
$$ C_{y,f} = \sum_{t \in y} \hat\beta_{y,f} F_{f,t-1}. $$贡献最大按 $|C_{y,f}|$ 排序(方向可看符号:正=同向拉动股票,负=拖累)。
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_decade")
DATA_DIR.mkdir(parents=True, exist_ok=True)
CACHE = DATA_DIR / "macro_factors_monthly.parquet"
def _month_end(s: pd.Series) -> pd.Series:
s = s.copy
s.index = pd.to_datetime(s.index)
return s.sort_index.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.rename(columns=lambda x: str(x).strip)
dcol = [c for c in pmi.columns if "月" in c or "日期" in c][0]
vcol = [c for c in pmi.columns if "制造" in c or "指数" in c][0]
raw = (
pmi[dcol].astype(str)
.str.replace("月份", "", regex=False)
.str.replace("年", "-", regex=False)
.str.replace("月", "", regex=False)
)
pmi[dcol] = pd.to_datetime(raw, errors="coerce")
pmi_s = _month_end(pd.Series(pd.to_numeric(pmi[vcol], errors="coerce").values, index=pmi[dcol], name="pmi"))
# CPI YoY
cpi = ak.macro_china_cpi_yearly.rename(columns=lambda x: str(x).strip)
dcol = [c for c in cpi.columns if "日期" in c or "月" in c][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 = _month_end(pd.Series(pd.to_numeric(cpi[vcol], errors="coerce").values, index=cpi[dcol], name="cpi_yoy"))
# 10Y yield (中美国债收益率接口更稳定)
y = ak.bond_zh_us_rate(start_date="20100101")
y = y.rename(columns=lambda x: str(x).strip)
dcol = "日期" if "日期" in y.columns else y.columns[0]
tcol = "中国国债收益率10年" if "中国国债收益率10年" in y.columns else [c for c in y.columns if "10" in str(c) and "中国" in str(c)][0]
y[dcol] = pd.to_datetime(y[dcol], errors="coerce")
y10 = _month_end(pd.Series(pd.to_numeric(y[tcol], errors="coerce").values, index=y[dcol], name="y10"))
print("y10 via bond_zh_us_rate:", tcol, "non-na=", y10.notna.sum)
# social financing YoY (credit) — best effort
credit_s = None
for fn_name in ["macro_china_shrzgm", "macro_china_new_financial_credit"]:
try:
fn = getattr(ak, fn_name)
sf = fn.rename(columns=lambda x: str(x).strip)
# try find a yoy-like column; else skip
dcol = [c for c in sf.columns if "日期" in c or "月" in c][0]
sf[dcol] = pd.to_datetime(sf[dcol], errors="coerce")
num_cols = [c for c in sf.columns if c != dcol]
# prefer 同比
yoy_cols = [c for c in num_cols if "同比" in str(c)]
col = yoy_cols[0] if yoy_cols else num_cols[0]
credit_s = _month_end(
pd.Series(pd.to_numeric(sf[col], errors="coerce").values, index=sf[dcol], name="credit")
)
print(f"credit proxy via {fn_name}: {col}")
break
except Exception as e:
print(f"skip credit source {fn_name}: {e}")
# HS300
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 = _month_end(px).pct_change.rename("stock_ret")
parts = [pmi_s, cpi_s, y10, stock]
if credit_s is not None:
parts.append(credit_s)
df = pd.concat(parts, axis=1).sort_index
df = df.loc["2014-01-01":].copy
df.to_parquet(CACHE)
print(f"saved {CACHE} rows={len(df)}")
return df
raw = download_or_load
print(raw.tail)
print("coverage:", raw.index.min.date, "→", raw.index.max.date)
0%| | 0/19 [00:00<?, ?it/s]
y10 via bond_zh_us_rate: 中国国债收益率10年 non-na= 200
skip credit source macro_china_shrzgm: 'NaTType' object has no attribute 'normalize'
skip credit source macro_china_new_financial_credit: 'NaTType' object has no attribute 'normalize'
saved Data\macro_decade\macro_factors_monthly.parquet rows=152
pmi cpi_yoy y10 stock_ret
2026-04-30 50.3 NaN 1.7473 0.080282
2026-05-31 50.0 NaN 1.7090 0.017643
2026-06-30 50.3 NaN 1.7330 0.017847
2026-07-31 49.2 NaN 1.7141 -0.078570
2026-08-31 NaN NaN 1.7114 0.023155
coverage: 2014-01-31 → 2026-08-31
增长 / 通胀用一阶差分;利率用 10Y 收益率变动。全部做 全样本 z-score,使 $\beta\times F$ 的量级可比。因子滞后一个月进入回归。
# --- build factors ---
fac = pd.DataFrame(index=raw.index)
fac["增长"] = raw["pmi"].diff
fac["通胀"] = raw["cpi_yoy"].diff
fac["利率"] = raw["y10"].diff
if "credit" in raw.columns and raw["credit"].notna.sum > 36:
# if level looks like YoY already, take diff; if huge levels, still diff
fac["信用"] = raw["credit"].diff
print("include 信用 factor")
else:
print("信用因子数据不足,本笔记使用增长/通胀/利率三因子")
fac = fac.dropna(how="all")
# standardize
fac_z = (fac - fac.mean) / fac.std(ddof=0)
stock = raw["stock_ret"]
factor_cols = list(fac_z.columns)
panel = fac_z.shift(1).join(stock, how="inner").dropna(subset=factor_cols + ["stock_ret"])
panel["year"] = panel.index.year
# focus on last ~10 calendar years with enough months
years = sorted([y for y in panel["year"].unique if y >= 2016])
panel = panel[panel["year"].isin(years)].copy
factors = [c for c in factor_cols if c in panel.columns]
print("factors:", factors)
print("sample months:", len(panel), "years:", years[0], "→", years[-1])
panel.tail
信用因子数据不足,本笔记使用增长/通胀/利率三因子 factors: ['增长', '通胀', '利率'] sample months: 118 years: 2016 → 2025
| 增长 | 通胀 | 利率 | stock_ret | year | |
|---|---|---|---|---|---|
| 2025-06-30 | 0.266108 | 1.286695 | 0.550046 | 0.024959 | 2025 |
| 2025-07-31 | 0.109163 | 0.044369 | -0.048963 | 0.035444 | 2025 |
| 2025-08-31 | -0.204725 | 0.044369 | 0.639224 | 0.103339 | 2025 |
| 2025-09-30 | 0.056849 | 0.458478 | 1.278616 | 0.032009 | 2025 |
| 2025-10-31 | 0.213793 | -0.162686 | 0.345609 | -0.000006 | 2025 |
from numpy.linalg import lstsq
MIN_OBS = 8 # months in a year to estimate
def year_regression(df_y: pd.DataFrame, factors: list[str]):
"""OLS stock_ret on factors; return betas, contrib series, R2."""
y = df_y["stock_ret"].values.astype(float)
X = np.column_stack([np.ones(len(df_y))] + [df_y[f].values.astype(float) for f in factors])
coef, *_ = lstsq(X, y, rcond=None)
alpha = coef[0]
betas = dict(zip(factors, coef[1:]))
fitted = X @ coef
ss_res = ((y - fitted) ** 2).sum
ss_tot = ((y - y.mean) ** 2).sum
r2 = 1 - ss_res / ss_tot if ss_tot > 0 else np.nan
contrib = {f: betas[f] * df_y[f] for f in factors}
return alpha, betas, contrib, r2
rows = []
contrib_yearly = []
beta_yearly = []
for y in years:
df_y = panel.loc[panel["year"] == y, factors + ["stock_ret"]].dropna
if len(df_y) < MIN_OBS:
print(f"skip {y}: only {len(df_y)} months")
continue
alpha, betas, contrib, r2 = year_regression(df_y, factors)
stock_ann = (1 + df_y["stock_ret"]).prod - 1
c_sum = {f: float(contrib[f].sum) for f in factors}
# winner by absolute contribution
winner = max(c_sum, key=lambda k: abs(c_sum[k]))
rows.append({
"年份": y,
"股票年收益": stock_ann,
"R2": r2,
"主导因子": winner,
"主导贡献": c_sum[winner],
**{f"贡献_{f}": c_sum[f] for f in factors},
**{f"beta_{f}": betas[f] for f in factors},
})
for f in factors:
contrib_yearly.append({"年份": y, "因子": f, "贡献": c_sum[f]})
beta_yearly.append({"年份": y, "因子": f, "beta": betas[f]})
summary = pd.DataFrame(rows).set_index("年份")
contrib_df = pd.DataFrame(contrib_yearly)
beta_df = pd.DataFrame(beta_yearly)
print("【逐年主导宏观因子】")
show_cols = ["股票年收益", "R2", "主导因子", "主导贡献"] + [f"贡献_{f}" for f in factors]
print(summary[show_cols].round(4).to_string)
summary.to_csv(DATA_DIR / "yearly_factor_contribution.csv", encoding="utf-8-sig")
contrib_df.to_csv(DATA_DIR / "contrib_long.csv", encoding="utf-8-sig")
print("saved tables →", DATA_DIR.resolve)
【逐年主导宏观因子】
股票年收益 R2 主导因子 主导贡献 贡献_增长 贡献_通胀 贡献_利率
年份
2016 -0.1128 0.2610 增长 -0.1100 -0.1100 -0.0610 0.0608
2017 0.2178 0.1485 利率 0.0740 0.0044 0.0001 0.0740
2018 -0.2531 0.1990 通胀 0.0467 -0.0085 0.0467 -0.0042
2019 0.3607 0.5440 通胀 -0.0469 -0.0069 -0.0469 -0.0105
2020 0.2721 0.7104 通胀 0.0639 0.0122 0.0639 -0.0042
2021 -0.0520 0.5050 利率 0.0305 0.0120 0.0247 0.0305
2022 -0.2163 0.1281 利率 -0.0550 0.0188 0.0220 -0.0550
2023 -0.1138 0.1653 通胀 0.0383 0.0098 0.0383 -0.0011
2024 0.1468 0.1008 利率 -0.1801 0.0025 -0.0198 -0.1801
2025 0.1794 0.3271 增长 0.0118 0.0118 -0.0004 0.0039
saved tables → D:\A_Topics\202607_03_TidyFinanceAShare\202608_02_传统量化\《因子投资进阶:从单因子到多因子体系》_notebooks\Data\macro_decade
# stacked bar of signed contributions + winner annotation
pivot = contrib_df.pivot(index="年份", columns="因子", values="贡献").reindex(columns=factors)
fig, axes = plt.subplots(2, 1, figsize=(11, 7.5), gridspec_kw={"height_ratios": [2.2, 1.2]})
# 1) contribution bars
ax = axes[0]
x = np.arange(len(pivot.index))
bottom_pos = np.zeros(len(pivot))
bottom_neg = np.zeros(len(pivot))
colors = {"增长": "#4C78A8", "通胀": "#F58518", "利率": "#54A24B", "信用": "#E45756"}
for f in factors:
vals = pivot[f].fillna(0).values
pos = np.where(vals >= 0, vals, 0.0)
neg = np.where(vals < 0, vals, 0.0)
ax.bar(x, pos, bottom=bottom_pos, label=f, color=colors.get(f, None), width=0.75)
ax.bar(x, neg, bottom=bottom_neg, color=colors.get(f, None), width=0.75)
bottom_pos += pos
bottom_neg += neg
ax.axhline(0, color="#333", lw=0.8)
ax.set_xticks(x)
ax.set_xticklabels(pivot.index.astype(int))
ax.set_ylabel("年贡献 Σ β·F(标准化因子)")
ax.set_title("沪深300:经典宏观因子年度贡献分解")
ax.legend(ncol=len(factors), frameon=False, loc="upper right")
# annotate winner
for i, y in enumerate(pivot.index):
w = summary.loc[y, "主导因子"] if y in summary.index else ""
v = summary.loc[y, "主导贡献"] if y in summary.index else 0
ymin, ymax = ax.get_ylim
ax.text(i, ymax * 0.92, f"{w}", ha="center", va="top", fontsize=9, color="#333")
# 2) |contribution| share
ax2 = axes[1]
abs_share = pivot.abs.div(pivot.abs.sum(axis=1), axis=0)
abs_share.plot(kind="bar", stacked=True, ax=ax2, color=[colors.get(f, None) for f in factors], width=0.75, legend=False)
ax2.set_ylabel("|贡献|占比")
ax2.set_xlabel("")
ax2.set_title("每年各因子绝对贡献占比")
ax2.set_xticklabels(abs_share.index.astype(int), rotation=0)
plt.tight_layout
fig_path = DATA_DIR / "decade_factor_contribution.png"
plt.savefig(fig_path, dpi=140, bbox_inches="tight")
plt.show
plt.close
print("figure:", fig_path)
figure: Data\macro_decade\decade_factor_contribution.png
下面用代码自动生成一年一句摘要,便于写进研究笔记或简历项目说明。解释时记住:
lines = []
for y, row in summary.iterrows:
parts = ", ".join(f"{f}={row[f'贡献_{f}']:+.3f}" for f in factors)
lines.append(
f"{int(y)}年:股票收益 {row['股票年收益']:+.1%},"
f"宏观回归 R²={row['R2']:.2f};"
f"|贡献|最大为【{row['主导因子']}】({row['主导贡献']:+.3f})。"
f"分解:{parts}。"
)
note = "\n".join(lines)
print(note)
(DATA_DIR / "decade_note.txt").write_text(note, encoding="utf-8")
# frequency of winners
print("\n【十年主导因子出现次数】")
print(summary["主导因子"].value_counts.to_string)
2016年:股票收益 -11.3%,宏观回归 R²=0.26;|贡献|最大为【增长】(-0.110)。分解:增长=-0.110, 通胀=-0.061, 利率=+0.061。 2017年:股票收益 +21.8%,宏观回归 R²=0.15;|贡献|最大为【利率】(+0.074)。分解:增长=+0.004, 通胀=+0.000, 利率=+0.074。 2018年:股票收益 -25.3%,宏观回归 R²=0.20;|贡献|最大为【通胀】(+0.047)。分解:增长=-0.009, 通胀=+0.047, 利率=-0.004。 2019年:股票收益 +36.1%,宏观回归 R²=0.54;|贡献|最大为【通胀】(-0.047)。分解:增长=-0.007, 通胀=-0.047, 利率=-0.010。 2020年:股票收益 +27.2%,宏观回归 R²=0.71;|贡献|最大为【通胀】(+0.064)。分解:增长=+0.012, 通胀=+0.064, 利率=-0.004。 2021年:股票收益 -5.2%,宏观回归 R²=0.51;|贡献|最大为【利率】(+0.030)。分解:增长=+0.012, 通胀=+0.025, 利率=+0.030。 2022年:股票收益 -21.6%,宏观回归 R²=0.13;|贡献|最大为【利率】(-0.055)。分解:增长=+0.019, 通胀=+0.022, 利率=-0.055。 2023年:股票收益 -11.4%,宏观回归 R²=0.17;|贡献|最大为【通胀】(+0.038)。分解:增长=+0.010, 通胀=+0.038, 利率=-0.001。 2024年:股票收益 +14.7%,宏观回归 R²=0.10;|贡献|最大为【利率】(-0.180)。分解:增长=+0.002, 通胀=-0.020, 利率=-0.180。 2025年:股票收益 +17.9%,宏观回归 R²=0.33;|贡献|最大为【增长】(+0.012)。分解:增长=+0.012, 通胀=-0.000, 利率=+0.004。 【十年主导因子出现次数】 主导因子 利率 4 通胀 4 增长 2
有人不用每年重估 β,而用全样本载荷,只让因子实现值随年变化。两种口径对照,可避免小样本年回归噪声。
# full-sample betas
df_all = panel[factors + ["stock_ret"]].dropna
_, betas_full, _, r2_full = year_regression(df_all, factors)
print("full-sample R2=", round(r2_full, 3), "betas=", {k: round(v, 3) for k, v in betas_full.items})
rows2 = []
for y in summary.index:
df_y = panel.loc[panel["year"] == y, factors].dropna
c_sum = {f: float((betas_full[f] * df_y[f]).sum) for f in factors}
winner = max(c_sum, key=lambda k: abs(c_sum[k]))
rows2.append({"年份": y, "主导因子_全样本β": winner, **{f"贡献_{f}": c_sum[f] for f in factors}})
robust = pd.DataFrame(rows2).set_index("年份")
cmp = summary[["主导因子"]].join(robust[["主导因子_全样本β"]])
cmp["一致"] = cmp["主导因子"] == cmp["主导因子_全样本β"]
print(cmp.to_string)
print("一致率:", f"{cmp['一致'].mean:.0%}")
robust.to_csv(DATA_DIR / "yearly_contrib_fullbeta.csv", encoding="utf-8-sig")
full-sample R2= 0.04 betas= {'增长': np.float64(0.008), '通胀': np.float64(-0.007), '利率': np.float64(0.0)}
主导因子 主导因子_全样本β 一致
年份
2016 增长 增长 True
2017 利率 利率 True
2018 通胀 通胀 True
2019 通胀 通胀 True
2020 通胀 通胀 True
2021 利率 通胀 False
2022 利率 通胀 False
2023 通胀 通胀 True
2024 利率 通胀 False
2025 增长 通胀 False
一致率: 60%
数据与图表默认缓存在 Data/macro_decade/。