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

wen python案例 2026-07-14 199

本文目录导读：

基础数据准备
基础分组重采样
不同频率的重采样
验证重采样结果
高级验证方法
实际应用案例
数据完整性验证

我来为你详细介绍如何使用Pandas进行数据分组重采样验证的完整案例。

基础数据准备

import pandas as pd
import numpy as np
from datetime import datetime, timedelta
# 创建示例数据
np.random.seed(42)
# 生成时间序列数据
date_rng = pd.date_range(start='2023-01-01', end='2023-01-10', freq='H')
n = len(date_rng)
df = pd.DataFrame({
    'timestamp': date_rng,
    'group': np.random.choice(['A', 'B', 'C'], n),
    'value1': np.random.randn(n) * 10 + 50,
    'value2': np.random.randn(n) * 5 + 20
})
print("原始数据:")
print(df.head())
print(f"数据形状: {df.shape}")

基础分组重采样

# 设置索引为时间列
df.set_index('timestamp', inplace=True)
# 按组进行重采样（每天）
daily_resampled = df.groupby('group').resample('D').agg({
    'value1': ['mean', 'std', 'count'],
    'value2': ['sum', 'min', 'max']
})
print("每日分组重采样结果:")
print(daily_resampled.head(12))

不同频率的重采样

# 按小时重采样（每6小时）
hourly_6h = df.groupby('group').resample('6H').mean()
# 按周重采样
weekly = df.groupby('group').resample('W').agg({
    'value1': 'mean',
    'value2': 'sum'
})
# 按月重采样
monthly = df.groupby('group').resample('M').agg({
    'value1': ['mean', 'std'],
    'value2': ['min', 'max']
})
print("6小时重采样:")
print(hourly_6h.head())
print("\n周重采样:")
print(weekly.head())

验证重采样结果

# 验证函数：检查重采样是否准确
def verify_resample(original_df, resampled_df, group_col='group', time_col='timestamp'):
    """验证重采样结果的准确性"""
    original_df = original_df.copy()
    original_df[time_col] = original_df.index if time_col == 'timestamp' else original_df[time_col]
    results = {}
    for group in resampled_df.index.get_level_values(0).unique():
        group_data = original_df[original_df[group_col] == group]
        # 计算实际均值
        actual_mean = group_data['value1'].mean()
        resampled_mean = resampled_df.xs(group, level=0)['value1'].mean()
        results[group] = {
            'actual_mean': actual_mean,
            'resampled_mean': resampled_mean,
            'difference': abs(actual_mean - resampled_mean),
            'accuracy': (1 - abs(actual_mean - resampled_mean) / actual_mean) * 100
        }
    return pd.DataFrame(results).T
# 执行验证
verification_results = verify_resample(df.reset_index(), daily_resampled)
print("重采样验证结果:")
print(verification_results)

高级验证方法

def advanced_resample_verification(original_df, resampled_df, time_freq='D'):
    """
    高级重采样验证，包括多个统计指标的对比
    """
    original_stats = {}
    resampled_stats = {}
    for group in original_df.index.get_level_values(0).unique():
        org_group = original_df.xs(group, level=0)
        res_group = resampled_df.xs(group, level=0)
        # 计算原始数据统计
        original_stats[group] = {
            'mean': org_group['value1'].mean(),
            'std': org_group['value1'].std(),
            'min': org_group['value1'].min(),
            'max': org_group['value1'].max(),
            'count': len(org_group),
            'sum': org_group['value1'].sum()
        }
        # 计算重采样后统计
        resampled_stats[group] = {
            'mean': res_group['value1']['mean'].mean(),
            'std': res_group['value1']['std'].mean(),
            'min': res_group['value2']['min'].min(),
            'max': res_group['value2']['max'].max(),
            'count': res_group['value1']['count'].sum(),
            'sum': res_group['value2']['sum'].sum()
        }
    # 对比分析
    comparison = pd.DataFrame(original_stats).T
    comparison_resampled = pd.DataFrame(resampled_stats).T
    print(f"{'='*60}")
    print(f"重采样频率: {time_freq}")
    print(f"{'='*60}")
    print("\n原始数据统计:")
    print(comparison)
    print("\n重采样数据统计:")
    print(comparison_resampled)
    # 计算误差
    metrics = ['mean', 'std', 'count']
    print("\n误差分析:")
    for metric in metrics:
        if metric in comparison.columns and metric in comparison_resampled.columns:
            error = abs(comparison[metric] - comparison_resampled[metric])
            print(f"{metric}: 平均误差 = {error.mean():.4f}")
    return comparison, comparison_resampled
# 执行高级验证
orig_stats, res_stats = advanced_resample_verification(df, daily_resampled)

实际应用案例

def sales_data_analysis():
    """销售数据分析案例"""
    # 生成销售数据
    dates = pd.date_range('2023-01-01', '2023-12-31', freq='D')
    sales_data = pd.DataFrame({
        'date': dates,
        'store': np.random.choice(['Store_A', 'Store_B', 'Store_C'], len(dates)),
        'product': np.random.choice(['Product_X', 'Product_Y', 'Product_Z'], len(dates)),
        'sales': np.random.randint(100, 1000, len(dates)),
        'customers': np.random.randint(10, 100, len(dates))
    })
    # 按店铺和产品分组，按月重采样
    sales_data.set_index('date', inplace=True)
    monthly_sales = sales_data.groupby(['store', 'product']).resample('M').agg({
        'sales': ['sum', 'mean', 'count'],
        'customers': ['sum', 'mean']
    })
    print("按月销售分析:")
    print(monthly_sales.head(12))
    # 验证重采样
    def verify_sales_resample(original, resampled):
        """验证销售数据重采样"""
        original_sum = original.groupby(['store', 'product'])['sales'].sum()
        resampled_sum = resampled.groupby(level=[0, 1])['sales']['sum'].sum()
        print(f"\n原始总销售额: {original_sum.sum():,.0f}")
        print(f"重采样总销售额: {resampled_sum.sum():,.0f}")
        print(f"差异: {abs(original_sum.sum() - resampled_sum.sum()):,.0f}")
        # 分组验证
        for store in ['Store_A', 'Store_B', 'Store_C']:
            for product in ['Product_X', 'Product_Y', 'Product_Z']:
                if (store, product) in original_sum.index:
                    orig_val = original_sum.loc[(store, product)]
                    resamp_val = resampled_sum.loc[(store, product)]
                    error = abs(orig_val - resamp_val)
                    print(f"{store}-{product}: 误差 = {error:.2f} ({error/orig_val*100:.2f}%)")
    verify_sales_resample(sales_data, monthly_sales)
# 运行案例
sales_data_analysis()

数据完整性验证

def verify_data_integrity(original_df, resampled_df):
    """验证数据完整性"""
    print("="*50)
    print("数据完整性验证")
    print("="*50)
    # 检查行数
    print(f"\n原始行数: {len(original_df)}")
    print(f"重采样行数: {len(resampled_df['value1'])}")
    # 检查NaN值
    original_nan = original_df.isnull().sum().sum()
    resampled_nan = resampled_df.isnull().sum().sum()
    print(f"原始数据NaN值: {original_nan}")
    print(f"重采样数据NaN值: {resampled_nan}")
    # 检查唯一值
    original_groups = original_df.index.get_level_values(0).nunique()
    resampled_groups = resampled_df.index.get_level_values(0).nunique()
    print(f"原始数据组数: {original_groups}")
    print(f"重采样数据组数: {resampled_groups}")
    # 验证数值范围
    for col in ['value1', 'value2']:
        if col in original_df.columns:
            orig_min = original_df[col].min()
            orig_max = original_df[col].max()
            if col in resampled_df.columns.get_level_values(1):
                res_min = resampled_df[col]['mean'].min()
                res_max = resampled_df[col]['mean'].max()
                print(f"\n{col} 数值范围:")
                print(f"原始: [{orig_min:.2f}, {orig_max:.2f}]")
                print(f"重采样: [{res_min:.2f}, {res_max:.2f}]")
    return True
# 执行完整性验证
verify_data_integrity(df, daily_resampled)

这些案例展示了如何：