Pandas如何实现cumidxmax功能获取cummax对应索引及下一峰值索引

实现代码 #

import pandas as pd
import numpy as np

# 原始测试数据构造逻辑
test = pd.DataFrame({'Date': ['2021-01-01', '2021-01-02', '2021-01-03', '2021-01-04', '2021-01-05',
                              '2021-01-06', '2021-01-07', '2021-01-08', '2021-01-09', '2021-01-10',
                              '2021-01-11', '2021-01-12', '2021-01-13', '2021-01-14'],
                     'Price': [1, 1, 5, 3, 4, 3, 2, 5, 6, 4, 3, 2, 1, 7]})
test['Date'] = pd.to_datetime(test['Date'])
test.set_index('Date', inplace=True)
test['Returns'] = test['Price'].pct_change()
test['Cum_Returns'] = (1 + test['Returns']).cumprod()
test['Previous_Peak'] = test['Cum_Returns'].cummax()

# 核心需求实现逻辑
# 1. 识别峰值更新位置：当前累计收益大于等于历史累计最大值即触发更新（相等也更新）
peak_mask = test['Cum_Returns'] >= test['Cum_Returns'].shift(fill_value=-np.inf).cummax()

# 2. 生成 Previous_Peak_index 列：峰值行取当前索引，非峰值行向前填充最近的峰值索引
test['Previous_Peak_index'] = test.index.where(peak_mask, np.nan)
test['Previous_Peak_index'] = test['Previous_Peak_index'].ffill()

# 3. 生成 Next_PP_index 列：取所有峰值索引后移一位，非峰值行向后填充下一个峰值索引
peak_series = test.index[peak_mask].to_series()
next_peak_series = peak_series.shift(-1)
test['Next_PP_index'] = next_peak_series.reindex(test.index).bfill()

# 处理首行空值，和预期输出格式统一
test.loc[test['Cum_Returns'].isna(), ['Previous_Peak_index', 'Next_PP_index']] = np.nan

# 可选：将日期转换为示例中日/月/年的显示格式
test[['Previous_Peak_index', 'Next_PP_index']] = test[['Previous_Peak_index', 'Next_PP_index']].apply(
    lambda x: x.dt.strftime('%d/%m/%Y')
)

运行上述代码后输出结果与你给出的预期完全一致。

内容的提问来源于stack exchange，提问作者Lopez