基于关联列取值返回合约状态的Pandas DataFrame实现

规则拆解 #

按照业务逻辑拆分为三层判定优先级，从基础规则到跨合约例外逐层校验：

• 基础判定项1：合约到期判定。基于date_activated + term_months计算合约到期时间，若到期时间早于基准日期（示例基准为2021-05-01），直接判定为Inactive，不受后续规则影响。
• 基础判定项2：无缴费失效判定。未到期的合约，从激活月份开始统计缴费序列，若最后一次缴费后，连续4个月及以上缴费值为0，初步判定为待失效状态。
• 跨合约豁免规则：同一账号下，若待失效的未到期合约，其激活时间早于同账号下某份Early类型且自身为有效状态的合约，则取消该待失效合约的Inactive判定，标记为Active。有效Early合约的定义为：未到期、自身未触发连续4个月无缴费失效的合约。

实现代码 #

import pandas as pd
import datetime

# 构造与示例一致的输入DataFrame
df = pd.DataFrame({
    'account_id': [1, 1, 1, 2, 2, 3, 3],
    'contract_id': ['A', 'B', 'C', 'K', 'F', 'Z', 'Y'],
    'date_activated': pd.to_datetime(['2020-12-20', '2020-12-25', '2021-03-06', '2020-12-01', '2021-02-23', '2020-12-01', '2021-01-20']),
    'term_months': [1, 12, 12, 4, 6, 12, 12],
    'contract_type': ['Original', 'Upgrade', 'Upgrade', 'Original', 'Renewal', 'Original', 'Upgrade'],
    'renewal_type': ['Regular', 'Regular', 'Early', 'Regular', 'Early', 'Regular', 'Early'],
    datetime.datetime(2020, 12, 1): [200.0, 300.0, 0.0, 100.0, 0.0, 80.0, 0.0],
    datetime.datetime(2021, 1, 1): [0.0, 0.0, 0.0, 100.0, 0.0, 0.0, 90.0],
    datetime.datetime(2021, 2, 1): [0.0, 0.0, 0.0, 100.0, 50.0, 0.0, 0.0],
    datetime.datetime(2021, 3, 1): [0.0, 0.0, 400.0, 0.0, 50.0, 0.0, 0.0],
    datetime.datetime(2021, 4, 1): [0.0, 0.0, 400.0, 0.0, 0.0, 0.0, 0.0],
    datetime.datetime(2021, 5, 1): [0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0],
    datetime.datetime(2021, 6, 1): [0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0]
})

# 配置基准日期
base_date = datetime.datetime(2021, 5, 1)
# 提取所有月度缴费列
date_cols = [col for col in df.columns if isinstance(col, datetime.datetime)]

# 第一步：计算单合约基础判定标记
df['is_expired'] = False
df['is_no_pay_inactive'] = False
for idx, row in df.iterrows():
    # 计算合约到期日
    expire_date = row['date_activated'] + pd.DateOffset(months=row['term_months'])
    if expire_date < base_date:
        df.loc[idx, 'is_expired'] = True
        continue
    # 提取合约有效期内、从激活到基准日的缴费序列
    valid_cols = [col for col in date_cols 
                  if row['date_activated'].replace(day=1) <= col <= base_date]
    pay_seq = row[valid_cols].values
    # 定位最后一次有缴费的位置
    last_pay_idx = -1
    for i, amount in enumerate(pay_seq):
        if amount > 0:
            last_pay_idx = i
    # 计算最后一次缴费后连续无缴费的月数
    no_pay_months = len(pay_seq) - last_pay_idx - 1
    # 无缴费月数≥4，标记待失效
    if no_pay_months >=4:
        df.loc[idx, 'is_no_pay_inactive'] = True

# 第二步：按账号分组应用跨合约豁免规则
status_res = []
for acc_id, acc_group in df.groupby('account_id'):
    # 组内按激活时间排序，方便定位前后合约
    acc_group_sorted = acc_group.sort_values('date_activated')
    # 筛选组内所有有效Early合约（未到期、无缴费失效）
    valid_early_contracts = acc_group_sorted[
        (~acc_group_sorted['is_expired']) &
        (~acc_group_sorted['is_no_pay_inactive']) &
        (acc_group_sorted['renewal_type'] == 'Early')
    ]['date_activated'].tolist()

    # 逐合约判定最终状态
    for idx, row in acc_group_sorted.iterrows():
        # 到期的合约直接标记Inactive
        if row['is_expired']:
            status_res.append((idx, 'Inactive'))
            continue
        # 待失效的未到期合约，检查是否有后续有效Early合约可豁免
        if row['is_no_pay_inactive']:
            has_exempt = any(early_date > row['date_activated'] 
                             for early_date in valid_early_contracts)
            status = 'Active' if has_exempt else 'Inactive'
            status_res.append((idx, status))
            continue
        # 既未到期也未触发无缴费失效的合约直接标记Active
        status_res.append((idx, 'Active'))

# 回写状态到原DataFrame
status_map = dict(status_res)
df['Status'] = df.index.map(status_map)
# 删除中间计算列
df = df.drop(columns=['is_expired', 'is_no_pay_inactive'])

输出验证 #

运行上述代码后，df['Status']列的结果将完全匹配示例预期：

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