如何用Pandas实现客户层级的12个月锁定与升级逻辑

需求规则 #

• 客户账户旅行消费按12个月滚动周期追踪，达标后即时获得对应层级及权益
• 层级达标后锁定12个月，期间消费下降仍保留当前层级至到期日
• 若达到更高层级，立即升级并重新锁定12个月
• 锁定周期结束后重新评估资格
• 收入按日计算，延迟7天统计（已完成滚动求和计算）

当前数据与现有代码 #

已通过SQLAlchemy拉取数据并转换为DataFrame，计算12个月滚动求和后的数据如下：

tot_rev_base    rolling
DATE                         
2022-08-21           0.00  139558.47
2022-08-22           0.00  139558.47
2022-08-23        3902.93  140314.65
2022-08-24         467.60  138790.25
2022-08-25           0.00  138235.25
2022-08-26        1265.10  135559.90
2022-08-27           0.00  134676.00
2022-08-28           0.00  134676.00
2022-08-29           0.00  134676.00
2022-08-30         664.25  134935.70
2022-08-31        1712.40  132872.30
2022-09-01         458.70  131291.50
2022-09-02         802.00  131305.75
2022-09-03           0.00  131305.75
2022-09-04           0.00  128518.30
2022-09-05           0.00  128518.30
2022-09-06           0.00  128518.30
2022-09-07         655.70  128626.60
2022-09-08        1238.25  129864.85 

现有仅能按当日滚动值分配层级的函数：

def Tiers(row):
    date = row[0]
    revenue = row[1][0]
    rolling = row[1][1]

    if rolling < 10000:
        df3.loc[date,'tier'] = 0
    elif rolling < 150000:
        df3.loc[date,'tier'] = 1
    elif rolling < 300000:
        df3.loc[date,'tier'] = 2
    elif rolling < 500000:
        df3.loc[date,'tier'] = 3
    elif rolling > 500000:
        df3.loc[date,'tier'] = 4

现存问题 #

上述代码无法实现层级锁定12个月、升级重置锁定周期、到期前保留层级的核心规则。

核心思路是追踪两个状态变量：当前生效的层级、层级的到期日期，按日期顺序遍历数据，根据规则动态更新这两个变量，最终赋值当日层级。

实现代码 #

import pandas as pd
from dateutil.relativedelta import relativedelta

# 1. 定义层级阈值（方便后续维护调整）
TIER_THRESHOLDS = [
    (0, 10000),    # tier 0: 滚动消费 <10000
    (1, 150000),   # tier 1: 10000 ≤ 滚动消费 <150000
    (2, 300000),   # tier 2: 150000 ≤ 滚动消费 <300000
    (3, 500000),   # tier 3: 300000 ≤ 滚动消费 <500000
    (4, float('inf')) # tier 4: 滚动消费 ≥500000
]

def get_tier_from_rolling(rolling_value):
    """根据滚动消费值返回对应的层级"""
    for tier, threshold in TIER_THRESHOLDS:
        if rolling_value < threshold:
            return tier
    return 0

# 2. 初始化状态变量
current_tier = 0
# 初始到期日设为数据最早日期的前一天，确保第一天就会评估
expiry_date = df3.index.min() - relativedelta(days=1)

# 3. 遍历每日数据，按规则更新层级
df3['tier'] = 0  # 先初始化tier列
for date, row in df3.iterrows():
    rolling = row['rolling']
    
    # 情况1：当前日期已过到期日，重新评估层级
    if date >= expiry_date:
        candidate_tier = get_tier_from_rolling(rolling)
        current_tier = candidate_tier
        expiry_date = date + relativedelta(years=1)
    else:
        # 情况2：仍在锁定周期内，仅检查是否能升级
        new_possible_tier = get_tier_from_rolling(rolling)
        if new_possible_tier > current_tier:
            current_tier = new_possible_tier
            expiry_date = date + relativedelta(years=1)
    
    # 给当日赋值当前生效的层级
    df3.loc[date, 'tier'] = current_tier

代码说明 #

• TIER_THRESHOLDS：将层级阈值封装为列表，后续调整层级标准时只需修改此处，无需改动核心逻辑
• get_tier_from_rolling：统一层级判断逻辑，避免重复代码
• 状态变量current_tier和expiry_date：分别追踪当前生效层级及该层级的到期时间
• 遍历逻辑：
  • 若当前日期已过到期日，直接按当日滚动值重新评估层级并更新锁定周期
  • 若仍在锁定周期内，仅当滚动值达到更高层级时才升级并重置锁定周期，否则保持原层级
• 使用relativedelta处理日期加减12个月，比pd.DateOffset更准确（自动处理闰年、不同月份天数差异）

注意事项 #

• 确保DataFrame的索引按日期升序排列，否则遍历顺序错误会导致逻辑失效
• 若处理多客户数据，需按客户分组后再执行上述逻辑
• 关于收入延迟7天统计：如果rolling列已考虑该规则则无需额外处理；若未考虑，需先调整滚动窗口的计算逻辑（例如将日期提前7天再计算滚动求和）

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