Over roughly the past decade, the macro forces most often cited behind A-share and bond moves remain growth, inflation, and rates (plus credit when needed). This chapter builds a reproducible ten-year note.
Factor definitions (growth, inflation, rates):
| Factor | Proxy | Monthly factor value |
|---|---|---|
| Growth | Official manufacturing PMI | $\Delta$ PMI |
| Inflation | CPI YoY | $\Delta$ CPI YoY (pp) |
| Rates | 10Y CGB yield | $\Delta$ yield (pp); rising yields usually compress valuations |
| Credit | Aggregate financing stock YoY (if available) | $\Delta$ financing YoY |
Contribution decomposition:
For each calendar year $y$, run a monthly regression on that year (or a rolling window)
$$ r_t = \alpha_y + \sum_f \beta_{y,f} F_{f,t-1} + \varepsilon_t $$where $F$ are standardized factors (for comparability), and a one-period lag reduces contemporaneous reverse causality. Annual contribution is
$$ C_{y,f} = \sum_{t \in y} \hat\beta_{y,f} F_{f,t-1}. $$Rank by $|C_{y,f}|$ (sign shows direction: positive = co-moving lift to equities; negative = drag).
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 (CN–US bond yield API is more stable)
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 column (同比 is Chinese API field name)
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
Growth and inflation use first differences; rates use changes in the 10Y yield. Apply a full-sample z-score so $\beta\times F$ magnitudes are comparable. Factors enter the regression lagged one month.
# --- build factors ---
fac = pd.DataFrame(index=raw.index)
fac["Growth"] = raw["pmi"].diff
fac["Inflation"] = raw["cpi_yoy"].diff
fac["Rates"] = 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["Credit"] = raw["credit"].diff
print("include Credit factor")
else:
print("Credit factor data insufficient; this note uses Growth / Inflation / Rates")
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({
"Year": y,
"Stock ann. return": stock_ann,
"R2": r2,
"Dominant factor": winner,
"Dominant contrib.": c_sum[winner],
**{f"Contrib_{f}": c_sum[f] for f in factors},
**{f"beta_{f}": betas[f] for f in factors},
})
for f in factors:
contrib_yearly.append({"Year": y, "Factor": f, "Contribution": c_sum[f]})
beta_yearly.append({"Year": y, "Factor": f, "beta": betas[f]})
summary = pd.DataFrame(rows).set_index("Year")
contrib_df = pd.DataFrame(contrib_yearly)
beta_df = pd.DataFrame(beta_yearly)
print("[Dominant macro factor by year]")
show_cols = ["Stock ann. return", "R2", "Dominant factor", "Dominant contrib."] + [f"Contrib_{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="Year", columns="Factor", values="Contribution").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 = {"Growth": "#4C78A8", "Inflation": "#F58518", "Rates": "#54A24B", "Credit": "#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("Annual contribution Σ β·F (standardized factors)")
ax.set_title("CSI 300: annual contribution of classic macro factors")
ax.legend(ncol=len(factors), frameon=False, loc="upper right")
# annotate winner
for i, y in enumerate(pivot.index):
w = summary.loc[y, "Dominant factor"] if y in summary.index else ""
v = summary.loc[y, "Dominant contrib."] 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("|Contribution| share")
ax2.set_xlabel("")
ax2.set_title("Share of absolute contribution by factor each year")
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
The next cell auto-generates a one-line summary per year for research notes or resume project write-ups. When interpreting:
lines = []
for y, row in summary.iterrows:
parts = ", ".join(f"{f}={row[f'Contrib_{f}']:+.3f}" for f in factors)
lines.append(
f"{int(y)}: stock return {row['Stock ann. return']:+.1%}, "
f"macro regression R²={row['R2']:.2f}; "
f"largest |contrib.| = [{row['Dominant factor']}] ({row['Dominant contrib.']:+.3f}). "
f"Breakdown: {parts}."
)
note = "\n".join(lines)
print(note)
(DATA_DIR / "decade_note.txt").write_text(note, encoding="utf-8")
# frequency of winners
print("\n[Dominant-factor counts over the decade]")
print(summary["Dominant factor"].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
An alternative skips re-estimating β each year and uses full-sample loadings, letting only factor realizations vary by year. Comparing the two definitions reduces noise from small-sample annual regressions.
# 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({"Year": y, "Dominant factor (full-sample β)": winner, **{f"Contrib_{f}": c_sum[f] for f in factors}})
robust = pd.DataFrame(rows2).set_index("Year")
cmp = summary[["Dominant factor"]].join(robust[["Dominant factor (full-sample β)"]])
cmp["Agree"] = cmp["Dominant factor"] == cmp["Dominant factor (full-sample β)"]
print(cmp.to_string)
print("Agreement rate:", f"{cmp['Agree'].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 and figures are cached by default under Data/macro_decade/.