筛选指定月份数的unique_id方法

本文深入解析了一种基于自然月连续性的高效数据筛选方法，专为时序建模前的数据质量校验而设计——它能精准识别出那些在最近N个自然月中拥有至少N条**严格相邻、无间断**月度观测（如2024-02与2024-01，而非2024-02与2023-12）的unique_id，并完整保留其所有历史记录；通过Pandas的Period('M')、倒序diff()和cumsum()巧妙标记连续段，配合可复用的filter_by_min_consecutive_months函数，既确保业务逻辑严谨性（强调“从最新月向前连续”），又兼顾性能与可扩展性，是构建可靠预测 pipeline 不可或缺的实用利器。

本文介绍如何基于时间序列连续性，从 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 日期会引发异常，建议前置清洗。

该方案兼顾准确性与可读性，是构建稳健时序数据管道的关键一环。

今天关于《筛选指定月份数的unique_id方法》的内容介绍就到此结束，如果有什么疑问或者建议，可以在golang学习网公众号下多多回复交流；文中若有不正之处，也希望回复留言以告知！