非唯一多索引Pandas DataFrame条件求和性能优化求助

需求说明 #

有一个含复合重复索引（numerocliente+numerocuenta）的Pandas DataFrame，需按以下规则对cargo列求和：

• 当tipocargo等于'C'且rfcconcepto为空时，求和结果命名为cargofavor；
• 当tipocargo不等于'C'且rfcconcepto不为空时，求和结果命名为cargocontra。

原实现通过遍历唯一索引处理，速度极慢且随数据量增大性能急剧下降，需替换为高效方案。

高效解决方案 #

核心思路是用Pandas矢量化操作+分组聚合替代循环遍历，利用Pandas底层优化的C实现提升性能，避免Python循环的开销。

步骤1：数据预处理 #

# 读取CSV文件（保留原分块读取逻辑）
chunk = pd.read_csv(rutaArchivosCSV + nombreArchivo, index_col=False, usecols=columnasLeer, 
                    low_memory=False, chunksize=1000000)
df_dia = pd.concat(chunk)

# 过滤无效的cargo字符串值，并转为数值类型
df_dia = df_dia[df_dia['cargo'] != 'cargo']
df_dia['cargo'] = pd.to_numeric(df_dia['cargo'], errors='coerce')

# 设置复合索引并排序
df_dia = df_dia.set_index(['numerocliente', 'numerocuenta']).sort_index()

步骤2：条件分组求和 #

方法一：临时列+分组聚合（推荐，性能最优）

import numpy as np

# 生成条件掩码
mask_favor = (df_dia['tipocargo'] == 'C') & df_dia['rfcconcepto'].isna()
mask_contra = (df_dia['tipocargo'] != 'C') & df_dia['rfcconcepto'].notna()

# 为符合条件的cargo值创建临时列，不符合的设为0
df_dia['cargofavor_temp'] = np.where(mask_favor, df_dia['cargo'], 0)
df_dia['cargocontra_temp'] = np.where(mask_contra, df_dia['cargo'], 0)

# 按复合索引分组，直接求和并提取每组第一个numeromovimiento值
df_agg = df_dia.groupby(level=[0, 1]).agg(
    cargofavor=('cargofavor_temp', 'sum'),
    cargocontra=('cargocontra_temp', 'sum'),
    numeromovimiento=('numeromovimiento', 'first')
)

方法二：Lambda聚合（灵活但性能略低）

如果不想创建临时列，可直接在聚合时用lambda过滤：

df_agg = df_dia.groupby(level=[0, 1]).agg(
    cargofavor=('cargo', lambda x: x[(df_dia.loc[x.index, 'tipocargo'] == 'C') & df_dia.loc[x.index, 'rfcconcepto'].isna()].sum()),
    cargocontra=('cargo', lambda x: x[(df_dia.loc[x.index, 'tipocargo'] != 'C') & df_dia.loc[x.index, 'rfcconcepto'].notna()].sum()),
    numeromovimiento=('numeromovimiento', 'first')
)

步骤3：处理特殊规则（numeromovimiento为0的情况） #

# 转换numeromovimiento为字符串并处理空白，判断是否为0
df_agg.loc[df_agg['numeromovimiento'].astype(str).str.strip() == '0', ['cargofavor', 'cargocontra']] = 0

步骤4：合并到结果DataFrame #

# 按月份重命名列
cargoFavorMes = f'cargofavor{consecutivoCol}'
cargoContraMes = f'cargocontra{consecutivoCol}'
df_agg_renamed = df_agg.rename(columns={'cargofavor': cargoFavorMes, 'cargocontra': cargoContraMes})

# 移除不需要的列
df_agg_renamed = df_agg_renamed.drop(['numeromovimiento', 'rfcconcepto', 'tipocargo', 'cargo'], axis=1, errors='ignore')

# 用concat替代append（append已废弃，且concat性能更优）
df_6_meses = pd.concat([df_6_meses, df_agg_renamed])

性能优化说明 #

• 避免循环遍历：Pandas的groupby操作底层基于C实现，比Python循环快10~100倍，数据量越大优势越明显；
• 矢量化操作：np.where和掩码过滤都是矢量化计算，避免逐行处理的开销；
• 替换append：pd.concat批量合并DataFrame，避免每次append创建新对象的内存开销；
• 提前类型转换：将cargo转为数值类型，避免聚合时的类型转换开销。

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