主流指数最大连续回撤时长与亏损幅度测算工具（量化持有亏损极限）｜教学级指数投资原型
内容包含免责声明和风险提示，不荐基、不预测底部、不引导开户、无任何引流。
一、实际应用场景描述
在智能证券投资课程中，"指数基金不会大跌套牢"是新手投资者最常见的认知偏差之一。
本程序适用于：
- 高校量化投资、基金投资课程实验
- 投资者教育（Investor Education）
- 指数基金定投入门教学
- 风险认知与最大回撤（Max Drawdown）概念演示
核心目标：
- 读取主流指数历史净值 / 价格数据
- 测算最大连续回撤时长与最大亏损幅度
- 量化指数投资可能面临的持有亏损极限
- 用数据回答："买指数基金真的不会深度套牢吗？"
✅ 不做未来预测
✅ 不构成投资建议
✅ 仅作为历史数据统计教学工具
二、痛点引入（真实可感知）
痛点 表现
"指数不会套牢" 2008 年沪深 300 最大回撤 −72%
只看年化收益 忽略极端风险
缺乏极限认知 "定投就行"但不知道最坏情况
回撤时长不可见 不知可能连续跌多久
工具门槛高 专业风险分析平台复杂
👉 需要一个轻量、本地、可解释、可复现的指数回撤分析工具
三、核心逻辑讲解（工程视角）
1️⃣ 数据模型设计
IndexDrawdownSession
├── index_name 指数名称
├── nav_list 净值 / 价格序列
└── drawdown_periods 回撤区间列表
2️⃣ 核心概念定义（教学用）
概念 定义
回撤（Drawdown） 从前期高点到当前低点的跌幅
最大回撤（Max Drawdown） 历史所有回撤中的最大值
连续回撤时长 从高点跌落到最低点经历的交易日数
恢复时长 从最低点回到前期高点所需交易日数
3️⃣ 回撤检测算法
核心思路（逐点扫描法）：
初始化：peak = nav[0], trough = nav[0]
遍历每个净值点：
如果 nav[i] > peak：
peak = nav[i]（创新高）
trough = nav[i]
如果 nav[i] < trough：
trough = nav[i]（继续探底）
当前回撤 = (trough − peak) / peak × 100%
更新全局最大回撤
4️⃣ 关键公式
单点回撤：
Drawdown(t) = (Nav(t) − Peak(t)) / Peak(t) × 100%
最大回撤：
Max Drawdown = min(Drawdown(t)) （取最小值，为负值）
回撤时长：
回撤时长 = 最低点索引 − 最高点索引（交易日数）
恢复时长：
恢复时长 = 回升至前高索引 − 最低点索引（交易日数）
5️⃣ 输出内容
指标 含义
最大回撤幅度 历史上最深的跌幅
回撤持续交易日 从顶部到底部用了多久
恢复交易日 从底部回到前高用了多久
回撤区间 具体起止日期
当前是否处于回撤中 风险提示
四、Python 模块化代码（可直接运行）
📁 项目结构
index_max_drawdown_analyzer/
│
├── main.py
├── models.py
├── analyzer.py
├── reporter.py
├── storage.py
├── README.md
└── DISCLAIMER.md
✅ models.py（数据建模）
"""
models.py
指数回撤分析数据模型
"""
class NavPoint:
"""单日净值 / 价格数据点"""
def __init__(self, date, nav):
self.date = date # "YYYY-MM-DD"
self.nav = nav
class IndexDrawdownSession:
"""指数回撤分析场景"""
def __init__(self, index_name, nav_points):
"""
index_name: 指数名称
nav_points: NavPoint 列表，按时间升序排列
"""
self.index_name = index_name
self.nav_points = nav_points
✅ analyzer.py（核心回撤检测引擎）
"""
analyzer.py
主流指数最大连续回撤时长与亏损幅度测算
"""
def find_max_drawdown(session):
"""
逐点扫描法检测最大回撤
返回所有回撤区间的列表
"""
navs = session.nav_points
n = len(navs)
if n < 2:
return []
# 存储所有回撤区间
drawdowns = []
# 当前回撤的起始点
peak_idx = 0
peak_nav = navs[0].nav
i = 1
while i < n:
current_nav = navs[i].nav
# 创新高：结束当前回撤区间，开启新区间
if current_nav > peak_nav:
# 记录之前的回撤
# 寻找当前区间的最低点
trough_idx, trough_nav = _find_trough(navs, peak_idx, i - 1)
if trough_nav < peak_nav:
dd_pct = (trough_nav - peak_nav) / peak_nav * 100
recovery_idx = _find_recovery(navs, trough_idx, trough_nav, n)
recovery_days = (recovery_idx - trough_idx) if recovery_idx else None
drawdowns.append({
"peak_date": navs[peak_idx].date,
"trough_date": navs[trough_idx].date,
"recovery_date": navs[recovery_idx].date if recovery_idx else "尚未恢复",
"peak_nav": peak_nav,
"trough_nav": trough_nav,
"drawdown_pct": round(dd_pct, 2),
"decline_days": trough_idx - peak_idx,
"recovery_days": recovery_days
})
# 重置
peak_idx = i
peak_nav = current_nav
i += 1
# 处理最后一个回撤区间
trough_idx, trough_nav = _find_trough(navs, peak_idx, n - 1)
if trough_nav < peak_nav:
dd_pct = (trough_nav - peak_nav) / peak_nav * 100
recovery_idx = _find_recovery(navs, trough_idx, trough_nav, n)
recovery_days = (recovery_idx - trough_idx) if recovery_idx else None
drawdowns.append({
"peak_date": navs[peak_idx].date,
"trough_date": navs[trough_idx].date,
"recovery_date": navs[recovery_idx].date if recovery_idx else "尚未恢复",
"peak_nav": peak_nav,
"trough_nav": trough_nav,
"drawdown_pct": round(dd_pct, 2),
"decline_days": trough_idx - peak_idx,
"recovery_days": recovery_days
})
# 按回撤幅度排序（从小到大，即最深的在前）
drawdowns.sort(key=lambda x: x["drawdown_pct"])
return drawdowns
def _find_trough(navs, start, end):
"""在 [start, end] 区间找最低点"""
min_idx = start
min_nav = navs[start].nav
for i in range(start, end + 1):
if navs[i].nav < min_nav:
min_nav = navs[i].nav
min_idx = i
return min_idx, min_nav
def _find_recovery(navs, start, trough_nav, total_len):
"""从 start 开始，找到第一个 >= trough_nav 的点"""
for i in range(start, total_len):
if navs[i].nav >= trough_nav:
return i
return None
def summarize(drawdowns):
"""汇总统计"""
if not drawdowns:
return None
max_dd = drawdowns[0] # 最深回撤
# 平均回撤
avg_dd = sum(abs(d["drawdown_pct"]) for d in drawdowns) / len(drawdowns)
# 平均恢复时长（排除尚未恢复的）
recovered = [d for d in drawdowns if d["recovery_days"] is not None]
avg_recovery = sum(d["recovery_days"] for d in recovered) / len(recovered) if recovered else None
return {
"total_drawdowns": len(drawdowns),
"max_drawdown_pct": max_dd["drawdown_pct"],
"max_drawdown_peak": max_dd["peak_date"],
"max_drawdown_trough": max_dd["trough_date"],
"max_drawdown_decline_days": max_dd["decline_days"],
"max_drawdown_recovery": max_dd["recovery_date"],
"max_drawdown_recovery_days": max_dd["recovery_days"],
"avg_drawdown_pct": round(avg_dd, 2),
"avg_recovery_days": round(avg_recovery, 2) if avg_recovery else "N/A",
"still_in_drawdown": any(d["recovery_days"] is None for d in drawdowns)
}
✅ reporter.py（回撤分析报告）
"""
reporter.py
指数最大回撤分析报告
"""
def report(session, drawdowns, summary):
print("\n" + "=" * 65)
print("【主流指数最大连续回撤分析报告】")
print("=" * 65)
print(f"指数：{session.index_name}")
print(f"数据点数：{len(session.nav_points)} 个")
print("-" * 65)
if not drawdowns:
print("✅ 该区间内无回撤记录")
return
print(f"\n📊 回撤总览：共 {summary['total_drawdowns']} 个回撤区间")
print(f" 平均回撤幅度：{summary['avg_drawdown_pct']}%")
print(f"\n🔴 最大回撤（持有亏损极限）：")
print(f" 回撤幅度：{summary['max_drawdown_pct']}%")
print(f" 高点日期：{summary['max_drawdown_peak']}")
print(f" 低点日期：{summary['max_drawdown_trough']}")
print(f" 连续下跌：{summary['max_drawdown_decline_days']} 个交易日")
print(f" 恢复日期：{summary['max_drawdown_recovery']}")
if summary['max_drawdown_recovery_days'] is not None:
print(f" 恢复用时：{summary['max_drawdown_recovery_days']} 个交易日")
else:
print(f" ⚠️ 截至目前尚未恢复至前高！")
if summary['avg_recovery_days'] != "N/A":
print(f"\n📈 平均恢复时长：{summary['avg_recovery_days']} 个交易日")
# 列出所有回撤区间（Top 5 最深）
print(f"\n📋 最深回撤 Top 5：")
print("-" * 65)
for i, dd in enumerate(drawdowns[:5], 1):
rec = f"{dd['recovery_days']} 天" if dd['recovery_days'] else "未恢复"
print(f" {i}. {dd['peak_date']} → {dd['trough_date']}")
print(f" 回撤：{dd['drawdown_pct']}% | 下跌：{dd['decline_days']} 天 | 恢复：{rec}")
print("\n" + "=" * 65)
# 教学结论
print(f"\n💡 教学启示：")
print("-" * 65)
max_dd = abs(summary['max_drawdown_pct'])
if max_dd > 50:
print(f" ⚠️ 最大回撤高达 {max_dd}%，意味着：")
print(f" 即使是指数基金，满仓持有也可能亏损过半！")
elif max_dd > 30:
print(f" ⚠️ 最大回撤 {max_dd}%，指数基金并非"不会套牢"")
else:
print(f" ✅ 最大回撤 {max_dd}%，相对温和")
if summary['still_in_drawdown']:
print(f"\n ⚠️ 当前仍处于回撤中，尚未恢复至前高")
print(f" → 即使是"买入并持有"策略，也可能面临长期浮亏")
print(f"\n 核心结论：")
print(f" 指数基金 ≠ 不会套牢，只是相对个股回撤更可控。")
print(f" 了解最大回撤，才能做好资金规划与心理准备。")
print("=" * 65)
✅ storage.py（本地存储）
"""
storage.py
JSON 本地存储
"""
import json
FILE_PATH = "index_drawdown_analysis.json"
def save_result(data):
with open(FILE_PATH, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2)
✅ main.py（交互入口）
"""
main.py
主流指数最大回撤测算工具
"""
from models import NavPoint, IndexDrawdownSession
from analyzer import find_max_drawdown, summarize
from reporter import report
from storage import save_result
def main():
print("=== 主流指数最大回撤测算工具（教学版）===")
print("量化「指数基金不会大跌套牢」是否成立\n")
index_name = input("指数名称（如 沪深300）：")
print(f"\n📌 请输入净值 / 价格数据（日期 净值，空行结束）：")
print(f"格式示例：2007-10-16 5758.89")
nav_points = []
while True:
line = input()
if not line:
break
parts = line.split()
date = parts[0]
nav = float(parts[1])
nav_points.append(NavPoint(date, nav))
if len(nav_points) < 2:
print("⚠️ 数据不足，无法分析")
return
session = IndexDrawdownSession(index_name, nav_points)
# 检测回撤
drawdowns = find_max_drawdown(session)
# 汇总
summary = summarize(drawdowns)
if summary is None:
print("⚠️ 无回撤数据")
return
# 输出报告
report(session, drawdowns, summary)
# 保存结果
result_data = {
"index": session.index_name,
"data_points": len(session.nav_points),
"summary": summary,
"all_drawdowns": drawdowns
}
save_result(result_data)
print("\n✅ 回撤分析结果已保存")
if __name__ == "__main__":
main()
五、README 与使用说明
# 主流指数最大回撤测算工具（教学版）
## 项目说明
测算主流指数最大连续回撤时长与亏损幅度，量化指数基金持有的亏损极限。
## 使用方式
```bash
python main.py
```
## 输入示例
```
指数名称：沪深300
2007-10-16 5758.89
2007-11-01 5338.27
2008-01-15 4800.00
2008-06-30 3100.00
2008-10-28 1714.82
2009-03-01 2100.00
2009-08-01 3500.00
2010-01-04 3700.00
（空行结束）
```
## 核心指标说明
| 指标 | 含义 |
|---|---|
| 最大回撤幅度 | 历史最深跌幅（%） |
| 连续回撤时长 | 从顶部到底部经历的交易天数 |
| 恢复时长 | 从底部回到前高所需交易天数 |
| 是否仍在回撤中 | 当前是否已恢复至前高 |
## 适用范围
- 量化投资课程
- 指数基金定投教学
- 风险管理与回撤认知训练
## 注意事项
- 仅基于历史数据
- 不构成任何投资建议
- 使用前请阅读 DISCLAIMER.md
六、DISCLAIMER.md（免责声明与风险提示）
# 免责声明与风险提示
## 免责声明
本程序仅供**教学与科研用途**，用于演示最大回撤（Max Drawdown）的计算方法。
作者不提供任何投资建议，不推荐任何指数或基金，不承诺任何收益。
## 风险提示
1. 历史最大回撤不代表未来最大回撤，极端行情可能突破历史极值
2. 恢复时长基于历史数据，未来恢复可能更慢甚至永久无法恢复
3. 指数退市、规则调整等因素未被纳入考虑
4. "最大回撤 −72%"意味着极端情况下满仓可能亏损超七成
5. 回撤分析仅为风险管理工具，不构成买卖依据
6. 定投可摊薄成本，但无法消除回撤风险
使用本工具产生的任何后果，作者概不负责。
七、核心知识点卡片（教学向）
分类 内容
Python 类、列表遍历、排序、异常处理
量化金融 最大回撤（Max Drawdown）、恢复时长
指数投资 指数基金同样存在深度回撤风险
风险管理 回撤时长 ≠ 恢复时长，两者都需关注
数据分析 逐点扫描算法、极值检测
工程思想 模块化、算法与展示解耦
可扩展性 可接入真实指数 API、支持可视化绘图
八、总结（工程师视角）
这是一个完全中立、去营销化、可教学的原型系统：
✅ 不鼓吹指数基金
✅ 不制造恐慌
✅ 不伪装成择时或预测工具
它真正展示的是：
如何用 Python 把"指数不会套牢"这个模糊信念，变成可量化、可验证、可反思的数据事实
核心教学价值：
传统观念 历史数据揭示的真相
"指数基金不会套牢" 沪深 300 历史最大回撤 −72%，恒生指数 −66%
"长期持有就没事" 2007 年高点买入沪深 300，7 年后才回本
"定投就安全" 定投能摊薄成本，但回撤期间仍会浮亏 30%+
"指数波动小" 单日 −7%、连续数月下跌均为历史事实
关键数字（教学参考）：
指数 历史最大回撤 回撤时长 恢复时长
沪深 300 −72%（2008） ~12 个月 ~7 年
中证 500 −65%（2008） ~12 个月 ~5 年
恒生指数 −66%（2008） ~17 个月 ~10 年
⚠️ 以上为历史数据，不代表未来表现。实际投资需结合个人风险承受能力。
本文代码仅供学习与技术交流，不构成任何投资建议，股市有风险，入市需谨慎！
利用AI解决实际问题，如果你觉得这个工具好用，欢迎关注长安牧笛！
原理与配置实战：基于SGI中断与共享内存的无锁通信)
