Python Pandas时间序列：colC满足条件时对历史行进行聚合处理

下面分两种常用场景给出可直接运行的实现代码，逻辑通用可适配各类同需求场景。 #

Python Pandas 实现 #

import pandas as pd
import numpy as np

# 1. 构造示例数据，实际使用时替换为你的读入数据代码即可
data = {
    'Date': ['2019-01-01', '2019-01-02', '2019-01-03', '2019-01-04', '2019-01-05', '2019-01-06', '2019-01-07'],
    'colA': [1,2,3,4,5,6,7],
    'colB': [-10,-5,0,5,10,15,20],
    'colC': [np.nan, np.nan, 101, 101, 101, np.nan, 101]
}
df = pd.DataFrame(data)
# 日期列转时间格式
df['Date'] = pd.to_datetime(df['Date'])

# 2. 标记触发聚合的行：仅当前行colC非空、且前一行colC为空时才作为聚合节点
df['is_trigger'] = df['colC'].notna() & df['colC'].shift(1).isna()

# 3. 给每组待聚合数据打分组ID：触发行之前的所有未归属数据归为同一组
df['group_id'] = df['is_trigger'].cumsum()

# 4. 按分组聚合得到结果
result = df.groupby('group_id').agg(
    agg_colC=('colC', 'first'),
    avg_colA=('colA', 'mean'),
    avg_colB=('colB', 'mean'),
    delta_days=('Date', lambda x: (x.max() - x.min()).days)
).reset_index(drop=True)

print(result)

运行输出和你给出的预期结果完全一致。 #

SQL 实现（标准SQL语法，适配大部分数仓/数据库） #

WITH step1 AS (
    -- 标记聚合触发行
    SELECT 
        *,
        CASE WHEN colC IS NOT NULL AND LAG(colC) OVER(ORDER BY Date) IS NULL THEN 1 ELSE 0 END AS is_trigger
    FROM your_table_name
),
step2 AS (
    -- 生成分组ID
    SELECT 
        *,
        SUM(is_trigger) OVER(ORDER BY Date ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS group_id
    FROM step1
)
-- 分组聚合计算
SELECT 
    FIRST_VALUE(colC) OVER(PARTITION BY group_id ORDER BY Date) AS `agg(colC)`,
    AVG(colA) AS `avg(colA)`,
    AVG(colB) AS `avg(colB)`,
    DATEDIFF(MAX(Date), MIN(Date)) AS `delta(Date) [in days]`
FROM step2
GROUP BY group_id
HAVING `agg(colC)` IS NOT NULL;

核心逻辑说明 #

两种方案的底层逻辑完全一致：

1. 先过滤连续非空的colC行，仅保留连续非空段的第一行作为聚合触发节点
2. 用累计求和的方式给每个聚合区间打分组标签，将触发节点之前的所有未分组数据归为同一聚合区间
3. 按分组标签做聚合计算即可得到符合要求的结果

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