Python案例如何用Pandas做数据分组上线

wen python案例 2026-07-14 163

本文目录导读：

基础数据分组与条件过滤
数据分组统计
分组阈值上线逻辑
条件筛选上市/下市决策
时间序列分组上线
高级分组上线策略
结果可视化和报告
完整上线流程示例

我来详细讲解如何使用Pandas进行数据分组上线的完整案例。

基础数据分组与条件过滤

import pandas as pd
import numpy as np
from datetime import datetime, timedelta
# 创建示例数据
np.random.seed(42)
data = {
    'user_id': [f'user_{i}' for i in range(100)],
    'group': np.random.choice(['A', 'B', 'C'], 100),
    'value': np.random.randint(1, 100, 100),
    'score': np.random.uniform(0, 100, 100),
    'date': pd.date_range('2024-01-01', periods=100, freq='D')
}
df = pd.DataFrame(data)
print("原始数据预览:")
print(df.head())
print("\n数据基本信息:")
print(df.info())

数据分组统计

# 按group分组计算统计量
group_stats = df.groupby('group').agg({
    'value': ['mean', 'std', 'min', 'max', 'count'],
    'score': ['mean', 'median', 'std'],
    'user_id': 'nunique'
})
print("分组统计信息:")
print(group_stats)
# 简化分组统计
simple_stats = df.groupby('group')['value'].agg(['mean', 'std', 'count'])
print("\n简单分组统计:")
print(simple_stats)

分组阈值上线逻辑

def calculate_threshold(group_data, method='percentile', threshold=0.9):
    """
    计算分组阈值
    """
    if method == 'percentile':
        return group_data.quantile(threshold)
    elif method == 'mean_std':
        return group_data.mean() + 1.5 * group_data.std()
    elif method == 'mad':
        median = group_data.median()
        mad = np.median(np.abs(group_data - median))
        return median + 3 * mad
    else:
        return group_data.max() * 0.8
# 应用分组阈值
df['value_threshold'] = df.groupby('group')['value'].transform(
    lambda x: calculate_threshold(x, method='percentile', threshold=0.9)
)
df['score_threshold'] = df.groupby('group')['score'].transform(
    lambda x: calculate_threshold(x, method='mean_std')
)
print("添加阈值后的数据:")
print(df[['user_id', 'group', 'value', 'value_threshold', 'score', 'score_threshold']].head())

条件筛选上市/下市决策

def online_decision(row, thresholds):
    """
    基于多条件的上市/下市决策
    """
    conditions = []
    # 条件1: 值是否超过阈值
    if row['value'] > row['value_threshold']:
        conditions.append('value_high')
    # 条件2: 分数是否超过阈值
    if row['score'] > row['score_threshold']:
        conditions.append('score_high')
    # 条件3: 综合判断
    if len(conditions) >= 2:
        return '上线'
    elif len(conditions) == 1:
        return '观察'
    else:
        return '下线'
# 应用决策逻辑
df['decision'] = df.apply(online_decision, axis=1, args=(None,))
# 分组统计上线情况
online_stats = df.groupby(['group', 'decision']).size().unstack(fill_value=0)
print("\n各分组上线情况:")
print(online_stats)
# 每组上线的比例
group_total = df.groupby('group').size()
online_ratio = df[df['decision'] == '上线'].groupby('group').size() / group_total
print("\n每组上线比例:")
print(online_ratio.round(3))

时间序列分组上线

# 创建时间序列示例
time_data = pd.DataFrame({
    'date': pd.date_range('2024-01-01', periods=365, freq='D'),
    'group': np.random.choice(['A', 'B'], 365),
    'metric': np.random.normal(50, 15, 365) + np.sin(np.arange(365) * 2 * np.pi / 30) * 10
})
# 按周分组统计
time_data['week'] = time_data['date'].dt.isocalendar().week
time_data['month'] = time_data['date'].dt.month
# 计算滚动阈值
def calculate_rolling_threshold(data, window=7):
    """
    计算滚动阈值
    """
    return data.rolling(window=window, min_periods=1).quantile(0.8)
# 按组和日期计算阈值
time_data['rolling_threshold'] = time_data.groupby('group')['metric'].transform(
    lambda x: calculate_rolling_threshold(x)
)
# 上线决策
time_data['online_flag'] = time_data['metric'] > time_data['rolling_threshold']
print("时间序列上线分析:")
print(time_data[time_data['online_flag']].head())

高级分组上线策略

class GroupOnlineStrategy:
    """
    分组上线策略管理器
    """
    def __init__(self, df, group_col, value_cols, methods='auto'):
        self.df = df.copy()
        self.group_col = group_col
        self.value_cols = value_cols
        self.methods = methods
        self.online_results = {}
    def percentile_based_online(self, threshold=0.85):
        """基于百分位数的上线策略"""
        for col in self.value_cols:
            self.df[f'{col}_threshold'] = self.df.groupby(self.group_col)[col].transform(
                lambda x: x.quantile(threshold)
            )
            self.df[f'{col}_online'] = self.df[col] > self.df[f'{col}_threshold']
        return self.df
    def zscore_based_online(self, z_threshold=2):
        """基于Z分数的上线策略"""
        for col in self.value_cols:
            group_stats = self.df.groupby(self.group_col)[col].agg(['mean', 'std'])
            group_stats.columns = ['mean', 'std']
            # 合并回原数据
            self.df = self.df.merge(group_stats, on=self.group_col, suffixes=('', '_stat'))
            # 计算Z分数
            self.df[f'{col}_zscore'] = (self.df[col] - self.df['mean']) / self.df['std']
            self.df[f'{col}_online'] = self.df[f'{col}_zscore'] > z_threshold
            # 删除临时列
            self.df.drop(['mean', 'std'], axis=1, inplace=True)
        return self.df
    def ensemble_online(self, methods=['percentile', 'zscore']):
        """集成多个策略"""
        voters = []
        if 'percentile' in methods:
            result_p = self.percentile_based_online()
            voters.append(result_p[[f'{col}_online' for col in self.value_cols]].any(axis=1))
        if 'zscore' in methods:
            result_z = self.zscore_based_online()
            voters.append(result_z[[f'{col}_online' for col in self.value_cols]].any(axis=1))
        # 多数投票
        if len(voters) == 2:
            self.df['ensemble_online'] = voters[0] & voters[1]  # 两个条件都满足
        else:
            self.df['ensemble_online'] = voters[0] if len(voters) == 1 else False
        return self.df
# 使用策略管理器
strategy = GroupOnlineStrategy(df, 'group', ['value', 'score'])
result = strategy.percentile_based_online(threshold=0.85)
print("\n百分位策略结果:")
print(result.groupby('group')[['value_online', 'score_online']].mean())
# 集成策略
ensemble_result = strategy.ensemble_online(['percentile', 'zscore'])
print("\n集成策略结果:")
print(ensemble_result.groupby('group')['ensemble_online'].mean())

结果可视化和报告

import matplotlib.pyplot as plt
import seaborn as sns
def visualize_online_results(df, group_col, metric_col):
    """
    可视化分组上线结果
    """
    fig, axes = plt.subplots(2, 2, figsize=(14, 10))
    # 1. 各组上线比例
    online_ratio = df.groupby(group_col)[f'{metric_col}_online'].mean()
    axes[0, 0].bar(online_ratio.index, online_ratio.values)
    axes[0, 0].set_title(f'{metric_col} 各分组上线比例')
    axes[0, 0].set_ylabel('上线比例')
    # 2. 阈值分布
    for group in df[group_col].unique():
        group_data = df[df[group_col] == group]
        axes[0, 1].hist(group_data[metric_col], alpha=0.5, label=group, bins=20)
    axes[0, 1].axvline(df[f'{metric_col}_threshold'].mean(), color='r', linestyle='--', label='平均阈值')
    axes[0, 1].set_title(f'{metric_col} 分布与阈值')
    axes[0, 1].legend()
    # 3. 上线 vs 下线对比
    online_data = df[df[f'{metric_col}_online'] == True][metric_col]
    offline_data = df[df[f'{metric_col}_online'] == False][metric_col]
    axes[1, 0].boxplot([online_data, offline_data], labels=['上线', '下线'])
    axes[1, 0].set_title(f'{metric_col} 上线/下线对比')
    # 4. 分组箱线图
    df.boxplot(column=metric_col, by=group_col, ax=axes[1, 1])
    axes[1, 1].set_title(f'{metric_col} 分组箱线图')
    plt.tight_layout()
    plt.show()
# 可视化结果
visualize_online_results(result, 'group', 'value')

完整上线流程示例

def complete_online_workflow(data_path=None):
    """
    完整的分组上线流程
    """
    # 1. 数据准备
    if data_path is None:
        # 创建模拟数据
        np.random.seed(42)
        df = pd.DataFrame({
            'item_id': [f'item_{i}' for i in range(1000)],
            'category': np.random.choice(['电子', '服装', '食品', '家居'], 1000),
            'price': np.random.uniform(10, 1000, 1000),
            'sales': np.random.randint(0, 1000, 1000),
            'rating': np.random.uniform(1, 5, 1000),
            'date': pd.date_range('2024-01-01', periods=1000, freq='H')
        })
    else:
        df = pd.read_csv(data_path)
    print("=== 第1步: 数据概览 ===")
    print(df.describe())
    # 2. 数据预处理
    print("\n=== 第2步: 数据预处理 ===")
    df['log_sales'] = np.log1p(df['sales'])  # 对销售额取对数
    df['price_per_sales'] = df['price'] / (df['sales'] + 1)  # 价格/销售比
    # 3. 分组统计
    print("\n=== 第3步: 分组统计 ===")
    category_stats = df.groupby('category').agg({
        'price': ['mean', 'std', 'quantile', 'max'],
        'sales': ['mean', 'median', 'std'],
        'rating': ['mean', 'std']
    })
    print(category_stats)
    # 4. 计算上线阈值
    print("\n=== 第4步: 计算阈值 ===")
    for metric in ['sales', 'price', 'rating']:
        df[f'{metric}_threshold'] = df.groupby('category')[metric].transform(
            lambda x: x.quantile(0.8)
        )
    # 5. 综合评分
    print("\n=== 第5步: 综合评估 ===")
    df['综合评分'] = (
        (df['sales'] / df['sales_threshold']) * 0.4 +
        (df['price'] / df['price_threshold']) * 0.3 +
        (df['rating'] / df['rating_threshold']) * 0.3
    )
    # 6. 上线决策
    print("\n=== 第6步: 上线决策 ===")
    def decision_logic(row):
        scores = 0
        if row['sales'] > row['sales_threshold']:
            scores += 1
        if row['price'] > row['price_threshold']:
            scores += 1
        if row['rating'] > row['rating_threshold']:
            scores += 1
        if row['综合评分'] > 1:
            scores += 1
        if scores >= 3:
            return '立即上线'
        elif scores == 2:
            return '观察后上线'
        else:
            return '暂缓上线'
    df['online_status'] = df.apply(decision_logic, axis=1)
    # 7. 结果汇总
    print("\n=== 第7步: 结果汇总 ===")
    result_summary = df.groupby(['category', 'online_status']).size().unstack(fill_value=0)
    print(result_summary)
    # 8. 生成上线报告
    print("\n=== 第8步: 上线报告 ===")
    for category in df['category'].unique():
        cat_data = df[df['category'] == category]
        online_count = len(cat_data[cat_data['online_status'] == '立即上线'])
        total_count = len(cat_data)
        print(f"{category}: 建议上線 {online_count}/{total_count} ({online_count/total_count*100:.1f}%)")
    return df
# 执行完整流程
result_df = complete_online_workflow()
print("\n最终上线结果:")
print(result_df[['item_id', 'category', 'online_status', '综合评分']].head(10))

这个案例涵盖了分组上线的完整流程：