本文介绍如何基于时间序列连续性筛选 dataframe 中满足最小连续月份数要求的 unique_id，核心是将日期转为月度周期、识别连续段并统计长度，最终保留符合阈值的 id 对应全部记录。
本文介绍如何基于时间序列连续性筛选 dataframe 中满足最小连续月份数要求的 unique_id，核心是将日期转为月度周期、识别连续段并统计长度，最终保留符合阈值的 id 对应全部记录。
在时间序列建模（如预测任务）中，常需确保每个 unique_id（例如不同产品、用户或门店）具备足够长且连续的历史观测——尤其关注从最新月份向前回溯的连续性。若仅统计总记录数而忽略时间断点，可能导致模型训练数据质量下降。本文提供一种健壮、可扩展的 Pandas 实现方案。
核心思路解析
关键在于：将日期转换为 Period('M') 类型 → 按 unique_id 分组 → 逆序排列后计算相邻月份差值 → 利用 MonthEnd() 判断是否连续 → 使用 cumsum() 标记连续段 → 统计每段长度并判断是否 ≥ 阈值。
✅ 连续性判定逻辑：ds.diff().ne(pd.offsets.MonthEnd())
当两行日期间隔恰好为一个自然月末（即 2024-01-31 → 2024-02-29），diff() 结果等于 MonthEnd()，此时 .ne(...) 返回 False，表示“连续”；反之为 True，表示“中断”。通过 .cumsum() 可对每个连续段分配唯一标识。
完整实现代码
import pandas as pd
from datetime import datetime
# 构造示例数据
data = {
'ds': ['2024-02-01', '2024-01-01', '2023-12-01', '2024-02-01', '2023-12-01'],
'y': [500, 600, 700, 800, 500],
'unique_id': [1, 1, 1, 2, 2]
}
df = pd.DataFrame(data)
df['ds'] = pd.to_datetime(df['ds']) # 确保日期为 datetime 类型
# 设置最小连续月份数
N = 2
def has_min_consecutive_months(group: pd.Series, min_count: int) -> bool:
"""
判断某 unique_id 的月份序列是否存在至少 min_count 个连续月份（从最新月向前追溯）
"""
# 转为月度周期，并按时间降序排列（最新月在前）
periods = group.dt.to_period('M').sort_values(ascending=False)
# 计算逆序差值：判断是否为连续月末
is_not_consecutive = periods.diff().ne(pd.offsets.MonthEnd())
# cumsum 标记连续段（每次中断则新段开始）
segment_id = is_not_consecutive.cumsum()
# 统计每段长度，检查是否有段 ≥ min_count
return segment_id.value_counts().ge(min_count).any()
# 按 unique_id 分组，应用判定函数
valid_ids = df.groupby('unique_id')['ds'].apply(has_min_consecutive_months, min_count=N)
selected_ids = valid_ids[valid_ids].index
# 筛选原始 DataFrame 中属于有效 unique_id 的所有行
result_df = df[df['unique_id'].isin(selected_ids)].reset_index(drop=True)
print(result_df)输出结果：
ds y unique_id 0 2024-02-01 500 1 1 2024-01-01 600 1 2 2023-12-01 700 1
进阶用法：获取每条记录的连续段长度
若需进一步分析（如过滤最长连续段、标记断点位置），可使用 transform 方式为每行添加 consecutive_stretch 字段：
def get_consecutive_length(group: pd.Series) -> pd.Series:
periods = group.dt.to_period('M').sort_values(ascending=False)
segment_id = periods.diff().ne(pd.offsets.MonthEnd()).cumsum()
return segment_id.map(segment_id.value_counts())
df['consecutive_stretch'] = df.groupby('unique_id')['ds'].transform(get_consecutive_length)
print(df[['ds', 'y', 'unique_id', 'consecutive_stretch']])输出：
ds y unique_id consecutive_stretch 0 2024-02-01 500 1 3 1 2024-01-01 600 1 3 2 2023-12-01 700 1 3 3 2024-02-01 800 2 1 4 2023-12-01 500 2 1
注意事项与最佳实践
- 日期格式统一性：务必先执行 pd.to_datetime()，避免字符串比较导致逻辑错误；
- 月末 vs 日粒度：本方案基于 MonthEnd 偏移量，适用于任意日（如 2023-12-15, 2024-01-20）自动对齐到自然月末，无需强制归一化为每月1日；
- 性能优化：对于超大规模数据，可先用 df.sort_values(['unique_id', 'ds'], ascending=[True, False]) 预排序，提升 groupby 效率；
- 扩展性提示：若需支持“最近 N 个月必须全存在”（而非任意连续段），可改用 period_range + isin 补全校验。
该方法兼顾准确性与可读性，适用于 Prophet、Nixtla（statsforecast）、Darts 等时序库的预处理流程，是构建高质量多序列数据集的关键一步。
