如何用Pandas/Python计算新/复购/流失客户及营收的同比环比指标？

1. 数据预处理与基础指标构建 #

先统一日期格式，提取年、季度、月度维度，同时标记每个客户的首次/末次购买时间，为后续指标计算打基础：

import pandas as pd

# 假设原始数据框为df，包含字段：order_date, customer_id, revenue
df['order_date'] = pd.to_datetime(df['order_date'])
df['year'] = df['order_date'].dt.year
df['quarter'] = df['order_date'].dt.to_period('Q')
df['month'] = df['order_date'].dt.to_period('M')

# 计算每个客户的首次/末次购买时间
customer_lifecycle = df.groupby('customer_id').agg(
    first_purchase=('order_date', 'min'),
    last_purchase=('order_date', 'max')
).reset_index()
df = df.merge(customer_lifecycle, on='customer_id', how='left')

2. 按时间维度聚合核心指标 #

以月度为例，聚合总客户数、新客户数、复购客户数、流失客户数及对应营收，季度/年度逻辑完全一致，仅需更换分组维度：

# 月度维度聚合核心指标
monthly_agg = df.groupby('month').agg(
    total_customers=('customer_id', 'nunique'),
    total_revenue=('revenue', 'sum'),
    # 新客户：首次购买时间落在当前月度的客户
    new_customers=('customer_id', lambda x: x[df.loc[x.index, 'first_purchase'].dt.to_period('M') == x.name].nunique()),
    # 复购客户：当前月度有购买且首次购买时间早于当前月度的客户
    repeat_customers=('customer_id', lambda x: x[df.loc[x.index, 'first_purchase'].dt.to_period('M') < x.name].nunique())
).reset_index()

# 计算流失客户数：上月有购买记录但当前月无购买的客户
monthly_customer_sets = df.groupby('month')['customer_id'].apply(set)
monthly_agg['churned_customers'] = [
    len(monthly_customer_sets.shift(1)[i] - monthly_customer_sets[i]) if i > 0 else 0
    for i in range(len(monthly_agg))
]

# 计算流失客户对应营收（取该客户上月的营收总和，可根据业务逻辑调整）
churn_revenue = []
for idx, row in monthly_agg.iterrows():
    if idx == 0:
        churn_revenue.append(0)
    else:
        prev_month = monthly_agg.loc[idx-1, 'month']
        churned_customers = monthly_customer_sets[prev_month] - monthly_customer_sets[row['month']]
        churn_rev = df[(df['month'] == prev_month) & (df['customer_id'].isin(churned_customers))]['revenue'].sum()
        churn_revenue.append(churn_rev)
monthly_agg['churned_revenue'] = churn_revenue

3. 计算同比（YoY）、环比（QoQ/MoM）指标 #

用Pandas的shift方法实现向量化计算，避免低效的自连接，适配大数据集：

# 月度环比（MoM）：当前值较上月的变化率
monthly_agg['total_customers_mom'] = monthly_agg['total_customers'].pct_change()
monthly_agg['total_revenue_mom'] = monthly_agg['total_revenue'].pct_change()
monthly_agg['new_customers_mom'] = monthly_agg['new_customers'].pct_change(fill_method=None)
monthly_agg['repeat_customers_mom'] = monthly_agg['repeat_customers'].pct_change(fill_method=None)
monthly_agg['churned_customers_mom'] = monthly_agg['churned_customers'].pct_change(fill_method=None)

# 年度同比（YoY）：当前值较去年同期的变化率
monthly_agg['total_customers_yoy'] = monthly_agg['total_customers'].pct_change(periods=12)
monthly_agg['total_revenue_yoy'] = monthly_agg['total_revenue'].pct_change(periods=12)
# 季度同比可将periods设为4，季度环比则设为1

4. 滚动指标实现 #

完全支持滚动类指标计算，用rolling方法即可：

# 滚动3个月的平均总客户数
monthly_agg['rolling_3m_total_customers'] = monthly_agg['total_customers'].rolling(window=3).mean()
# 滚动12个月的累计营收
monthly_agg['rolling_12m_total_revenue'] = monthly_agg['total_revenue'].rolling(window=12).sum()

性能优化建议 #

• 优先使用groupby+agg的向量化操作，减少循环；若流失客户营收计算效率低，可改用merge+groupby优化
• 超大数据集可替换为Dask，支持并行计算与分块处理，API与Pandas基本兼容
• 提前过滤无效数据（如空customer_id、营收为0的记录），降低计算量

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