本文目录导读：
我来详细讲解如何使用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))
这个案例涵盖了分组上线的完整流程：
- 基础数据分组统计
- 多种阈值计算方法
- 条件筛选和决策逻辑
- 时间序列数据分析
- 高级策略管理
- 可视化和报告生成
- 完整的上线工作流
你可以根据实际业务需求调整阈值、权重和决策规则。
