Python案例如何用Pandas做数据分组验证

wen python案例 2026-07-14 1

本文目录导读：

基础分组验证
数据一致性验证
数据范围和边界验证
多条件分组验证
数据分组验证报告
自定义验证规则

我来介绍几个用Pandas进行数据分组验证的实用案例。

基础分组验证

import pandas as pd
import numpy as np
# 创建示例数据
df = pd.DataFrame({
    '部门': ['技术部', '市场部', '技术部', '人事部', '市场部', '技术部'],
    '员工': ['张三', '李四', '王五', '赵六', '钱七', '孙八'],
    '工资': [15000, 12000, 18000, 10000, 13000, 16000],
    '绩效': ['A', 'B', 'A', 'C', 'B', 'A']
})
# 1. 验证每组数据量（数据完整性检查）
group_counts = df.groupby('部门').size()
print("各部门人数验证：")
print(group_counts)
print(f"总人数验证：{group_counts.sum()} 人\n")
# 2. 验证每组非空值
missing_check = df.groupby('部门').apply(lambda x: x.isnull().sum())
print("缺失值检查：")
print(missing_check)

数据一致性验证

# 创建有问题的数据
df_validate = pd.DataFrame({
    '订单号': ['ORD001', 'ORD001', 'ORD002', 'ORD002', 'ORD003'],
    '产品': ['A', 'B', 'A', 'B', 'A'],
    '数量': [10, 5, 15, 3, 8],
    '金额': [100, 50, 150, 30, 80],
    '状态': ['已完成', '已完成', '处理中', '处理中', '已完成']
})
# 验证同一订单的状态一致性
def check_status_consistency(group):
    """检查组内所有状态是否一致"""
    unique_status = group['状态'].unique()
    return len(unique_status) == 1
status_check = df_validate.groupby('订单号').apply(check_status_consistency)
print("订单状态一致性验证：")
print(status_check)
print(f"状态一致的订单数：{status_check.sum()} / {len(status_check)}\n")
# 验证金额总和与单价关系
def validate_amount(group):
    """验证分组内的金额计算"""
    # 假设单价为10，验证金额 = 数量 * 单价
    expected_amount = group['数量'] * 10
    actual_amount = group['金额']
    is_valid = (expected_amount == actual_amount).all()
    return is_valid
amount_check = df_validate.groupby('订单号').apply(validate_amount)
print("金额计算验证：")
print(amount_check)

数据范围和边界验证

# 创建包含异常值的数据
df_range = pd.DataFrame({
    '地区': ['华东', '华东', '华南', '华南', '华北', '华北'],
    '月份': [1, 2, 1, 2, 1, 2],
    '销售额': [100, 200, 300, 4000, 150, 250],  # 4000可能为异常值
    '温度': [25, 26, 28, 29, -10, 5]  # -10可能为异常值
})
# 验证每组数值范围
def check_value_range(group, column, min_val, max_val):
    """检查指定列的值是否在范围内"""
    values = group[column]
    in_range = (values >= min_val) & (values <= max_val)
    return in_range.all()
# 检查销售额范围（100-500）
sales_check = df_range.groupby('地区').apply(
    lambda x: check_value_range(x, '销售额', 100, 500)
)
print("销售额范围验证：")
print(sales_check)
# 检查温度范围（-5到40度）
temp_check = df_range.groupby('地区').apply(
    lambda x: check_value_range(x, '温度', -5, 40)
)
print("\n温度范围验证：")
print(temp_check)

多条件分组验证

# 创建更复杂的数据
df_complex = pd.DataFrame({
    '项目': ['P1', 'P1', 'P1', 'P2', 'P2', 'P3'],
    '阶段': ['规划', '执行', '完成', '规划', '执行', '规划'],
    '预算': [10000, 5000, 3000, 20000, 8000, 15000],
    '实际支出': [9500, 4800, 2800, 21000, 7500, 16000]
})
def comprehensive_validation(group):
    """综合验证函数"""
    results = {}
    # 1. 验证预算不超支
    total_budget = group['预算'].sum()
    total_expense = group['实际支出'].sum()
    results['预算充足'] = total_expense <= total_budget * 1.1
    # 2. 验证各阶段支出占比合理
    expense_ratio = group['实际支出'] / group['预算']
    results['支出比例合理'] = (expense_ratio >= 0.5).all() and (expense_ratio <= 1.0).all()
    # 3. 验证阶段顺序
    expected_order = {'规划': 1, '执行': 2, '完成': 3}
    phases = group['阶段'].tolist()
    phase_order = [expected_order.get(p, 0) for p in phases]
    results['阶段顺序正确'] = phase_order == sorted(phase_order)
    return pd.Series(results)
validation_results = df_complex.groupby('项目').apply(comprehensive_validation)
print("多条件综合验证结果：")
print(validation_results)

数据分组验证报告

def generate_validation_report(data, group_col, validation_cols):
    """生成分组验证报告"""
    report = {}
    for col in validation_cols:
        col_report = {}
        # 基本统计
        grouped = data.groupby(group_col)[col]
        col_report['均值'] = grouped.mean()
        col_report['标准差'] = grouped.std()
        col_report['最小值'] = grouped.min()
        col_report['最大值'] = grouped.max()
        col_report['缺失值数'] = data.groupby(group_col).apply(
            lambda x: x[col].isnull().sum()
        )
        # 异常检测（使用IQR方法）
        def detect_outliers(group):
            Q1 = group.quantile(0.25)
            Q3 = group.quantile(0.75)
            IQR = Q3 - Q1
            lower_bound = Q1 - 1.5 * IQR
            upper_bound = Q3 + 1.5 * IQR
            outliers = group[(group < lower_bound) | (group > upper_bound)]
            return len(outliers)
        col_report['异常值数'] = data.groupby(group_col)[col].apply(detect_outliers)
        # 数据完整性
        total = data.groupby(group_col).size()
        valid = data.groupby(group_col)[col].count()
        col_report['数据完整率'] = (valid / total * 100).round(2)
        report[col] = pd.DataFrame(col_report)
    return report
# 生成验证报告
validation_report = generate_validation_report(
    df_complex, '项目', ['预算', '实际支出']
)
print("数据分组验证报告：")
for column, report_df in validation_report.items():
    print(f"\n列 '{column}' 验证报告：")
    print(report_df)

自定义验证规则

class DataValidater:
    def __init__(self, df):
        self.df = df
        self.validation_rules = []
        self.results = {}
    def add_rule(self, name, func):
        """添加自定义验证规则"""
        self.validation_rules.append((name, func))
    def validate_by_group(self, group_col):
        """按分组执行所有验证规则"""
        for rule_name, rule_func in self.validation_rules:
            self.results[rule_name] = self.df.groupby(
                group_col
            ).apply(rule_func)
        return pd.DataFrame(self.results)
    def print_summary(self):
        """打印验证总结"""
        for rule, result in self.results.items():
            pass_count = result.sum()
            fail_count = len(result) - pass_count
            print(f"\n规则 '{rule}':")
            print(f"  通过: {pass_count} 组")
            print(f"  失败: {fail_count} 组")
            if fail_count > 0:
                print(f"  失败的组: {result[~result].index.tolist()}")
# 使用示例
validator = DataValidater(df_complex)
# 添加验证规则
validator.add_rule('预算未超支', 
    lambda x: x['实际支出'].sum() <= x['预算'].sum()
)
validator.add_rule('单阶段不超支', 
    lambda x: (x['实际支出'] <= x['预算']).all()
)
# 执行验证
results = validator.validate_by_group('项目')
print("自定义规则验证结果：")
print(results)
validator.print_summary()

这些案例涵盖了：

数据完整性验证
一致性验证
范围验证
多条件验证
报告生成
自定义规则

你可以根据实际需求选择合适的验证方式。