本文目录导读：
我来为你详细介绍如何使用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)
这些案例展示了如何：
- 基础实现：使用
groupby()和resample()进行分组重采样 - 多种聚合方式：支持mean、sum、count、min、max等统计
- 验证方法：通过原始数据对比验证重采样准确性
- 完整性检查：确保数据在重采样过程中没有丢失或错误
在实际应用中，你可以根据具体需求调整重采样频率、聚合函数和验证标准。
