基于Pandas利用日期列追踪保险理赔状态并统计状态数量

需求概述 #

基于理赔状态的日期记录，追踪患者保险理赔的状态变化，按日统计各状态的存续数量及状态转换次数，同时需按指定分组执行统计。

数据集 #

统计规则 #

• 若理赔当日转换状态，原状态数量次日才减少；
• 理赔在状态变更前保持原状态；
• 跨日变更状态的理赔，原状态数量在变更当日减少；
• 不同状态仅需关注特定转换方向（如New仅关注转换到Accepted/Denied）；
• 统计日期范围从New列最早日期至当日；
• 需按Group列分组统计；
• 所有状态均需执行上述统计逻辑。

现有方案问题 #

原代码存在以下缺陷：

1. 变量名错误：使用未定义的Approved列（实际应为Accepted）；
2. 逻辑冗余：手动拆分「当日转换」「跨日转换」等场景，代码重复且易遗漏；
3. 效率低下：多次手动合并、分组操作，未利用pandas向量化特性处理时间序列；
4. 统计逻辑错误：状态存续的时间区间计算不符合规则，导致每日数量统计偏差。

优化实现方案 #

核心思路：无需循环，用长表事件流+区间重叠统计实现 #

最佳方式是将宽表结构的状态数据转换为「状态事件长表」，通过向量化操作计算每个状态的存续区间，再结合日期序列统计每日状态数量和转换次数，完全避免循环，效率和可维护性大幅提升。

具体步骤代码 #

1. 数据预处理：解析时间格式，转换为状态事件长表

import pandas as pd
from datetime import timedelta

# 处理特殊时间格式（将最后一个冒号替换为小数点，适配pd.to_datetime）
def parse_datetime(s):
    if pd.isna(s):
        return pd.NaT
    return pd.to_datetime(s.replace(':', '.', 2), utc=True)

# 加载示例数据
data = [
    ["001", "2021-01-01T09:58:35:335Z", "2021-01-01T10:05:43:000Z", "", "", "", "A"],
    ["002", "2021-01-01T06:30:30:000Z", "2021-03-01T04:11:45:000Z", "2021-03-01T04:11:53:000Z", "", "", "A"],
    ["003", "2021-02-14T14:23:54:154Z", "2021-02-15T11:11:56:000Z", "", "", "2021-02-15T11:15:00:000Z", "A"],
    ["004", "2021-02-14T15:36:05:335Z", "2021-02-14T17:15:30:000Z", "", "", "", "A"],
    ["005", "2021-02-14T15:56:59:009Z", "2021-03-01T04:11:45:000Z", "", "", "", "A"]
]
df = pd.DataFrame(data, columns=["ClaimID", "New", "Accepted", "Denied", "Pending", "Expired", "Group"])

# 解析所有状态时间列
status_cols = ["New", "Accepted", "Denied", "Pending", "Expired"]
for col in status_cols:
    df[col] = df[col].apply(parse_datetime)

# 生成每个理赔的状态存续区间
def get_status_intervals(row):
    # 提取非空的状态时间并按时间排序
    status_events = sorted([(row[col], col) for col in status_cols if not pd.isna(row[col])], key=lambda x: x[0])
    intervals = []
    for i in range(len(status_events)):
        start_time, status = status_events[i]
        # 确定状态结束时间
        if i == len(status_events) - 1:
            # 最后一个状态持续到当日
            end_time = pd.Timestamp.utcnow()
        else:
            next_time, _ = status_events[i+1]
            # 按规则处理结束日期：当日转换则次日失效，跨日则当日失效
            if start_time.date() == next_time.date():
                end_time = next_time + timedelta(days=1)
            else:
                end_time = next_time
        intervals.append({
            "ClaimID": row["ClaimID"],
            "Group": row["Group"],
            "Status": status,
            "StartDate": start_time.date(),
            "EndDate": end_time.date()
        })
    return pd.DataFrame(intervals)

# 合并所有理赔的状态区间
status_intervals_df = pd.concat(df.apply(get_status_intervals, axis=1).tolist(), ignore_index=True)

2. 生成统计日期序列

# 确定统计日期范围：从最早的New日期到当日
min_new_date = df["New"].min().date()
today = pd.Timestamp.utcnow().date()

# 按分组生成每日日期序列
date_list = []
for group in df["Group"].unique():
    dates = pd.date_range(start=min_new_date, end=today, freq="D").date
    date_list.extend([{"Group": group, "Date": d} for d in dates])
date_df = pd.DataFrame(date_list)

3. 统计每日状态数量

# 通过区间重叠统计每日各状态的理赔数
daily_status_counts = status_intervals_df.merge(date_df, on="Group", how="right")
# 筛选出日期在状态存续区间内的记录
daily_status_counts = daily_status_counts[
    (daily_status_counts["StartDate"] <= daily_status_counts["Date"]) &
    (daily_status_counts["Date"] < daily_status_counts["EndDate"])
]
# 分组统计数量
daily_status_counts = daily_status_counts.groupby(["Group", "Date", "Status"]).size().unstack(fill_value=0).reset_index()

4. 统计每日状态转换次数

# 生成状态转换事件
def get_transition_events(row):
    status_events = sorted([(row[col], col) for col in status_cols if not pd.isna(row[col])], key=lambda x: x[0])
    transitions = []
    # 提取相邻状态的转换事件
    for i in range(len(status_events)-1):
        from_time, from_status = status_events[i]
        to_time, to_status = status_events[i+1]
        transitions.append({
            "Group": row["Group"],
            "TransitionDate": to_time.date(),
            "FromStatus": from_status,
            "ToStatus": to_status
        })
    return pd.DataFrame(transitions)

transition_events_df = pd.concat(df.apply(get_transition_events, axis=1).tolist(), ignore_index=True)
# 分组统计转换次数
daily_transition_counts = transition_events_df.groupby(["Group", "TransitionDate", "FromStatus", "ToStatus"]).size().unstack(level=[2,3], fill_value=0).reset_index().rename(columns={"TransitionDate": "Date"})

5. 合并最终统计结果

final_stats = date_df.merge(daily_status_counts, on=["Group", "Date"], how="left").fillna(0)
final_stats = final_stats.merge(daily_transition_counts, on=["Group", "Date"], how="left").fillna(0)

方案优势 #

1. 无循环、全向量化：利用pandas的内置函数处理时间序列和分组统计，避免手动循环带来的效率问题；
2. 逻辑清晰：通过状态区间统一处理「当日转换」「跨日转换」规则，避免场景拆分遗漏；
3. 可扩展性强：新增状态时只需修改status_cols列表，无需调整核心逻辑；
4. 统计准确：严格按照规则计算状态存续区间，确保每日数量统计符合需求。

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