Python基金收益分析：用pandas计算累计收益与最大回撤

昨晚上十一点多，我还在公司楼下啃着个凉了的麻辣烫，准备回去收个尾，结果手机“叮”一下，同事在群里发问：“我这基金又绿了，想看下这几只这一年到底亏了多少，每次都得在几个 App 里来回点，烦死了……”

我当时人已经有点困糊了，但是程序员的职业病就是这样——只要一提到“重复点来点去”，脑子立马清醒：这活儿用 Python 搞一搞不就完了？

说下那晚的场景先。他的问题其实特别典型，你估计也遇到过：

• 手里有三五只甚至十几只基金
• 有的是支付宝买的，有的是蚂蚁，有的是券商 App
• 想看一个简单的问题：今年到底赚了还是亏了？哪只拖后腿最严重？

结果在手机上要来回切页面、选时间区间、自己脑子里算，还容易算错。我当时就跟他说：你要不干脆把这些基金当成普通数据，丢给 Python 帮你算，想看啥指标就自己加。

先搞清楚：到底想“追踪”什么？

我问：你到底想看啥？别说“看收益”这么笼统。他想了半天，说大概就这几件事：

• 每只基金从某一天（比如自己开始买入那天）到今天，累计涨跌多少
• 最近一段时间（比如近 30 天）大概是涨是跌
• 哪只回撤最狠，也就是中途跌得最惨的那一只

听上去挺专业，其实就三件事：总收益、阶段收益、最大回撤。这些在 Python 里就是几列数据和几个公式的问题，难点反而在“数据哪儿来”。

数据这块就别太纠结了

我说，先别想着一上来就什么自动爬取、对接券商开放平台，这种一搞就是半夜两点的事。先用最土但稳定的方式：

1. 去基金网站或者 App 上，把你每只基金的历史净值导出成 CSV
2. 简单处理成这种格式就行（csv 文件大概长这样）：

date,nav
2023-01-03,1.235
2023-01-04,1.242
2023-01-05,1.227
...

一只基金一个文件，比如：110022.csv、161725.csv 这种。以后你要是懒得手动导了，再说自动爬取的事。

用 pandas 把净值读进来

我回到工位，随手写了个特别简单的读文件函数：

import pandas as pd
from pathlib import Path

def load_fund_nav(csv_path: str) -> pd.DataFrame:
    """
    读取单只基金的净值数据，并按日期排序
    """
    df = pd.read_csv(csv_path)
    # 兜底一下列名，很多导出的文件可能是中文
    rename_map = {}
    for col in df.columns:
        if 'date' in col.lower() or '日期' in col:
            rename_map[col] = 'date'
        if 'nav' in col.lower() or '净值' in col:
            rename_map[col] = 'nav'
    df = df.rename(columns=rename_map)

    df['date'] = pd.to_datetime(df['date'])
    df = df.sort_values('date').reset_index(drop=True)

    df['nav'] = df['nav'].astype(float)
    return df

这段没啥花活，就是把日期、净值两列整理好，保证按时间升序排好。我当时还提醒一句：别小看“排序”这一步，时间序列一乱，后面所有收益率都是瞎算。

收益怎么算？就那几行代码

他最关心的是：如果当初每只基金各投 1 万，现在到底变成多少。我就写了一个小函数，专门算某只基金从第一天到现在的表现，中间还顺手算了下最大回撤：

def calc_fund_stats(df: pd.DataFrame, invest_amount: float = 10000):
    """
    基于历史净值，计算一些常用指标：
    - 累计收益率
    - 年化收益率（粗略）
    - 最大回撤
    - 当前市值
    """
    # 每天的日收益率
    df['ret'] = df['nav'].pct_change().fillna(0)

    # 从第一天净值倒推，假设那天买入 invest_amount 元
    first_nav = df['nav'].iloc[0]
    shares = invest_amount / first_nav
    df['value'] = df['nav'] * shares

    # 累计收益率：最后一天 vs 第一天
    total_return = df['value'].iloc[-1] / df['value'].iloc[0] - 1

    # 简单按“交易天数≈一年 365 天”算年化，别较真精准金融公式
    days = (df['date'].iloc[-1] - df['date'].iloc[0]).days
    if days > 0:
        years = days / 365
        annual_return = (1 + total_return) ** (1 / years) - 1
    else:
        annual_return = 0.0

    # 最大回撤：从历史高点回落的最大比例
    df['cum_max'] = df['value'].cummax()
    df['drawdown'] = df['value'] / df['cum_max'] - 1
    max_drawdown = df['drawdown'].min()

    return {
        'invest_amount': invest_amount,
        'current_value': df['value'].iloc[-1],
        'total_return': total_return,
        'annual_return': annual_return,
        'max_drawdown': max_drawdown
    }

这里面有几个点我特地说明了一嘴：

• shares = invest_amount / first_nav：就是你买了多少份额，后面全靠它。
• 最大回撤那块其实就是“一路走来最高点”和“现在的位置”的差值里找最惨的那次。
• 年化收益算得比较粗暴，但对于“心里有个数”完全够用，别拿去写研报就行。

把多只基金串起来，一跑就出结果

光算一只没意思呀，他手里有七八只，我就顺手把脚本凑成一个可以直接跑的小工具，他只要配置下“基金代码 + 文件路径”就行。

import pandas as pd

FUNDS = {
    "110022": "某消费混合",
    "161725": "某创业板指数",
    "005911": "某科技主题",
    # 想加自己继续往下写
}

def analyze_all_funds(base_dir: str = "./data", invest_each: float = 10000):
    results = []
    base = Path(base_dir)

    for code, name in FUNDS.items():
        csv_path = base / f"{code}.csv"
        if not csv_path.exists():
            print(f"[WARN] 找不到 {code} ({name}) 的数据文件：{csv_path}")
            continue

        df = load_fund_nav(str(csv_path))
        stats = calc_fund_stats(df, invest_each)
        stats['code'] = code
        stats['name'] = name
        results.append(stats)

    if not results:
        print("一个基金都没算成，先确认下 csv 文件有没有放对位置。")
        return

    res_df = pd.DataFrame(results)
    # 按总收益排序，看谁最拉胯
    res_df = res_df.sort_values('total_return', ascending=True)

    # 打印一个还算好看的表
    print("\n=== 基金收益汇总 ===")
    for _, row in res_df.iterrows():
        print(
            f"{row['code']} {row['name']:<10} "
            f"投入: {row['invest_amount']:.0f} "
            f"当前: {row['current_value']:.0f} "
            f"总收益: {row['total_return']*100:6.2f}% "
            f"年化: {row['annual_return']*100:6.2f}% "
            f"最大回撤: {row['max_drawdown']*100:6.2f}%"
        )

    # 组合整体情况也算一下
    total_invest = res_df['invest_amount'].sum()
    total_value = res_df['current_value'].sum()
    total_ret = total_value / total_invest - 1
    print("\n=== 整体组合 ===")
    print(f"总投入: {total_invest:.0f}  当前市值: {total_value:.0f}  总收益: {total_ret*100:.2f}%")

if __name__ == "__main__":
    analyze_all_funds("./data", invest_each=10000)

我一边敲一边吐槽：“你以后要换基金，只要把那只的 csv 换了再跑一遍，就比你在 App 里挨个点快多了。” 他跑完之后，第一句原话是：“卧槽，原来拖我后腿最狠的是这货啊，我天天还以为是另一只。”

就这玩意儿，把“感觉自己亏麻了”和“到底亏多少”这件事分开了。心态会好很多，至少数字是清楚的。

顺手加一个“最近 30 天”的小视角

聊着聊着，他又问：那我能不能只看最近 30 天，别从建仓那天算，我中间还加过仓呢。我懒得解释交易行为复杂这事，就先弄了个“简单近 30 天收益”，当个大概参考：

from datetime import timedelta

def recent_performance(df: pd.DataFrame, days: int = 30):
    """
    看最近 N 天的表现，简单从 N 天前的净值到今天
    """
    if df.empty:
        return 0.0

    end_date = df['date'].iloc[-1]
    start_date = end_date - timedelta(days=days)

    sub = df[df['date'] >= start_date]
    if len(sub) < 2:
        return 0.0

    start_nav = sub['nav'].iloc[0]
    end_nav = sub['nav'].iloc[-1]
    return end_nav / start_nav - 1

然后在上面那个汇总里多打一个字段出来就完事，其实就是多一列 % 让你看最近是在恢复还是继续躺平。

这些东西日后都能慢慢“进化”

那天晚上折腾完差不多快一点了，他已经在那边计划说要不要搞成一个每天自动跑、自动丢到企业微信的日报。我就跟他说：先别急，脚本能跑起来、数据结构搞清楚，比啥都重要，后面要想玩花的：

• 可以尝试自己去对接某些公开接口，不手动导 csv
• 可以把结果写进 SQLite 或者直接丢到一个网页上
• 可以在最大回撤那块再细化一下，比如统计某一段时间的回撤

就像之前写数据库性能测试那堆脚本，一开始也是瞎写，慢慢才变成工作里常用的工具。

最后啰嗦两句

这套东西说白了就干两件事：

• 把“基金”降维成“时间 + 数字”
• 把“感觉”换成“算出来的结果”

别把脚本当什么高大上的金融系统，它更多就是帮你少点几下手机、少骂几句“怎么又是绿的”。剩下买不买、怎么配仓，那是完全另外一件事了，也不能指望几行 Python 就替你做决策。

如果你对这类利用 pandas 进行数据处理和分析的实战项目感兴趣，欢迎到 云栈社区 的 Python 板块与更多开发者交流心得，获取更多实用脚本和数据分析思路。