Pandas DataFrame逐列校验过滤并收集不合规行的实现方案

逐轮校验时拆分合规与不合规行，将不合规行汇总存入错误表即可，完整实现代码如下：

import pandas as pd

# 你原有数据读取、合并逻辑
vaex_df2 = pd.read_csv('D:\\test\\CR6645DOC\\GMAPS5536C_01092021_01.csv')
vaex_df3 = pd.read_csv('D:\\test\\CR6645DOC\\CM_TM_UCC_2.csv')
df = pd.merge(vaex_df2, vaex_df3,  how='inner', left_on=['Clearing Member PAN','Trading Member PAN','Client PAN'], right_on = ['OFF_TM_PANNO','OFF_TM_PANNO_1','CCD_PAN_NO']).drop_duplicates()
df.drop(['OFF_TM_PANNO','OFF_TM_PANNO_1','CCD_PAN_NO'], axis=1, inplace=True)

# 初始化错误表，新增error_reason字段记录错误原因
errors_df = pd.DataFrame(columns=df.columns.tolist() + ['error_reason'])
NumberRegex = r"^[0-9]\d{1,20}(?:\.\d{1,3})?$" # 原正则的.转义避免匹配任意字符

# 规则1：校验ValueofMTM1列正则匹配
mask = df['ValueofMTM1'].astype(str).str.contains(NumberRegex, regex=True)
# 筛选不符合的行，添加错误原因后加入错误表
bad_rows = df[~mask].copy()
bad_rows['error_reason'] = 'ValueofMTM1格式不符合正则要求'
errors_df = pd.concat([errors_df, bad_rows], ignore_index=True)
# 保留符合规则的行进入下一轮校验
df = df[mask].reset_index(drop=True)

# 规则2：校验TCM_CM_ID等于'M50219'
mask = df['TCM_CM_ID'] == 'M50219'
bad_rows = df[~mask].copy()
bad_rows['error_reason'] = 'TCM_CM_ID取值不等于M50219'
errors_df = pd.concat([errors_df, bad_rows], ignore_index=True)
df = df[mask].reset_index(drop=True)

# 规则3：校验AType属于['P','C']
mask = df['AType'].isin(['P','C'])
bad_rows = df[~mask].copy()
bad_rows['error_reason'] = 'AType取值不在允许范围内'
errors_df = pd.concat([errors_df, bad_rows], ignore_index=True)
df = df[mask].reset_index(drop=True)

# 最终输出：df为全合规数据，errors_df为所有错误行汇总

代码说明 #

• 每轮校验通过布尔掩码拆分合规与不合规行，不合规行追加到错误表后，仅保留合规行进入下一轮校验，符合你要求的「每次校验后移除不符合行」的逻辑
• 新增error_reason字段可直接定位每行错误原因，不需要可自行删除
• 注意核对你实际数据的列名，比如如果MTM列实际名为ValueofMTM，修改代码中对应的列名即可

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