基于Python DataFrame按Item_Id计算销货成本及成本余额问题

问题背景 #

已合并采购、采购退回、销售、销售退回四张业务表为单个DataFrame，且已按item_Id、日期、交易类型（Purchase/return purchase/return sales/sales）完成排序，成功计算出BalanceQty字段。但在计算BalanceCost和UnitCost时，使用shift()方法无法获取上一行已更新的计算结果，导致公式无法正确执行：

• BalanceCost规则：
  • 交易类型为Purchase/return purchase：上一行BalanceCost + 当前交易Cost
  • 交易类型为sales/return sales：(上一行BalanceCost / 上一行BalanceQty) * 当前BalanceQty
• UnitCost规则：BalanceCost / BalanceQty

核心问题 #

shift()方法仅能获取原始列的上一行值，无法追踪逐行计算后更新的BalanceCost值，因此必须采用逐组迭代计算的方式，确保每一行的计算依赖上一行的最终结果。

解决方案代码 #

1. 模拟业务数据（匹配实际场景） #

import pandas as pd

# 模拟合并后的业务数据表
data = {
    'item_Id': ['A', 'A', 'A', 'B', 'B'],
    'date': ['2023-01-01', '2023-01-02', '2023-01-03', '2023-01-01', '2023-01-02'],
    'transaction_type': ['Purchase', 'sales', 'return purchase', 'Purchase', 'return sales'],
    'Qty': [100, -30, 20, 50, 10],
    'Cost': [1000, 0, 200, 500, 0],  # 销售类交易无直接Cost，成本来自库存结转
    'BalanceQty': [100, 70, 90, 50, 60]
}
df = pd.DataFrame(data)
df['date'] = pd.to_datetime(df['date'])

2. 自定义分组计算函数 #

def calculate_inventory_metrics(group):
    # 重置分组内索引，方便逐行迭代
    group = group.reset_index(drop=True)
    
    # 初始化第一行数据（默认第一行为采购类交易，做容错处理）
    first_trans_type = group.iloc[0]['transaction_type']
    if first_trans_type in ['Purchase', 'return purchase']:
        group.loc[0, 'BalanceCost'] = group.iloc[0]['Cost']
    else:
        group.loc[0, 'BalanceCost'] = 0.0  # 无初始采购的特殊情况设为0
    
    group.loc[0, 'UnitCost'] = group.loc[0, 'BalanceCost'] / group.loc[0, 'BalanceQty']
    
    # 逐行迭代计算后续交易的指标
    for idx in range(1, len(group)):
        # 获取上一行已计算的结果
        prev_balance_cost = group.loc[idx-1, 'BalanceCost']
        prev_balance_qty = group.loc[idx-1, 'BalanceQty']
        
        # 获取当前行的参数
        current_balance_qty = group.loc[idx, 'BalanceQty']
        current_cost = group.loc[idx, 'Cost']
        trans_type = group.loc[idx, 'transaction_type']
        
        # 按规则计算BalanceCost
        if trans_type in ['Purchase', 'return purchase']:
            group.loc[idx, 'BalanceCost'] = prev_balance_cost + current_cost
        else:
            group.loc[idx, 'BalanceCost'] = (prev_balance_cost / prev_balance_qty) * current_balance_qty
        
        # 计算UnitCost
        group.loc[idx, 'UnitCost'] = group.loc[idx, 'BalanceCost'] / current_balance_qty
    
    return group

# 按item_Id分组应用计算函数
final_df = df.groupby('item_Id', group_keys=False).apply(calculate_inventory_metrics)

3. 最终结果示例 #

执行后final_df将包含正确计算的BalanceCost和UnitCost字段，输出如下：

关键说明 #

• 迭代计算确保每一行的BalanceCost都依赖上一行的最终结果，而非原始列的偏移值，完美解决shift()的局限性。
• 针对分组内第一行做了容错处理，避免无初始采购数据时出现计算错误。
• 若处理超大型数据集，可结合numba库对迭代函数进行加速，提升计算效率。

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