为DataFrame中每笔交易添加相反类别上一笔交易金额列

我有一个存储客户交易数据的DataFrame df，每行代表一笔交易，包含以下列：

• KEY_ID：交易客户的ID
• TYPE：交易类型，分为两类
• DATE：交易日期
• AMOUNT：交易金额

需要为该DataFrame新增一列PREV_AMOUNT，存储每笔交易对应相反类别上一笔交易的金额——也就是同一客户当前交易日期前，最后一笔不同类型交易的金额。

尝试过的方法 #

1. 先按KEY_ID和DATE排序（原代码中FECHA为笔误，应为DATE）：

df = df.sort_values(by=['KEY_ID', 'DATE'])

2. 编写逐行查找函数，但应用时内核崩溃：

def find_previous_request(row):
    # 获取相反类型的历史请求
    previous_requests = df[
        (df['KEY_ID'] == row['KEY_ID']) &
        (df['TYPE'] != row['TYPE']) &
        (df['DATE'] < row['DATE'])
    ]
    
    # 若存在历史请求，返回最新的金额
    if not previous_requests.empty:
        return previous_requests.iloc[-1]['AMOUNT']
    
    # 若无历史请求，返回NaN
    return np.nan

3. 尝试分组shift，但结果不符合需求：

df['prev_amount'] = df.groupby(['KEY_ID', 'TYPE'])['AMOUNT'].shift().fillna(0).astype(int)

输入输出示例 #

输入数据 #

期望输出 #

方法一：分组维护最新金额（高效低耗） #

因为交易类型只有两类，按KEY_ID分组后，遍历每组交易时维护一个字典，记录另一类型的最新交易金额，时间复杂度为O(n)，不会出现内存溢出问题。

import pandas as pd
import numpy as np

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

def process_customer_group(group):
    type_latest = {}
    prev_amount_list = []
    for _, row in group.iterrows():
        current_type = row['TYPE']
        # 获取当前类型之外的另一类型
        other_type = (set(group['TYPE'].unique()) - {current_type}).pop()
        # 取另一类型的最新金额，无记录则为NaN
        prev_amount = type_latest.get(other_type, np.nan)
        prev_amount_list.append(prev_amount)
        # 更新当前类型的最新金额为当前交易金额
        type_latest[current_type] = row['AMOUNT']
    group['PREV_AMOUNT'] = prev_amount_list
    return group

# 应用到每个客户分组
df = df.groupby('KEY_ID', group_keys=False).apply(process_customer_group)

方法二：用merge_asof实现（简洁直观） #

利用pandas.merge_asof按日期匹配最近交易的特性，拆分两种类型的数据后分别匹配：

# 先排序
df = df.sort_values(by=['KEY_ID', 'DATE'])

# 拆分两种类型的交易数据
type_a, type_b = df['TYPE'].unique()
df_a = df[df['TYPE'] == type_a].rename(columns={'AMOUNT': 'PREV_AMOUNT'})
df_b = df[df['TYPE'] == type_b].rename(columns={'AMOUNT': 'PREV_AMOUNT'})

# 分别为两种类型匹配另一类型的最近交易
result_a = pd.merge_asof(df[df['TYPE'] == type_b], df_a[['KEY_ID', 'DATE', 'PREV_AMOUNT']],
                        on='DATE', by='KEY_ID', direction='backward')
result_b = pd.merge_asof(df[df['TYPE'] == type_a], df_b[['KEY_ID', 'DATE', 'PREV_AMOUNT']],
                        on='DATE', by='KEY_ID', direction='backward')

# 合并结果并恢复原顺序
final_df = pd.concat([result_a, result_b]).sort_values(by=['KEY_ID', 'DATE']).reset_index(drop=True)

之前方法失效原因 #

• 逐行筛选函数：每一行都要遍历整个DataFrame，时间复杂度为O(n²)，数据量大时会触发内存溢出导致内核崩溃。
• 分组shift代码：是按KEY_ID+TYPE分组取同类型的上一笔交易，和需求的“不同类型”完全不符，因此结果错误。

内容的提问来源于stack exchange，提问作者Gustavo Brieva