本文介绍如何基于时间序列连续性，从 pandas dataframe 中筛选出满足“最近 n 个自然月内存在至少 n 条连续月度观测”的 unique_id 及其全部记录，适用于时序建模前的数据质量校验。
本文介绍如何基于时间序列连续性，从 pandas dataframe 中筛选出满足“最近 n 个自然月内存在至少 n 条连续月度观测”的 unique_id 及其全部记录，适用于时序建模前的数据质量校验。
在时间序列分析（如 Prophet、N-BEATS 或 LightGBM 多步预测）中，常需确保每个 unique_id（如商品 ID、门店 ID）具备足够长且连续的月度历史数据——尤其强调“从最新月份向前连续”这一业务逻辑。例如：要求每个 ID 至少有 2 个相邻自然月（如 2024-02 和 2024-01）的观测，而非任意两个月（如 2024-02 和 2023-12）。本文提供一种高效、可扩展的 Pandas 实现方案。
核心思路：用 MonthEnd 判断月度连续性
关键在于将日期转换为 Period('M')（代表自然月），再利用 diff() 计算相邻记录间的时间差。若两月连续（如 2024-02 → 2024-01），其差值应严格等于 pd.offsets.MonthEnd()；否则视为断点。通过 cumsum() 标记连续段，即可统计每段长度。
✅ 推荐方案：直接筛选满足条件的 unique_id
以下函数接受 DataFrame 和最小连续月数 N，返回所有满足条件的子集：
import pandas as pd
import numpy as np
def filter_by_min_consecutive_months(df, unique_col='unique_id', date_col='ds', N=2):
"""
筛选在时间上具有至少 N 个连续自然月观测的 unique_id 全部记录
Parameters:
-----------
df : pd.DataFrame
输入数据，需包含 date_col（日期字符串或 datetime）和 unique_col（分组列）
unique_col : str
唯一标识列名，默认 'unique_id'
date_col : str
日期列名，默认 'ds'
N : int
最小连续月份数（必须为自然月，如 2024-02, 2024-01）
Returns:
--------
pd.DataFrame : 满足条件的原始记录子集
"""
# 确保日期列为 datetime
df = df.copy()
df[date_col] = pd.to_datetime(df[date_col])
# 转换为月度 Period，并按 unique_id 分组
monthly_period = df[date_col].dt.to_period('M')
def has_min_consecutive(g):
# 对每个 unique_id 的月度周期序列，从最新月开始倒序处理
# 计算相邻月差：连续则 diff == MonthEnd()，否则不等
diff_mask = g[::-1].diff().ne(pd.offsets.MonthEnd())
# cumsum 标记连续段（断点处 cumsum+1）
streak_id = diff_mask.cumsum()
# 统计每段长度，检查是否存在 ≥ N 的段
return (g.groupby(streak_id).size() >= N).any()
# 对每个 unique_id 判断是否达标
keep_ids = monthly_period.groupby(df[unique_col]).agg(has_min_consecutive)
# 返回所有属于达标 unique_id 的原始行
return df[df[unique_col].isin(keep_ids[keep_ids].index)]
# 示例使用
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]
}
input_df = pd.DataFrame(data)
result = filter_by_min_consecutive_months(input_df, N=2)
print(result)输出：
ds y unique_id 0 2024-02-01 500 1 1 2024-01-01 600 1 2 2023-12-01 700 1
? 进阶用法：查看各记录所属连续段长度
若需进一步分析（如保留最长连续段、排除短于阈值的片段），可计算每行对应的连续段长度：
def add_consecutive_streak(df, unique_col='unique_id', date_col='ds'):
df = df.copy()
df[date_col] = pd.to_datetime(df[date_col])
monthly_period = df[date_col].dt.to_period('M')
def calc_streak(g):
diff_mask = g[::-1].diff().ne(pd.offsets.MonthEnd())
streak_id = diff_mask.cumsum()
return g.groupby(streak_id).transform('size')
df['consecutive_streak'] = (
monthly_period
.groupby(df[unique_col])
.transform(calc_streak)
)
return df
# 应用后可灵活过滤
enriched = add_consecutive_streak(input_df)
print(enriched[['ds', 'y', 'unique_id', 'consecutive_streak']])输出：
ds y unique_id consecutive_streak 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
⚠️ 注意事项
- 时间精度：本方法依赖 Period('M')，自动对齐自然月（忽略日/时/分），适合月度聚合场景；若需精确到日级连续性，请改用 pd.Timedelta('30D') 并调整阈值。
- 排序无关性：内部使用 [::-1] 倒序处理，无需预先按日期排序，但建议保持原始顺序便于理解。
- 性能优化：对于超大数据集（>100 万行），可先用 df.sort_values([unique_col, date_col], inplace=True) 提升 groupby 效率。
- 边界情况：单条记录自动视为长度为 1 的连续段；空组或全 NaN 日期会引发异常，建议前置清洗。
该方案兼顾准确性与可读性，是构建稳健时序数据管道的关键一环。
