Python数据聚合案例如何汇总统计数据

wen python案例 2026-06-30 277

本文目录导读：

Pandas数据聚合案例
SQL风格聚合（pandasql）
实战案例：电商订单分析
使用字典进行灵活聚合
实际业务统计案例
可视化汇总
最佳实践建议

我来分享几个Python数据聚合的典型案例,涵盖不同场景和常用库。

Pandas数据聚合案例

案例1：销售数据汇总

import pandas as pd
import numpy as np
# 创建示例销售数据
sales_data = pd.DataFrame({
    '日期': ['2024-01-01', '2024-01-01', '2024-01-02', '2024-01-02', '2024-01-03'],
    '城市': ['北京', '上海', '北京', '上海', '广州'],
    '产品': ['A', 'B', 'A', 'B', 'A'],
    '销售额': [1000, 1500, 1200, 1800, 900],
    '数量': [10, 15, 12, 18, 9]
})
# 按城市汇总销售额和数量
city_summary = sales_data.groupby('城市').agg({
    '销售额': ['sum', 'mean', 'max'],
    '数量': ['sum', 'mean']
}).round(2)
print("=== 按城市汇总 ===")
print(city_summary)
print("\n")
# 按日期和城市汇总
daily_city = sales_data.groupby(['日期', '城市']).agg({
    '销售额': 'sum',
    '数量': 'sum'
}).reset_index()
print("=== 按日期和城市汇总 ===")
print(daily_city)

案例2：多维度统计分析

# 创建更复杂的数据集
import pandas as pd
import numpy as np
np.random.seed(42)
df = pd.DataFrame({
    '部门': np.random.choice(['销售部', '市场部', '技术部', '人事部'], 100),
    '级别': np.random.choice(['初级', '中级', '高级'], 100),
    '薪资': np.random.randint(5000, 30000, 100),
    '绩效分': np.random.randint(60, 100, 100),
    '项目数': np.random.randint(1, 10, 100)
})
# 多维度聚合
summary = df.groupby(['部门', '级别']).agg({
    '薪资': ['mean', 'std', 'min', 'max'],
    '绩效分': ['mean', 'std'],
    '项目数': ['sum', 'mean']
}).round(2)
print("=== 多维度统计分析 ===")
print(summary.head(10))

SQL风格聚合（pandasql）

from pandasql import sqldf
import pandas as pd
# 使用SQL语法进行聚合
sales_df = pd.DataFrame({
    'region': ['华北', '华东', '华南', '华北', '华东'],
    'product': ['A', 'A', 'B', 'B', 'A'],
    'revenue': [100, 200, 150, 120, 180],
    'quantity': [10, 20, 15, 12, 18]
})
query = """
SELECT 
    region,
    COUNT(*) as order_count,
    SUM(revenue) as total_revenue,
    AVG(revenue) as avg_revenue,
    SUM(quantity) as total_quantity,
    MAX(revenue) as max_revenue
FROM sales_df
GROUP BY region
ORDER BY total_revenue DESC
"""
result = sqldf(query, locals())
print("=== SQL风格聚合 ===")
print(result)

实战案例：电商订单分析

import pandas as pd
import numpy as np
from datetime import datetime, timedelta
# 生成模拟订单数据
np.random.seed(123)
dates = pd.date_range('2024-01-01', '2024-01-10', freq='H')
orders = pd.DataFrame({
    'order_id': range(1, 1001),
    'customer_id': np.random.randint(1000, 2000, 1000),
    'category': np.random.choice(['电子产品', '服装', '食品', '日用品'], 1000),
    'amount': np.random.uniform(10, 500, 1000).round(2),
    'quantity': np.random.randint(1, 10, 1000),
    'order_date': np.random.choice(dates, 1000)
})
# 1. 按类别汇总销售额
category_sales = orders.groupby('category').agg({
    'amount': ['sum', 'mean', 'count'],
    'quantity': 'sum'
}).round(2)
category_sales.columns = ['总销售额', '平均订单额', '订单数', '总数量']
print("=== 类别销售汇总 ===")
print(category_sales)
print("\n")
# 2. 按日期聚合（每日销售额趋势）
daily_sales = orders.groupby(orders['order_date'].dt.date).agg({
    'amount': 'sum',
    'order_id': 'count'
}).reset_index()
daily_sales.columns = ['日期', '日销售额', '订单数']
print("=== 日销售趋势（前10天）===")
print(daily_sales.head(10))
print("\n")
# 3. RFM分析（客户价值分析）
from datetime import datetime
current_date = max(orders['order_date'])
rfm = orders.groupby('customer_id').agg({
    'order_date': lambda x: (current_date - x.max()).days,  # 最近购买时间
    'order_id': 'count',  # 购买频率
    'amount': 'sum'  # 总消费金额
}).reset_index()
rfm.columns = ['客户ID', '最近购买天数', '购买次数', '总消费']
print("=== RFM客户分析（前10名）===")
print(rfm.head(10))

使用字典进行灵活聚合

# 灵活定义聚合函数
agg_functions = {
    '销售额': ['sum', 'mean', 'std', 'count'],
    '数量': ['sum', 'mean'],
    '利润': lambda x: x.sum() * 0.3  # 自定义函数
}
# 添加利润列
sales_data['利润'] = sales_data['销售额'] * 0.3
# 使用自定义聚合
result = sales_data.groupby('城市').agg(agg_functions)
print("=== 灵活聚合 ===")
print(result)

实际业务统计案例

# 假设这是某公司的员工绩效数据
performance = pd.DataFrame({
    'employee_id': range(1, 101),
    'department': np.random.choice(['技术', '产品', '销售', '运营'], 100),
    'level': np.random.choice(['P3', 'P4', 'P5', 'P6', 'P7'], 100),
    'salary': np.random.randint(15000, 50000, 100),
    'performance_score': np.random.randint(60, 100, 100),
    'projects_completed': np.random.randint(1, 20, 100),
    'satisfaction': np.random.uniform(3, 5, 100).round(2)
})
# 综合绩效分析
def performance_analysis(data):
    """综合绩效分析函数"""
    # 1. 部门维度统计
    dept_stats = data.groupby('department').agg({
        'salary': ['mean', 'median', 'std'],
        'performance_score': ['mean', 'min', 'max'],
        'projects_completed': 'sum',
        'satisfaction': 'mean'
    }).round(2)
    # 2. 等级维度统计
    level_stats = data.groupby('level').agg({
        'salary': 'mean',
        'performance_score': 'mean',
        'projects_completed': 'mean',
        'employee_id': 'count'
    }).round(2)
    # 3. 绩效等级分布（自定义区间）
    def performance_level(score):
        if score >= 90: return 'A'
        elif score >= 80: return 'B'
        elif score >= 70: return 'C'
        else: return 'D'
    data['绩效等级'] = data['performance_score'].apply(performance_level)
    grade_dist = data.groupby('绩效等级').agg({
        'employee_id': 'count',
        'salary': 'mean',
        'projects_completed': 'mean'
    }).round(2)
    return dept_stats, level_stats, grade_dist
dept_stats, level_stats, grade_dist = performance_analysis(performance)
print("=== 部门绩效统计 ===")
print(dept_stats)
print("\n=== 等级绩效统计 ===")
print(level_stats)
print("\n=== 绩效等级分布 ===")
print(grade_dist)

可视化汇总

import matplotlib.pyplot as plt
import seaborn as sns
# 创建可视化数据汇总
fig, axes = plt.subplots(2, 2, figsize=(15, 10))
# 1. 部门薪资分布
dept_salary = performance.groupby('department')['salary'].mean().sort_values()
axes[0, 0].barh(dept_salary.index, dept_salary.values)
axes[0, 0].set_title('部门平均薪资')
# 2. 等级分布
level_count = performance['level'].value_counts()
axes[0, 1].pie(level_count.values, labels=level_count.index, autopct='%1.1f%%')
axes[0, 1].set_title('员工等级分布')
# 3. 绩效与薪资关系
axes[1, 0].scatter(performance['performance_score'], performance['salary'])
axes[1, 0].set_xlabel('绩效分')
axes[1, 0].set_ylabel('薪资')
axes[1, 0].set_title('绩效-薪资关系')
plt.tight_layout()
plt.show()

最佳实践建议

选择合适的方法：
- 简单统计用 describe()
- 分组汇总用 groupby().agg()
- 复杂分析用 pivot_table()
性能优化：
- 使用 agg() 代替多次 apply()
- 大数据集考虑使用 dask 或 polars
数据质量：
- 聚合前先处理缺失值
- 使用 dropna() 或 fillna()

这些案例涵盖了日常工作中80%以上的数据聚合场景，可以根据实际需求灵活调整使用。

本文地址： https://www.bmjup.com/post/12204.html