基于窗口条件优化Pandas中Groupby与Transform的高效实现

针对你的需求，直接逐行循环处理效率极低，这里用Pandas原生矢量化+二分查找的方法实现，性能提升几个数量级：

前提准备：确保日期列格式正确 #

先将date1和sample_date转为datetime类型，否则日期计算会失效：

import pandas as pd
import numpy as np

# 转换日期格式（如果原始数据是字符串的话）
df['date1'] = pd.to_datetime(df['date1'])
df['sample_date'] = pd.to_datetime(df['sample_date'])

方案1：统计满足条件的产品行数（不做产品去重） #

如果需求是统计该客户下，所有date1 + 12个月 ≤ 当前sample_date的记录行数（每条符合条件的产品记录都计入统计）：

# 1. 按客户+采样日期排序，保证分组内数据有序
df_sorted = df.sort_values(['clients', 'sample_date'])

# 2. 计算每个采样日期的截止日期：当前sample_date往前推12个月
df_sorted['cutoff_date'] = df_sorted['sample_date'] - pd.DateOffset(months=12)

# 3. 分组内用二分查找快速计数
def count_valid_rows(group):
    # 对当前客户的所有date1排序
    sorted_dates = np.sort(group['date1'].values)
    # 用searchsorted统计每个cutoff_date前的有效date1数量（二分查找，O(log n)时间）
    counts = np.searchsorted(sorted_dates, group['cutoff_date'].values, side='right')
    return pd.Series(counts, index=group.index)

# 4. 分组应用并赋值
df_sorted['count_products'] = df_sorted.groupby('clients', group_keys=False).apply(count_valid_rows)

# 5. 恢复原数据顺序（如果需要）
df = df_sorted.sort_index()

方案2：统计满足条件的唯一产品数（产品去重） #

如果需求是统计该客户下，存在至少一个date1 + 12个月 ≤ 当前sample_date的不同产品数量：

# 1. 预处理：先按客户+产品分组，取每个产品最早的date1（只要最早的date1满足条件，该产品就算有效）
product_min_dates = df.groupby(['clients', 'products'])['date1'].min().reset_index()

# 2. 按客户+采样日期排序
df_sorted = df.sort_values(['clients', 'sample_date'])
df_sorted['cutoff_date'] = df_sorted['sample_date'] - pd.DateOffset(months=12)

# 3. 分组内统计有效唯一产品数
def count_unique_valid_products(group):
    # 获取当前客户的所有产品最早date1并排序
    client_prod_dates = product_min_dates[product_min_dates['clients'] == group.name]['date1'].values
    sorted_prod_dates = np.sort(client_prod_dates)
    # 二分查找统计符合条件的产品数
    counts = np.searchsorted(sorted_prod_dates, group['cutoff_date'].values, side='right')
    return pd.Series(counts, index=group.index)

# 4. 分组应用并赋值
df_sorted['count_products'] = df_sorted.groupby('clients', group_keys=False).apply(count_unique_valid_products)

# 5. 恢复原顺序
df = df_sorted.sort_index()

效率说明 #

• 用np.searchsorted替代逐行循环：基于二分查找的矢量化操作，时间复杂度从O(n²)降到O(n log n)
• 分组操作是Pandas原生优化的，避免了Python级别的循环开销
• 提前排序保证了二分查找的有效性，同时减少分组内的计算冗余

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