带总值的不规则重叠时间序列按ID拆分无重叠最小时间区间

问题背景 #

现有如下格式的时间序列数据，存在不规则时间步长且区间可能重叠，其中value为对应区间的总值：

import pandas as pd
from io import StringIO

df = pd.read_csv(StringIO('''
id,start,end,value
1,2021-01-01,2021-03-31,1000
1,2021-01-01,2021-06-30,2000
1,2021-01-01,2021-12-31,5000
2,2021-01-01,2021-02-01,100
2,2021-02-02,2021-05-04,200
2,2021-01-01,2021-08-24,1000
'''))

需要按id将其转换为无重叠的最小时间区间，并扣除重叠部分的value，最终期望输出如下：

output = pd.read_csv(StringIO('''
id,start,end,value
1,2021-01-01,2021-03-31,1000
1,2021-04-01,2021-06-30,1000
1,2021-07-01,2021-12-31,3000
2,2021-01-01,2021-02-01,100
2,2021-02-02,2021-05-04,200
2,2021-05-05,2021-08-24,700
'''))

此前尝试适配相关日期区间重叠处理方案但未成功，寻求可行解决方法。

解决方案 #

核心思路：提取所有关键日期点分割出无重叠子区间，再针对每个子区间计算扣除重叠后的value（大总值减去已覆盖的小总值）。

import pandas as pd
from io import StringIO

# 读取原始数据
df = pd.read_csv(StringIO('''
id,start,end,value
1,2021-01-01,2021-03-31,1000
1,2021-01-01,2021-06-30,2000
1,2021-01-01,2021-12-31,5000
2,2021-01-01,2021-02-01,100
2,2021-02-02,2021-05-04,200
2,2021-01-01,2021-08-24,1000
'''))

# 转换日期格式为datetime，方便运算
df['start'] = pd.to_datetime(df['start'])
df['end'] = pd.to_datetime(df['end'])

# 处理单个id分组的函数
def process_id_group(group):
    # 提取所有关键日期：原始区间的起始日、结束日+1天（用于分割区间）
    dates = pd.concat([group['start'], group['end'] + pd.Timedelta(days=1)]).unique()
    # 排序日期
    dates = sorted(dates)
    # 生成无重叠的子区间
    intervals = []
    for i in range(len(dates)-1):
        interval_start = dates[i]
        interval_end = dates[i+1] - pd.Timedelta(days=1)
        intervals.append({'start': interval_start, 'end': interval_end})
    intervals_df = pd.DataFrame(intervals)
    
    # 计算每个子区间的扣除后value
    def calculate_interval_value(interval):
        # 找到所有包含当前子区间的原始区间
        mask = (group['start'] <= interval['start']) & (group['end'] >= interval['end'])
        matching_values = group.loc[mask, 'value'].sort_values(ascending=False).tolist()
        # 大总值减去次大值（扣除已覆盖部分），只有一个区间则直接取原值
        return matching_values[0] - (matching_values[1] if len(matching_values) > 1 else 0)
    
    intervals_df['value'] = intervals_df.apply(calculate_interval_value, axis=1)
    intervals_df['id'] = group['id'].iloc[0]
    # 调整列顺序匹配期望输出
    return intervals_df[['id', 'start', 'end', 'value']]

# 按id分组处理并合并结果
result = df.groupby('id').apply(process_id_group).reset_index(drop=True)
# 转换日期为字符串格式
result['start'] = result['start'].dt.strftime('%Y-%m-%d')
result['end'] = result['end'].dt.strftime('%Y-%m-%d')

print(result)

代码说明 #

1. 日期格式转换：将start和end转为datetime类型，支持日期加减运算。
2. 提取关键日期：收集原始区间的起始日和结束日+1天，这些日期是分割无重叠区间的节点。
3. 生成无重叠子区间：排序关键日期后，相邻日期组成子区间（结束日减1天避免区间重叠）。
4. 计算扣除后value：对每个子区间，找到所有包含它的原始区间，按value降序排列，用最大的总值减去次大值（扣除重叠部分已统计的value）；仅单个区间包含时直接取原值。
5. 结果合并：分组处理后合并数据，将日期转回字符串格式匹配期望输出。

运行代码后，输出结果与期望的output完全一致。

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