Pandas DataFrame中计算客户前一笔不同类型交易金额的教程

本教程详细阐述如何在pandas dataframe中，为每笔交易高效地查找同一客户之前发生的不同类型交易的金额。针对传统`apply`方法性能瓶颈和`shift`函数局限性，本文提出一种基于客户分组迭代并维护状态变量的解决方案，确保准确捕获时间序列中异类交易的关联信息，并提供详细代码示例与注意事项。

问题描述

在处理客户交易数据时，经常需要分析交易之间的关联性。一个常见的需求是，对于DataFrame中的每一笔交易，我们希望找到该客户之前发生过的、且交易类型与当前交易不同的最近一笔交易的金额。这要求我们不仅要考虑客户ID和交易日期，还要区分交易类型。

具体来说，给定一个包含KEY_ID（客户ID）、TYPE（交易类型，假设有两种类别）、DATE（交易日期）和AMOUNT（交易金额）的DataFrame，我们需要添加一个新列，记录当前交易的同一客户在当前交易日期之前发生的最近一笔不同类型交易的金额。如果不存在这样的交易，则记为NaN。

以下是一个示例输入数据和期望输出：

输入数据:

期望输出:

常见误区与挑战

在尝试解决此类问题时，开发者常会遇到以下挑战或误区：

1. 使用 DataFrame.apply() 方法配合自定义函数： 这种方法虽然直观，但在自定义函数内部对整个DataFrame进行过滤操作（例如 df[(df['KEY_ID'] == row['KEY_ID']) & (df['TYPE'] != row['TYPE']) & (df['DATE'] < row['DATE'])]）会导致严重的性能问题。对于DataFrame的每一行，都会执行一次全局筛选，这在数据量较大时会造成计算资源耗尽，甚至导致Python内核崩溃。

2. 使用 groupby().shift() 方法： Pandas的 shift() 函数通常用于获取前一个或后一个值，并且可以与 groupby() 结合使用，在每个组内进行偏移。例如，df.groupby(['KEY_ID', 'TYPE'])['AMOUNT'].shift() 会返回同一客户、同一类型的前一笔交易金额。然而，本问题要求的是“不同类型”的交易金额，shift() 无法直接实现这种跨类型查找的逻辑。

解决方案

解决此类问题的关键在于：按客户分组，并在每个客户组内按时间顺序迭代，同时维护不同交易类型的最新金额状态。

这种方法避免了全局筛选的性能瓶颈，并通过局部变量高效地跟踪所需信息。

核心思路：

1. 首先，确保DataFrame已按 KEY_ID 和 DATE 升序排序，这对于处理时间序列数据至关重要。
2. 然后，按 KEY_ID 对DataFrame进行分组。
3. 对于每个客户组，初始化变量来存储不同交易类型的最新金额（例如 last_motor_amount 和 last_tool_amount）。
4. 遍历客户组内的每一行交易：
   • 根据当前交易的 TYPE，将其对应的“前一笔不同类型交易金额”赋值给新列。
   • 更新当前交易类型对应的最新金额变量。

代码实现

下面是使用Python和Pandas实现此解决方案的完整代码：

import pandas as pd
import numpy as np

# 示例数据
data = {
    'KEY_ID': [1, 1, 1, 2, 2, 2],
    'TYPE': ['Motor', 'Tool', 'Tool', 'Tool', 'Motor', 'Tool'],
    'AMOUNT': [5000, 3000, 7000, 2000, 6000, 4000],
    'DATE': pd.to_datetime(['2020-01-01', '2020-02-01', '2020-03-01', '2020-01-15', '2020-02-15', '2020-03-15'])
}
df = pd.DataFrame(data)

# 1. 确保数据按客户ID和日期排序
df = df.sort_values(by=['KEY_ID', 'DATE']).reset_index(drop=True)

# 初始化新列
df['PREV_AMOUNT'] = np.nan

# 2. 按KEY_ID分组并迭代
grouped = df.groupby('KEY_ID')

for key_id, group in grouped:
    # 为每个客户初始化不同类型的最新交易金额
    last_motor_amount = np.nan
    last_tool_amount = np.nan

    # 3. 遍历组内的每一行
    for ind, row in group.iterrows():
        current_type = row['TYPE']
        current_amount = row['AMOUNT']

        if current_type == 'Motor':
            # 如果当前交易是Motor，则需要前一笔Tool交易的金额
            df.loc[ind, 'PREV_AMOUNT'] = last_tool_amount
            # 更新Motor类型的最新金额
            last_motor_amount = current_amount
        elif current_type == 'Tool':
            # 如果当前交易是Tool，则需要前一笔Motor交易的金额
            df.loc[ind, 'PREV_AMOUNT'] = last_motor_amount
            # 更新Tool类型的最新金额
            last_tool_amount = current_amount
        # 可以根据需要添加更多交易类型的处理

print(df)

代码解释：

闪念贝壳

闪念贝壳是一款AI 驱动的智能语音笔记，随时随地用语音记录你的每一个想法。

下载

1. 数据准备与排序：

   • 首先，创建示例DataFrame并确保 DATE 列为 datetime 类型。
   • df = df.sort_values(by=['KEY_ID', 'DATE']).reset_index(drop=True)：这一步至关重要。它确保了在处理每个客户的交易时，数据是按时间顺序排列的，这样我们才能正确地找到“前一笔”交易。reset_index(drop=True) 是为了重置索引，避免后续 loc 操作可能出现的索引错位问题。
   • df['PREV_AMOUNT'] = np.nan：初始化一个新列 PREV_AMOUNT，默认值为 NaN，因为最初可能没有前一笔不同类型的交易。

2. 按客户分组迭代：

   • grouped = df.groupby('KEY_ID')：将DataFrame按 KEY_ID 分组。这将生成一个迭代器，每次迭代返回一个客户ID和该客户对应的子DataFrame。
   • for key_id, group in grouped:：循环遍历每个客户组。

3. 组内行迭代与状态维护：

   • last_motor_amount = np.nan 和 last_tool_amount = np.nan：在进入每个客户组的循环时，都会为该客户初始化这两个变量。它们分别用于存储该客户最近一笔 'Motor' 类型和 'Tool' 类型的交易金额。
   • for ind, row in group.iterrows():：遍历当前客户组内的每一行交易。ind 是原始DataFrame的索引，row 是当前行的Series。
   • 条件判断与赋值：
     • 如果 current_type == 'Motor'：这意味着我们正在处理一笔 'Motor' 交易。根据问题要求，我们需要找到它之前最近的“不同类型”交易金额，即 Tool 类型的金额。因此，我们将 last_tool_amount 赋值给当前行的 PREV_AMOUNT。
     • 紧接着，last_motor_amount = current_amount：更新 last_motor_amount 为当前 'Motor' 交易的金额，以便后续 'Tool' 交易可以使用它。
     • 同理，如果 current_type == 'Tool'，则将 last_motor_amount 赋值给 PREV_AMOUNT，并更新 last_tool_amount。
   • df.loc[ind, 'PREV_AMOUNT'] = ...：使用 df.loc 根据原始索引 ind 精确地更新DataFrame中的 PREV_AMOUNT 列。

注意事项与扩展

1. 性能考量： 尽管此方法涉及Python级别的循环，但由于 groupby 操作在C语言层面进行了优化，并且每个组内的迭代是线性的，它比 df.apply 结合全局过滤的方案效率高得多。对于百万级甚至千万级的数据，只要客户数量不是极其庞大且每个客户的交易数量不是极少（导致频繁创建组），这种方法通常是可接受的。

2. 交易类型数量： 示例代码是针对两种交易类型（'Motor' 和 'Tool'）硬编码的。如果交易类型数量更多，可以考虑使用字典来动态存储不同类型的最新金额，例如：

   last_amounts = {} # 例如 {'Motor': np.nan, 'Tool': np.nan, 'Service': np.nan}
   # 或者更动态地：
   # all_types = df['TYPE'].unique()
   # last_amounts = {t: np.nan for t in all_types}
   
   for ind, row in group.iterrows():
       current_type = row['TYPE']
       current_amount = row['AMOUNT']
   
       # 获取除当前类型外所有其他类型的最新金额，并找到其中最近的一个（如果需要）
       # 对于本问题，是找到除当前类型外，特定“对立”类型的金额
       # 如果是任意不同类型，则需要更复杂的逻辑，例如存储所有类型的最新交易时间戳和金额，然后查找
   
       # 对于只有两种类型的情况，可以这样通用化：
       opposite_type = 'Tool' if current_type == 'Motor' else 'Motor' # 假设只有两种类型
       df.loc[ind, 'PREV_AMOUNT'] = last_amounts.get(opposite_type, np.nan)
       last_amounts[current_type] = current_amount

   对于多于两种类型且要求是“任意不同类型”的最近交易，则需要维护一个包含所有类型最新交易时间和金额的字典，并在每次迭代时遍历这个字典来找到最近的不同类型交易。

3. 初始值处理： np.nan 作为初始值是处理没有前一笔交易的正确方式。在后续分析中，可以根据需要使用 fillna() 方法将 NaN 替换为0或其他默认值。

4. 日期处理： 确保 DATE 列是 datetime 类型，这对于正确的排序和时间比较至关重要。

总结

在Pandas DataFrame中处理涉及跨行、跨类型且基于时间序列的复杂逻辑时，直接使用 df.apply() 配合复杂的行级查询通常效率低下。通过将问题分解为客户分组，并在每个组内进行迭代，同时巧妙地利用局部变量维护状态信息，可以构建出既高效又准确的解决方案。这种模式在处理各种时间序列相关的分组计算时都非常有用，例如计算滚动窗口指标、序列依赖性分析等。