Python债务期限计算算法优化：3000+合同动态统计需求

业务需求 #

• 处理3000+份合同数据，每份包含账单（billed）与付款（paid）记录，存在一笔账单对应多笔付款、无固定付款期等场景
• 需完成两项核心计算：
  1. 单份合同的平均债务期限（账单金额到全额支付的天数，按金额加权）
  2. 按<1个月、1-2个月、2-5个月、>6个月四个期限区间，统计所有合同每日债务金额总和的动态变化
• 现有算法复杂度高、调试困难，寻求简洁高效的Python实现方案

示例数据 #

债务期限计算规则示例 #

2021年6月30日账单6919.07于2021年9月5日全额付清：

• 3000元于2021年8月12日支付，债务期限43天
• 剩余3919.07元于2021年9月5日支付，债务期限67天
  2021年9月5日支付的10000元扣除上述3919.07元后，剩余3080.93元用于抵扣2021年7月31日的账单

实现步骤与代码 #

1. 数据预处理 #

统一日期格式、清洗金额数据（替换逗号为小数点），处理空值：

import pandas as pd

# 读取单份合同数据（实际场景可从CSV/数据库读取）
df = pd.DataFrame([
    {"date": "30.06.2021", "type of operation": "billed", "sum (billed)": "6919,07", "sum (paid)": ""},
    {"date": "31.07.2021", "type of operation": "billed", "sum (billed)": "4829,65", "sum (paid)": ""},
    {"date": "12.08.2021", "type of operation": "paid", "sum (billed)": "", "sum (paid)": "3000"},
    {"date": "31.08.2021", "type of operation": "billed", "sum (billed)": "3845,6", "sum (paid)": ""},
    {"date": "05.09.2021", "type of operation": "paid", "sum (billed)": "", "sum (paid)": "10000"},
])

# 数据清洗：日期转datetime，金额转数值
df["date"] = pd.to_datetime(df["date"], format="%d.%m.%Y")
df["sum (billed)"] = df["sum (billed)"].str.replace(",", ".").astype(float).fillna(0)
df["sum (paid)"] = df["sum (paid)"].str.replace(",", ".").astype(float).fillna(0)

2. FIFO匹配账单与付款，计算平均债务期限 #

采用先进先出逻辑匹配账单和付款，按金额加权计算平均债务期限：

def process_single_contract(df):
    # 分离并排序账单、付款数据
    billed_records = df[df["type of operation"] == "billed"].sort_values("date").reset_index(drop=True)
    paid_records = df[df["type of operation"] == "paid"].sort_values("date").reset_index(drop=True)
    
    # 追踪账单剩余金额
    billed_remaining = billed_records["sum (billed)"].copy()
    payment_mapping = []
    total_weighted_days = 0.0
    total_billed = billed_records["sum (billed)"].sum()
    
    paid_idx = 0
    for billed_idx, (billed_date, billed_amount) in enumerate(zip(billed_records["date"], billed_records["sum (billed)"])):
        remaining = billed_remaining[billed_idx]
        # 用付款逐笔抵扣当前账单剩余金额
        while remaining > 1e-6 and paid_idx < len(paid_records):
            paid_date = paid_records.loc[paid_idx, "date"]
            paid_amount = paid_records.loc[paid_idx, "sum (paid)"]
            
            deduct_amount = min(remaining, paid_amount)
            days_outstanding = (paid_date - billed_date).days
            
            # 记录匹配明细
            payment_mapping.append({
                "billed_date": billed_date,
                "paid_date": paid_date,
                "amount": deduct_amount,
                "days": days_outstanding
            })
            
            # 更新剩余金额
            remaining -= deduct_amount
            billed_remaining[billed_idx] = remaining
            paid_records.loc[paid_idx, "sum (paid)"] -= deduct_amount
            
            # 付款耗尽则切换到下一笔
            if paid_records.loc[paid_idx, "sum (paid)"] < 1e-6:
                paid_idx += 1
        
        # 累加当前账单的加权天数贡献
        billed_total_days = sum(p["amount"] * p["days"] for p in payment_mapping if p["billed_date"] == billed_date)
        total_weighted_days += billed_total_days
    
    # 计算平均债务期限（加权平均）
    avg_debt_days = total_weighted_days / total_billed if total_billed > 0 else 0.0
    return payment_mapping, avg_debt_days

# 处理示例合同
payment_mapping, avg_debt_days = process_single_contract(df)
print(f"单份合同平均债务期限：{avg_debt_days:.2f}天")

3. 每日债务金额与区间统计 #

生成完整日期序列，按天统计各期限区间的未结清债务总和：

def calculate_daily_debt_buckets(payment_mapping):
    # 获取所有涉及的日期范围
    min_date = min(p["billed_date"] for p in payment_mapping)
    max_date = max(p["paid_date"] for p in payment_mapping)
    date_range = pd.date_range(start=min_date, end=max_date, freq="D")
    
    daily_debt_stats = []
    for current_date in date_range:
        debt_buckets = {"<1个月": 0.0, "1-2个月": 0.0, "2-5个月": 0.0, ">6个月": 0.0}
        
        for p in payment_mapping:
            # 判断当前日期是否处于债务存续期
            if p["billed_date"] <= current_date < p["paid_date"]:
                days_outstanding = (current_date - p["billed_date"]).days
                amount = p["amount"]
                
                # 按区间分类
                if days_outstanding < 30:
                    debt_buckets["<1个月"] += amount
                elif 30 <= days_outstanding < 60:
                    debt_buckets["1-2个月"] += amount
                elif 60 <= days_outstanding < 150:
                    debt_buckets["2-5个月"] += amount
                elif days_outstanding >= 180:
                    debt_buckets[">6个月"] += amount
        
        daily_debt_stats.append({
            "date": current_date,
            **debt_buckets
        })
    
    return pd.DataFrame(daily_debt_stats)

# 生成每日债务统计
daily_debt_df = calculate_daily_debt_buckets(payment_mapping)
print(daily_debt_df.head())

批量处理扩展 #

针对3000+份合同，可通过并行处理提升效率：

# 示例：批量处理合同列表（假设contracts是包含每份合同DataFrame的列表）
from concurrent.futures import ProcessPoolExecutor

def batch_process_contracts(contracts):
    with ProcessPoolExecutor() as executor:
        results = executor.map(process_single_contract, contracts)
    
    # 合并所有合同的付款明细，用于后续全局每日债务统计
    all_payment_mappings = []
    all_avg_days = []
    for mapping, avg_days in results:
        all_payment_mappings.extend(mapping)
        all_avg_days.append(avg_days)
    
    # 全局每日债务统计
    global_daily_debt = calculate_daily_debt_buckets(all_payment_mappings)
    return global_daily_debt, all_avg_days

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