

> 本文你将得到什么：
> 1. 三个组合层面的风险探针：滚动相关性恶化检测、权重集中度 HHI、组合回撤阈值预警
> 2. 一套"报警器自检"方法——阈值收紧一档，验证报警器真的会响
> 3. 一组实测数字：等权三腿组合 HHI 稳在 0.336，回撤最深 -13.72%，-20% 阈值下绿灯、收紧到 -12% 触发 3 天报警
> 4. 一张三面板风险仪表盘图（相关性 / 权重与 HHI / 回撤），可挂进你的监控脚本

---

## 痛点：组合不是建完就完事，风险会悄悄变脸

上一篇我们把再平衡的账算清楚了：不调仓，单腿会从 33% 漂到 40%。但漂移只是组合会"变质"的方式之一——还有两个更隐蔽的：

- **相关性恶化**：建仓时两个策略负相关、分散效果良好，半年后市场风格一变，它们同涨同跌，你的"分散"悄悄失效；
- **回撤越限**：组合回撤 -13% 时你觉得"还在容忍范围内"，等发现时已经 -25%——因为没人盯着，或者盯的人没有一条明确的报警线。

这篇把这三个探针做成一个最小风险仪表盘：每天跑一次，绿灯放行、红灯报警。

---

## 环境准备

```bash
pip install mairui pandas numpy matplotlib
```

---

## 第一步：拿到组合的收益流

口径与前三篇完全一致：三个策略各盯一只股票（600519 跑 MA5/20 双均线、000001 跑 MA3/10 快速均线、300750 跑 20 日动量），仓位由昨日收盘信号决定（shift 1，无前视），等权合成组合：

```python
from mairui import Client
import pandas as pd
import numpy as np

client = Client("LICENCE-66D8-9F96-0C7F0FBCD073")  # 演示证书，请换成你自己的正式证书

UNIVERSE = [
    ("600519", "sma520", "MA5/20动量"),
    ("000001", "sma310", "MA3/10快速"),
    ("300750", "mom20", "20日动量"),
]

def strategy_daily_returns(df, kind):
    c = df["c"]
    if kind == "sma520":
        sig = (c.rolling(5).mean() > c.rolling(20).mean()).astype(float)
    elif kind == "sma310":
        sig = (c.rolling(3).mean() > c.rolling(10).mean()).astype(float)
    elif kind == "mom20":
        sig = (c / c.shift(20) - 1 > 0).astype(float)
    pos = sig.shift(1).fillna(0.0)          # 昨收信号 → 今仓，无前视
    ret = c.pct_change().fillna(0.0)
    return pos * ret

rets = {}
for code, kind, label in UNIVERSE:
    bars = client.stock_history(code=code, period="d", dividend="f")
    df = pd.DataFrame(bars).rename(columns={"t": "date"}).sort_values("date").reset_index(drop=True)
    df = df.tail(50).reset_index(drop=True)   # 本文统一取最近 50 个交易日做演示窗口
    for col in ("o", "h", "l", "c", "v"):
        df[col] = df[col].astype(float)
    rets[label] = strategy_daily_returns(df, kind)

R = pd.DataFrame(rets)
port = R.mean(axis=1)   # 等权组合（每日再平衡口径）
```

说明：这里用的是公开演示证书，为演示方便明文写入；实际项目请换成你自己的正式证书（申请配置见 SDK 文档 https://github.com/MaiRuiApi/mairui ），换证后代码无需修改。

---

## 第二步：探针一——滚动相关性恶化检测

窗口 20 个交易日，滚动计算每对策略的相关系数，超过 0.7 报警（0.7 以上基本等于"换名字的同一个策略"）：

```python
CORR_WIN, CORR_ALERT = 20, 0.7
pairs = [("MA5/20动量", "MA3/10快速"), ("MA5/20动量", "20日动量"), ("MA3/10快速", "20日动量")]

for a, b in pairs:
    rc = R[a].rolling(CORR_WIN).corr(R[b])
    last, peak = float(rc.iloc[-1]), float(rc.max())
    print(f"{a} × {b}: 最新滚动相关 {last:+.3f}  峰值 {peak:+.3f}  "
          f"{'⚠️ 超阈值' if last > CORR_ALERT else '正常'}")
```

**输出示例**（演示证样例数据，仅供演示）：

```
MA5/20动量 × MA3/10快速: 最新滚动相关 +0.049  峰值 +0.049  正常
MA5/20动量 × 20日动量: 最新滚动相关 +0.214  峰值 +0.458  正常
MA3/10快速 × 20日动量: 最新滚动相关 -0.121  峰值 +0.120  正常
```

三个组合的最新相关性都在 0.25 以下，远低于 0.7 报警线——分散结构当前健康。峰值 +0.458 也提示：相关性是波动的，要滚动看，不能建仓时算一次管一年。

---

## 第三步：探针二——权重漂移与集中度 HHI

不再平衡时各腿独立漂移。集中度用 **HHI（赫芬达尔指数）** 度量：各案权重的平方和。等权三腿的基线是 1/3 ≈ 0.333，腿数越少、权重越偏，HHI 越高：

```python
HHI_ALERT = 0.45

vals = np.full(3, 1.0 / 3.0)
hhi_track, w_track = [], []
for t in range(len(R)):
    if t > 0:
        vals = vals * (1 + R.iloc[t].values)   # 各腿独立漂移
    w = vals / vals.sum()
    w_track.append(w.copy())
    hhi_track.append(float((w ** 2).sum()))
hhi = pd.Series(hhi_track)
w_end = w_track[-1]

print("期末权重:", dict(zip(R.columns, np.round(w_end, 3))))
print(f"期末 HHI = {hhi.iloc[-1]:.3f}  峰值 HHI = {hhi.max():.3f}  "
      f"{'⚠️ 超阈值' if hhi.iloc[-1] > HHI_ALERT else '正常'}")
```

**输出示例**：

```
期末权重: {'MA5/20动量': 0.363, 'MA3/10快速': 0.344, '20日动量': 0.293}
期末 HHI = 0.336  峰值 HHI = 0.336  正常
```

50 个交易日内三腿漂移温和（0.333 → 0.363 / 0.344 / 0.293），HHI 稳在 0.336，未触发 0.45 报警线。如果换五腿等权，基线只有 0.2——**HHI 的阈值要随腿数设定**，不能一个数字用到底。

---

## 第四步：探针三——回撤阈值预警 + 报警器自检

组合回撤按净值曲线相对历史高点的回吐计算，超过阈值（本例 -20%）报警：

```python
DD_ALERT = -0.20

eq = (1 + port).cumprod()
dd = eq / eq.cummax() - 1
print(f"最新回撤 = {dd.iloc[-1]:+.2%}  最深回撤 = {dd.min():+.2%}")
n_alert = int((dd < DD_ALERT).sum())
print(f"阈值 {DD_ALERT:+.0%}: 触发 {n_alert} 天 {'⚠️' if n_alert > 0 else '（未触发）'}")
```

**输出示例**：

```
最新回撤 = -13.72%  最深回撤 = -13.72%
阈值 -20%: 触发 0 天 （未触发）
```

绿灯。但**绿灯不等于报警器是好的**——一个从没响过的报警器，你不知道它是真没险情、还是根本不会响。所以加一步灵敏度自检：把阈值收紧一档再看：

```python
for th in (-0.20, -0.12):
    n = int((dd < th).sum())
    print(f"阈值 {th:+.0%}: 触发 {n} 天 {'⚠️' if n > 0 else '（未触发）'}")
```

**输出示例**：

```
阈值 -20%: 触发 0 天 （未触发）
阈值 -12%: 触发 3 天 ⚠️
```

阈值收紧到 -12%，报警器触发 3 天——机制验证通过。阈值本身没有对错，它表达的是你的风险预算：设 -20% 意味着你愿意承受两成的回吐才动手，设 -12% 则更早介入、也可能更早被震出去。

---

## 第五步：把三个探针画成一张仪表盘

```python
import matplotlib.pyplot as plt
plt.rcParams["font.sans-serif"] = ["SimHei", "Microsoft YaHei"]
plt.rcParams["axes.unicode_minus"] = False

fig, axes = plt.subplots(3, 1, figsize=(12, 10), sharex=True)

for a, b in pairs:
    axes[0].plot(R[a].rolling(CORR_WIN).corr(R[b]).values, label=f"{a}×{b}")
axes[0].axhline(CORR_ALERT, color="red", linestyle="--", label=f"报警线 {CORR_ALERT}")
axes[0].set_title("滚动 20 日策略相关性（恶化检测）")
axes[0].legend(fontsize=8)

for i, c in enumerate(R.columns):
    axes[1].plot([w[i] for w in w_track], label=c)
axes[1].plot(hhi.values, color="black", linestyle=":", linewidth=2, label="HHI 集中度")
axes[1].axhline(HHI_ALERT, color="red", linestyle="--", label=f"HHI 报警线 {HHI_ALERT}")
axes[1].set_title("权重漂移与集中度 HHI（不再平衡）")
axes[1].legend(fontsize=8)

axes[2].plot(dd.values, color="darkred")
axes[2].axhline(DD_ALERT, color="red", linestyle="--", label=f"回撤报警线 {DD_ALERT:.0%}")
axes[2].set_title("组合回撤与阈值预警")

plt.tight_layout()
plt.savefig("risk_dashboard.png", dpi=150)
```

三面板从上到下：相关性会不会恶化、权重往哪漂、回撤离报警线多远——组合的健康状况一屏看完。

---

## 结果怎么读（诚实的部分）

1. **三项全绿是本次数据的真实结果，不是演示失败**：样例窗口内相关性温和、漂移有限、回撤 -13.72% 没碰到 -20% 线。仪表盘的价值不在每次都报警，而在**险情出现的第一天就让红灯亮起来**；
2. **报警器要自检**：阈值 -20% 从不触发，是数据温和还是逻辑写错？收紧到 -12% 触发 3 天，证明机制正常——任何监控系统上线前都值得做一次这种"人造险情"验证；
3. **HHI 阈值随腿数变**：三腿等权基线 0.333，五腿 0.2，十腿 0.1。用同一个 0.45 阈值管不同腿数的组合，等于对小组合宽容、对大组合苛刻；
4. 数字为演示证样例数据结果，仅演示方法与阈值口径；换正式证书即得真实行情下的仪表盘。

---

## 常见坑

1. **阈值拍脑袋**：回撤报警线应该从风险预算倒推（你最多能接受多少回吐），而不是抄别人的参数。设太松等于没有，设太紧天天被震出去；
2. **相关窗口太短**：10 日窗口的相关系数噪声极大，今天正明天负；20 日是常见起点，样本越长越稳，但对"突然恶化"的反应也越慢——窗口长度本身是灵敏度与稳定性的权衡；
3. **只看期末值**：漂移和相关恶化都是渐进过程，本文所有探针都输出**整条时间序列**而非单点，峰值线（如相关峰值 +0.458）比最新值信息量更大；
4. **报警了但没人处理**：阈值预警只是触发器，触发后的动作（减仓、再平衡、暂停策略）要事先写好，否则仪表盘只是焦虑生成器。

---

## 小结与下篇预告

- 组合的三个隐蔽变质方式——相关性恶化、权重漂移、回撤失控——各配一个探针：滚动相关 > 0.7 报警、HHI 超阈值报警、回撤穿线报警；
- 本例实测：三项全绿，但阈值收紧到 -12% 触发 3 天，报警器机制验证通过；
- 仪表盘之外，组合还有一层"人事问题"：策略本身也会过气。表现垫底的策略该不该踢出组合？下一篇用滚动夏普排名做一套自动入选/淘汰的轮动机制。

---

## 免责声明

本文所有内容仅用于技术学习与交流，不构成任何投资建议。股市有风险，投资需谨慎。


一组实测数字：等权三腿组合 HHI 稳在 0.336，回撤最深 -13.72%，-20% 阈值下绿灯、收紧到 -12% 触发 3 天报警

【Python量化组合管理实战 04】组合的风险仪表盘：相关性恶化、集中度与回撤预警

人工智能

三探针风险监控框架：滚动相关性>0.7报警、HHI权重集中度预警、回撤阈值触发。实测三腿组合HHI稳定0.336，回撤-13.72%未破-20%线；阈值收紧至-12%触发3天，验证报警机制有效。配套三面板仪表盘，每日扫描组合健康度，及时发现相关性恶化与漂移风险。

代码解释

GitHub

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

【Python量化组合管理实战 04】组合的风险仪表盘：相关性恶化、集中度与回撤预警-腾讯云开发者社区-腾讯云

【Python量化组合管理实战 04】组合的风险仪表盘：相关性恶化、集中度与回撤预警

【Python量化组合管理实战 04】组合的风险仪表盘：相关性恶化、集中度与回撤预警

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐