如何使用Python Pandas生成时间序列连续Y指标分组的聚合表

前置预处理 #

首先完成基础字段校验和规则字段生成：

import pandas as pd
import numpy as np

# 1. 转换时间列为datetime格式，保证时间计算准确性
df['Time'] = pd.to_datetime(df['Time'])

# 2. 按规则生成Indicator标识列，若原始数据已有Indicator可跳过本步骤
df['Indicator'] = np.where(
    (df['Value'].isna()) | (df['Value']>2) | (df['Value'] < -2),
    'Y',
    'N'
)

步骤1：生成连续Y的分组ID #

# 按Key分组，标记组内连续相同Indicator的分组ID
df['group_id'] = (df['Indicator'] != df.groupby('Key')['Indicator'].shift(1)).cumsum()

# 预计算同Key下每行的下一行时间，用于后续end_time规则计算
df['next_time'] = df.groupby('Key')['Time'].shift(-1)

步骤2：过滤异常记录并聚合结果 #

# 仅保留Indicator为Y的记录参与聚合计算
df_y = df[df['Indicator'] == 'Y'].copy()

# 按Key和连续组ID分组，统计中间字段
df_s = df_y.groupby(['Key', 'group_id'], as_index=False).agg(
    start_time = ('Time', 'min'),
    group_last_time = ('Time', 'max'),
    group_last_next_time = ('next_time', 'last'),
    max_value = ('Value', 'max'),
    min_value = ('Value', 'min')
)

# 按规则计算end_time
df_s['end_time'] = np.where(
    df_s['group_last_next_time'].notna(),
    df_s['group_last_next_time'],
    df_s['group_last_time']
)

# 计算时间差，单位转换为分钟
df_s['duration(min)'] = (df_s['end_time'] - df_s['start_time']).dt.total_seconds() / 60

# 输出最终需要的字段
df_s = df_s[['Key', 'start_time', 'end_time', 'duration(min)', 'max_value', 'min_value']]

补充说明 #

• pandas内置的max/min计算默认自动忽略空值，直接符合空值不参与统计的要求
• 可根据业务需要对duration(min)增加.round(2)之类的处理，保留指定小数位
• 百万行以上的大数据量场景下，建议提前对Key和Time做排序，保证连续分组逻辑正确

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