本教程旨在解决在pandas dataframe中，为每笔交易查找同一客户之前不同类别交易金额的问题。文章将分析常见错误方法及其局限性，并提供一种基于分组迭代和状态变量的优化解决方案，通过维护不同交易类型的最新金额，高效地计算出所需的跨品类前一笔交易金额。
1. 问题描述
在处理客户交易数据时，我们经常需要进行复杂的序列分析。本教程将聚焦于一个具体场景：给定一个包含客户交易记录的DataFrame，我们需要为每条交易记录添加一个新列PREV_AMOUNT。该列的值应为同一客户在当前交易日期之前，最近一次发生的、且交易类型与当前交易类型不同的交易金额。
输入数据结构:
DataFrame df 包含以下列：
- KEY_ID: 客户的唯一标识符。
- TYPE: 交易类型，例如 'Motor' 或 'Tool'。
- DATE: 交易发生的日期。
- AMOUNT: 交易金额。
示例输入数据:
| KEY_ID | TYPE | AMOUNT | DATE |
|---|---|---|---|
| 1 | Motor | 5000 | 2020-01-01 |
| 1 | Tool | 3000 | 2020-02-01 |
| 1 | Tool | 7000 | 2020-03-01 |
| 2 | Tool | 2000 | 2020-01-15 |
| 2 | Motor | 6000 | 2020-02-15 |
| 2 | Tool | 4000 | 2020-03-15 |
期望输出数据:
| KEY_ID | TYPE | AMOUNT | DATE | PREV_AMOUNT |
|---|---|---|---|---|
| 1 | Motor | 5000 | 2020-01-01 | NaN |
| 1 | Tool | 3000 | 2020-02-01 | 5000 |
| 1 | Tool | 7000 | 2020-03-01 | 5000 |
| 2 | Tool | 2000 | 2020-01-15 | NaN |
| 2 | Motor | 6000 | 2020-02-15 | 2000 |
| 2 | Tool | 4000 | 2020-03-15 | 6000 |
2. 常见误区与性能瓶颈
在尝试解决这类问题时，初学者常会遇到一些效率低下的方法，导致性能问题甚至程序崩溃。
2.1 使用 df.apply() 结合行级过滤
一种直观的思路是为每一行应用一个自定义函数，该函数在整个DataFrame中查找符合条件的先前交易。
import pandas as pd
import numpy as np
def find_previous_request(row, full_df):
# 查找同一客户、不同类型、日期早于当前行的交易
previous_requests = full_df[
(full_df['KEY_ID'] == row['KEY_ID']) &
(full_df['TYPE'] != row['TYPE']) &
(full_df['DATE'] < row['DATE'])
]
if not previous_requests.empty:
return previous_requests.iloc[-1]['AMOUNT'] # 返回最近一笔的金额
return np.nan
# 假设df是已排序的原始DataFrame
# df['PREV_AMOUNT'] = df.apply(lambda row: find_previous_request(row, df), axis=1)问题分析: 这种方法对DataFrame中的每一行都会执行一次全局过滤操作，即创建一个新的子DataFrame。当数据量较大时，这种重复的过滤和子DataFrame创建会导致极高的计算开销和内存消耗，从而使程序运行缓慢，甚至引发内核崩溃。df.apply(axis=1) 虽然灵活，但通常应避免在其中进行涉及整个DataFrame的复杂查询。
2.2 使用 groupby().shift()
另一种尝试是利用Pandas的 shift() 函数，它可以在分组内获取前一行的值。
# df['prev_amount'] = df.groupby(['KEY_ID', 'TYPE'])['AMOUNT'].shift().fillna(0).astype(int)
问题分析:shift() 函数确实适用于获取分组内的前一个值。然而，df.groupby(['KEY_ID', 'TYPE'])['AMOUNT'].shift() 的问题在于，它只会在同一客户、同一类型的分组内进行偏移。而我们的需求是查找不同类型的交易金额，因此这种方法无法满足条件。
3. 优化解决方案：基于分组迭代的状态维护
解决此类问题的更高效方法是结合Pandas的分组能力和Python的迭代逻辑，通过在每个客户组内部维护状态变量来实现。
3.1 核心思想
- 预排序: 首先，确保整个DataFrame按 KEY_ID 升序排列，然后在每个 KEY_ID 内部按 DATE 升序排列。这是进行时间序列分析的基础。
- 按客户分组: 使用 df.groupby('KEY_ID') 将DataFrame分割成独立的客户组。
- 组内迭代与状态维护: 对每个客户组，按时间顺序迭代其交易记录。在迭代过程中，为该客户维护一个或多个状态变量，记录不同交易类型的最新金额。
- 条件赋值: 当处理当前交易时，根据其类型，从对应的状态变量中获取所需的不同类型交易金额，并将其赋值给新列。随后，更新当前交易类型的最新金额状态。
3.2 实现步骤
- 导入必要的库 pandas 和 numpy。
- 创建示例DataFrame。
- 对DataFrame进行预排序，确保数据按 KEY_ID 和 DATE 的升序排列。
- 初始化新列 PREV_AMOUNT，默认值为 np.nan。
- 使用 df.groupby('KEY_ID') 按客户ID进行分组。
- 遍历每个客户组，并在组内使用循环（iterrows()）处理每条交易记录。
- 在循环内部，根据交易类型更新和获取状态变量的值，并使用 df.loc 进行高效赋值。
3.3 示例代码
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和日期排序
# reset_index(drop=True) 是为了在后续使用loc时避免索引错乱，尤其是在排序后原始索引被打乱的情况下。
df = df.sort_values(by=['KEY_ID', 'DATE']).reset_index(drop=True)
# 2. 初始化 PREV_AMOUNT 列，默认值为 NaN
df['PREV_AMOUNT'] = np.nan
# 3. 按 KEY_ID 分组并迭代
grouped = df.groupby('KEY_ID')
# 遍历每个客户组
for name, group in grouped:
# 为每个客户初始化状态变量，记录该客户最近一次'Motor'和'Tool'交易的金额
last_motor_amount = np.nan
last_tool_amount = np.nan
# 在当前客户组内，按时间顺序迭代其交易记录
# 由于df已经整体排序，group内部的顺序也是正确的
for ind, row in group.iterrows():
if row[' 
