Python版本：Python 3.12+
开发工具：PyCharm 或 VS Code
操作系统：Windows / macOS / Linux (通用)
本章难度：中级到高级
预计学习时间：4-5小时

学习目标

完成本章学习后，你将能够：

  1. 深入理解merge/join底层原理，掌握不同连接类型的适用场景
  2. 掌握groupby内部执行机制，编写高效的分组聚合代码
  3. 运用窗口函数进行高级时间序列分析
  4. 掌握性能优化技巧和内存管理策略
  5. 完成复杂实战项目，解决真实业务问题

1. 数据合并与连接：深入理解merge/join原理

1.1 连接类型图解

理解merge的关键在于理解SQL风格的连接操作。以下是四种基本连接类型的图解：

内连接 (Inner Join)          左连接 (Left Join)
    A        B                   A        B
   / \      / \                 / \      / 
  1---2----2---3               1---2----2
  |   |    |   |               |   |    |
  *---*----*---*               *---*----*
  保留交集                      保留A全部，B无匹配填NaN

右连接 (Right Join)          外连接 (Outer Join)
    A        B                   A        B
     \      / \                 / \      / \
      2----2---3               1---2----2---3
      |    |   |               |   |    |   |
      *----*---*               *---*----*---*
  保留B全部，A无匹配填NaN      保留全部，无匹配填NaN

1.2 merge底层原理

merge操作的核心是**哈希连接（Hash Join）**算法：

步骤1: 构建哈希表
       右表(dept_id)          哈希表
       ┌──────────┐          ┌─────────┬─────────┐
       │dept_id   │    ->    │Hash Key │ Rows    │
       │ 1:技术部  │          │   1     │ [技术部] │
       │ 2:销售部  │          │   2     │ [销售部] │
       │ 3:财务部  │          │   3     │ [财务部] │
       └──────────┘          └─────────┴─────────┘

步骤2: 探测匹配
       左表(emp_id, dept_id)      匹配结果
       ┌─────────┬─────────┐      ┌─────────┬─────────┬─────────┐
       │emp_id   │dept_id  │  ->  │emp_id   │dept_id  │dept_name│
       │ 101     │    1    │      │ 101     │    1    │  技术部  │
       │ 102     │    2    │      │ 102     │    2    │  销售部  │
       │ 103     │    1    │      │ 103     │    1    │  技术部  │
       └─────────┴─────────┘      └─────────┴─────────┴─────────┘

时间复杂度分析：

  • 构建哈希表：O(n)，n为右表行数
  • 探测匹配：O(m)，m为左表行数
  • 总复杂度：O(n + m)

1.3 merge核心参数详解

参数 说明 默认值 使用场景
how 连接方式 ‘inner’ 控制结果保留哪些行
on 连接键 None 两边列名相同时使用
left_on/right_on 左右连接键 None 列名不同时使用
left_index/right_index 使用索引连接 False 基于索引合并
suffixes 重名列后缀 (‘_x’, ‘_y’) 处理列名冲突
validate 关系验证 None 检查数据一致性

1.4 代码示例：merge实战

import pandas as pd
import numpy as np

# 创建示例数据
employees = pd.DataFrame({
    'emp_id': [101, 102, 103, 104, 105],
    'name': ['张三', '李四', '王五', '赵六', '孙七'],
    'dept_id': [1, 2, 1, 3, 2],
    'salary': [8000, 12000, 9500, 15000, 11000]
})

departments = pd.DataFrame({
    'dept_id': [1, 2, 3, 4],
    'dept_name': ['技术部', '销售部', '财务部', '人事部'],
    'location': ['北京', '上海', '广州', '深圳']
})

# 1. 内连接：只保留匹配的数据
inner_result = pd.merge(
    employees, 
    departments, 
    on='dept_id', 
    how='inner'
)
print("内连接结果：")
print(inner_result)

# 2. 左连接：保留左表全部数据
left_result = pd.merge(
    employees, 
    departments, 
    on='dept_id', 
    how='left'
)
print("\n左连接结果：")
print(left_result)

# 3. 使用validate检查数据关系
try:
    # one_to_many: 左表dept_id唯一，右表可重复
    validated = pd.merge(
        employees, 
        departments, 
        on='dept_id', 
        how='left',
        validate='many_to_one'  # employees多对departments一
    )
    print("\n验证通过！")
except Exception as e:
    print(f"\n验证失败：{e}")

1.5 join方法：基于索引的高效合并

join是merge的特例，专用于索引对齐场景，性能更优：

# 设置索引后使用join
emp_idx = employees.set_index('dept_id')
dept_idx = departments.set_index('dept_id')

# join默认是左连接
joined = emp_idx.join(dept_idx, how='left')
print("join结果：")
print(joined)

# 多表链式join
performance = pd.DataFrame({
    'score': [88, 92, 85, 95, 90]
}, index=[101, 102, 103, 104, 105])

# 链式连接
full_data = emp_idx.join(dept_idx).join(performance, on='emp_id')
print("\n链式join结果：")
print(full_data)

merge vs join 选择指南：

场景 推荐方法 原因
基于列连接 merge 灵活，支持多键
基于索引连接 join 语法简洁，性能略优
多表连接 merge 更清晰的可读性
链式操作 join 代码更流畅

1.6 concat方法：轴向堆叠

# 纵向堆叠（增加行）
df1 = pd.DataFrame({'A': [1, 2], 'B': [3, 4]})
df2 = pd.DataFrame({'A': [5, 6], 'B': [7, 8]})

vertical = pd.concat([df1, df2], axis=0, ignore_index=True)
print("纵向堆叠：")
print(vertical)

# 横向堆叠（增加列）
df3 = pd.DataFrame({'C': [9, 10]}, index=[0, 1])
horizontal = pd.concat([df1, df3], axis=1)
print("\n横向堆叠：")
print(horizontal)

# 使用keys创建层次化索引
hierarchical = pd.concat([df1, df2], keys=['第一季度', '第二季度'])
print("\n层次化索引：")
print(hierarchical)

2. 分组聚合深度解析：groupby内部机制

2.1 groupby执行流程图解

groupby采用**“拆分-应用-合并”（Split-Apply-Combine）**模式：

原始数据                    拆分(Split)              应用(Apply)
┌────┬────────┬───────┐     ┌───────────┐           ┌───────────┐
│地区│ 产品   │ 销售额 │     │ 华北组    │           │ sum: 3000 │
├────┼────────┼───────┤     │ ┌───────┐ │           │ mean: 1000│
│华北│ 手机   │ 1000  │     │ │华北   │ │           │ max: 1500 │
│华北│ 电脑   │ 2000  │ ->  │ │华东   │ │  -> agg ->├───────────┤
│华东│ 手机   │ 1500  │     │ │华南   │ │           │ sum: 2700 │
│华东│ 电脑   │ 1200  │     │ └───────┘ │           │ mean: 1350│
│华南│ 手机   │ 1200  │     └───────────┘           │ max: 1500 │
└────┴────────┴───────┘                             └───────────┘
                              按地区分组                        合并结果

2.2 groupby内部数据结构

groupby对象的核心是Grouper和分组字典：

sales_data = pd.DataFrame({
    'region': ['华北', '华北', '华东', '华东', '华南'],
    'product': ['手机', '电脑', '手机', '电脑', '手机'],
    'sales': [1000, 2000, 1500, 1200, 1200]
})

# 创建groupby对象
grouped = sales_data.groupby('region')

# 查看分组信息
print("分组键：", list(grouped.groups.keys()))
print("\n分组字典（索引映射）：")
for key, idx in grouped.groups.items():
    print(f"  {key}: {list(idx)}")

# 迭代分组
print("\n迭代分组对象：")
for name, group in grouped:
    print(f"\n{name}:")
    print(group)

2.3 聚合函数执行机制

聚合类型 方法 返回值形状 适用场景
聚合 agg/aggregate 每个组一个值 统计汇总
转换 transform 与原数据同形状 组内标准化、填充
过滤 filter 筛选后的子集 按组条件筛选
应用 apply 灵活 自定义复杂操作 
# agg：多维度聚合
multi_agg = sales_data.groupby('region').agg({
    'sales': ['sum', 'mean', 'std', 'count'],
    'product': 'nunique'
})
print("多维度聚合：")
print(multi_agg)

# 命名聚合（推荐，结果列名更清晰）
named_agg = sales_data.groupby('region').agg(
    total_sales=('sales', 'sum'),
    avg_sales=('sales', 'mean'),
    product_count=('product', 'nunique')
)
print("\n命名聚合：")
print(named_agg)

# transform：保持原数据结构
sales_data['region_avg'] = sales_data.groupby('region')['sales'].transform('mean')
sales_data['region_ratio'] = sales_data['sales'] / sales_data['region_avg']
print("\ntransform结果：")
print(sales_data)

# filter：基于组特征筛选
high_sales = sales_data.groupby('region').filter(
    lambda x: x['sales'].sum() > 2500
)
print("\nfilter结果（销售额>2500的地区）：")
print(high_sales)

2.4 自定义聚合函数

# 定义自定义聚合函数
def range_calc(x):
    """计算极差"""
    return x.max() - x.min()

def cv(x):
    """计算变异系数"""
    return x.std() / x.mean()

def percentile_90(x):
    """计算90分位数"""
    return x.quantile(0.9)

# 应用自定义函数
custom_result = sales_data.groupby('region')['sales'].agg([
    ('总和', 'sum'),
    ('均值', 'mean'),
    ('极差', range_calc),
    ('变异系数', cv),
    ('P90', percentile_90)
])
print("自定义聚合：")
print(custom_result)

3. 透视表与交叉表

3.1 pivot_table详解

# 创建示例数据
np.random.seed(42)
sales_detail = pd.DataFrame({
    'date': pd.date_range('2024-01-01', periods=100, freq='D'),
    'region': np.random.choice(['华北', '华东', '华南'], 100),
    'product': np.random.choice(['手机', '电脑', '平板'], 100),
    'sales': np.random.randint(500, 3000, 100),
    'profit': np.random.randint(50, 500, 100)
})

# 基础透视表
pivot_basic = pd.pivot_table(
    sales_detail,
    values='sales',
    index='region',
    columns='product',
    aggfunc='sum',
    fill_value=0,
    margins=True,
    margins_name='总计'
)
print("基础透视表：")
print(pivot_basic)

# 多级透视表
sales_detail['month'] = sales_detail['date'].dt.to_period('M')
pivot_multi = pd.pivot_table(
    sales_detail,
    values=['sales', 'profit'],
    index=['region', 'month'],
    columns='product',
    aggfunc={'sales': 'sum', 'profit': 'mean'},
    fill_value=0
)
print("\n多级透视表：")
print(pivot_multi)

3.2 crosstab交叉表

# 创建类别数据
category_data = pd.DataFrame({
    'gender': np.random.choice(['男', '女'], 200),
    'age_group': np.random.choice(['青年', '中年', '老年'], 200),
    'preference': np.random.choice(['线上', '线下'], 200)
})

# 基础交叉表
cross_basic = pd.crosstab(
    category_data['gender'], 
    category_data['age_group']
)
print("基础交叉表：")
print(cross_basic)

# 归一化交叉表
cross_norm = pd.crosstab(
    category_data['gender'], 
    category_data['age_group'],
    normalize='index'  # 行归一化
).round(3)
print("\n归一化交叉表（行百分比）：")
print(cross_norm)

# 三维交叉表
cross_3d = pd.crosstab(
    [category_data['gender'], category_data['age_group']],
    category_data['preference']
)
print("\n三维交叉表：")
print(cross_3d)

4. 时间序列处理与窗口函数

4.1 时间序列基础

# 创建时间序列数据
dates = pd.date_range('2024-01-01', periods=100, freq='D')
ts_data = pd.DataFrame({
    'value': np.random.randn(100).cumsum() + 100
}, index=dates)

# 时间特征提取
ts_data['year'] = ts_data.index.year
ts_data['month'] = ts_data.index.month
ts_data['day'] = ts_data.index.day
ts_data['weekday'] = ts_data.index.weekday
ts_data['quarter'] = ts_data.index.quarter

print("时间特征：")
print(ts_data.head())

# 时间筛选
print("\n2024年1月的数据：")
print(ts_data['2024-01'])

print("\n2024-01-15到2024-01-20的数据：")
print(ts_data['2024-01-15':'2024-01-20'])

4.2 重采样（Resample）

# 创建小时级数据
hours = pd.date_range('2024-01-01', periods=168, freq='H')  # 一周数据
hourly_data = pd.DataFrame({
    'value': np.random.randn(168).cumsum() + 100
}, index=hours)

# 降采样：小时 -> 日
daily = hourly_data.resample('D').agg({
    'value': ['mean', 'max', 'min', 'std']
})
print("日级聚合：")
print(daily.head())

# 常用频率别名
# D: 日历日, W: 周, M: 月末, MS: 月初
# Q: 季末, A: 年末, H: 小时, T: 分钟

# 上采样（增加频率）并插值
upsampled = hourly_data.resample('30T').interpolate(method='linear')
print("\n上采样并插值（前5行）：")
print(upsampled.head())

4.3 窗口函数详解

窗口函数对比：

窗口类型 特点 适用场景
rolling 固定大小滑动窗口 移动平均线、波动率
expanding 逐渐扩大的窗口 累计统计、历史均值
ewm 指数加权，近期权重高 趋势跟踪、风险指标 
# 创建股票价格数据
np.random.seed(42)
days = pd.date_range('2024-01-01', periods=60, freq='D')
stock = pd.DataFrame({
    'price': 100 + np.random.randn(60).cumsum()
}, index=days)

# 1. Rolling滑动窗口
stock['ma_5'] = stock['price'].rolling(window=5).mean()
stock['ma_10'] = stock['price'].rolling(window=10).mean()
stock['volatility'] = stock['price'].rolling(window=10).std()

# 2. Expanding扩展窗口
stock['expanding_mean'] = stock['price'].expanding().mean()
stock['expanding_max'] = stock['price'].expanding().max()

# 3. EWM指数加权移动平均
stock['ewm_5'] = stock['price'].ewm(span=5).mean()
stock['ewm_10'] = stock['price'].ewm(span=10).mean()

print("窗口函数结果（前15行）：")
print(stock.head(15))

# 自定义窗口函数
stock['range_10'] = stock['price'].rolling(window=10).apply(
    lambda x: x.max() - x.min(),
    raw=True  # 使用numpy数组，性能更好
)
print("\n移动极差（前15行）：")
print(stock[['price', 'range_10']].head(15))

5. 性能优化与内存管理

5.1 数据类型优化

# 创建大数据集测试
np.random.seed(42)
large_df = pd.DataFrame({
    'int_col': np.random.randint(0, 100, 1000000),
    'float_col': np.random.randn(1000000),
    'category_col': np.random.choice(['A', 'B', 'C', 'D'], 1000000),
    'bool_col': np.random.choice([True, False], 1000000),
})

print("原始内存使用：")
print(large_df.memory_usage(deep=True))
print(f"总内存: {large_df.memory_usage(deep=True).sum() / 1024**2:.2f} MB")

# 优化数据类型
optimized_df = large_df.copy()

# 1. 整数类型优化
optimized_df['int_col'] = optimized_df['int_col'].astype('int8')  # -128~127

# 2. 浮点数类型优化
optimized_df['float_col'] = optimized_df['float_col'].astype('float32')

# 3. 类别类型优化
optimized_df['category_col'] = optimized_df['category_col'].astype('category')

# 4. 布尔类型已经是bool，无需优化

print("\n优化后内存使用：")
print(optimized_df.memory_usage(deep=True))
print(f"总内存: {optimized_df.memory_usage(deep=True).sum() / 1024**2:.2f} MB")

5.2 类型优化建议表

原始类型 优化策略 适用条件
int64 int8/int16/int32 数值范围确定时
float64 float32 精度要求不高时
object(字符串) category 重复值较多时
bool bool 已经是最佳
datetime64[ns] datetime64[s] 不需要纳秒精度

5.3 向量化操作 vs 循环

import time

df = pd.DataFrame({
    'A': np.random.randn(100000),
    'B': np.random.randn(100000)
})

# 方法1：循环（慢）
start = time.time()
result_loop = []
for i in range(len(df)):
    result_loop.append(df.iloc[i]['A'] + df.iloc[i]['B'])
result_loop = pd.Series(result_loop)
time_loop = time.time() - start

# 方法2：iterrows（较慢）
start = time.time()
result_iterrows = []
for _, row in df.iterrows():
    result_iterrows.append(row['A'] + row['B'])
result_iterrows = pd.Series(result_iterrows)
time_iterrows = time.time() - start

# 方法3：itertuples（较快）
start = time.time()
result_itertuples = [row.A + row.B for row in df.itertuples()]
result_itertuples = pd.Series(result_itertuples)
time_itertuples = time.time() - start

# 方法4：向量化（最快）
start = time.time()
result_vectorized = df['A'] + df['B']
time_vectorized = time.time() - start

print(f"循环耗时: {time_loop:.4f}s")
print(f"iterrows耗时: {time_iterrows:.4f}s")
print(f"itertuples耗时: {time_itertuples:.4f}s")
print(f"向量化耗时: {time_vectorized:.4f}s")
print(f"\n向量化比循环快 {time_loop/time_vectorized:.0f} 倍")

5.4 分块处理大文件

# 模拟大文件处理
# 实际使用时：pd.read_csv('large_file.csv', chunksize=10000)

chunksize = 10000
chunks = []

# 模拟分块读取
for i in range(10):
    chunk = pd.DataFrame({
        'A': np.random.randn(chunksize),
        'B': np.random.randint(0, 100, chunksize)
    })
    # 对每个块进行处理
    chunk['C'] = chunk['A'] * chunk['B']
    chunks.append(chunk)

# 合并结果
result = pd.concat(chunks, ignore_index=True)
print(f"处理完成，总行数: {len(result)}")

# 使用query进行筛选（比布尔索引快）
filtered = result.query('A > 0 and B < 50')
print(f"筛选后行数: {len(filtered)}")

5.5 性能优化检查清单

  • 使用合适的数据类型（category、int32、float32）
  • 避免循环，使用向量化操作
  • 使用eval/query处理复杂条件
  • 分块处理超大文件
  • 使用inplace参数减少内存拷贝
  • 及时删除不再需要的变量（del）
  • 使用numba/Cython加速关键函数

6. 高级索引技巧

6.1 MultiIndex多级索引

# 创建多级索引
arrays = [
    ['华北', '华北', '华东', '华东', '华南', '华南'],
    ['北京', '天津', '上海', '杭州', '广州', '深圳']
]
index = pd.MultiIndex.from_arrays(arrays, names=['region', 'city'])

df_multi = pd.DataFrame({
    'sales': [1000, 800, 1200, 900, 1100, 950],
    'population': [2000, 1500, 2400, 1200, 1800, 1300]
}, index=index)

print("多级索引DataFrame：")
print(df_multi)

# 多级索引选择
print("\n选择华北地区：")
print(df_multi.loc['华北'])

print("\n选择华北-北京：")
print(df_multi.loc[('华北', '北京')])

print("\n使用xs选择所有北京：")
print(df_multi.xs('北京', level='city'))

6.2 索引对齐机制

# 索引对齐示例
df1 = pd.DataFrame({'A': [1, 2, 3]}, index=['a', 'b', 'c'])
df2 = pd.DataFrame({'B': [10, 20, 30]}, index=['b', 'c', 'd'])

print("DataFrame 1：")
print(df1)
print("\nDataFrame 2：")
print(df2)

# 自动对齐
result = df1['A'] + df2['B']
print("\n加法结果（自动对齐）：")
print(result)

# 使用fill_value填充
result_fill = df1['A'].add(df2['B'], fill_value=0)
print("\n加法结果（填充0）：")
print(result_fill)

7. 避坑小贴士

7.1 merge常见陷阱

# 陷阱1：未指定how参数，默认inner导致数据丢失
# 错误：pd.merge(df1, df2, on='key')  # 可能丢失数据
# 正确：pd.merge(df1, df2, on='key', how='outer')  # 保留所有数据

# 陷阱2：连接键数据类型不一致
df1 = pd.DataFrame({'key': [1, 2, 3], 'val': ['a', 'b', 'c']})
df2 = pd.DataFrame({'key': ['1', '2', '3'], 'val2': ['x', 'y', 'z']})

print("数据类型不一致：")
print(f"df1['key']: {df1['key'].dtype}")
print(f"df2['key']: {df2['key'].dtype}")

# 解决：统一数据类型
df2['key'] = df2['key'].astype(int)
merged = pd.merge(df1, df2, on='key')
print("\n合并结果：")
print(merged)

7.2 groupby常见陷阱

# 陷阱：apply中使用循环导致性能问题
# 错误：grouped.apply(lambda x: sum([i**2 for i in x['col']]))
# 正确：grouped['col'].apply(lambda x: (x**2).sum())

# 陷阱：忽略缺失值处理
df_na = pd.DataFrame({
    'group': ['A', 'A', 'B', 'B', None],
    'value': [1, 2, 3, 4, 5]
})

# 默认dropna=True，缺失值被排除
print("默认（排除NA）：")
print(df_na.groupby('group')['value'].sum())

# 保留NA作为独立组
print("\n保留NA：")
print(df_na.groupby('group', dropna=False)['value'].sum())

8. 复杂实战案例：电商销售数据分析

8.1 项目背景

分析某电商平台一年的销售数据，完成以下任务：

  1. 数据整合与清洗
  2. 多维度销售分析
  3. 时间序列趋势分析
  4. RFM用户价值分析
  5. 库存预警模型

8.2 数据准备

np.random.seed(2024)

# 1. 用户数据（10万用户）
users = pd.DataFrame({
    'user_id': range(1, 100001),
    'gender': np.random.choice(['男', '女'], 100000),
    'age_group': np.random.choice(['18-25', '26-35', '36-45', '46+'], 100000),
    'city': np.random.choice(['北京', '上海', '广州', '深圳', '杭州', '成都'], 100000),
    'register_date': pd.date_range('2022-01-01', periods=100000, freq='H')
})

# 2. 商品数据（500个SKU）
products = pd.DataFrame({
    'product_id': range(1, 501),
    'category': np.random.choice(['电子产品', '服装', '食品', '家居', '美妆'], 500),
    'sub_category': np.random.choice(['手机', '电脑', '上衣', '裤子', '零食', '饮料'], 500),
    'price': np.random.randint(50, 5000, 500),
    'cost': np.random.randint(30, 3500, 500),
    'stock': np.random.randint(0, 1000, 500)
})
products['profit_margin'] = (products['price'] - products['cost']) / products['price']

# 3. 订单数据（100万条）
order_dates = pd.date_range('2023-01-01', '2023-12-31', freq='H')
orders = pd.DataFrame({
    'order_id': range(1, 1000001),
    'user_id': np.random.choice(users['user_id'], 1000000),
    'product_id': np.random.choice(products['product_id'], 1000000),
    'quantity': np.random.randint(1, 10, 1000000),
    'order_date': np.random.choice(order_dates, 1000000)
})

print("数据集创建完成！")
print(f"用户数：{len(users):,}")
print(f"商品数：{len(products):,}")
print(f"订单数：{len(orders):,}")

8.3 数据整合

# 合并数据
print("开始数据整合...")

# Step 1: 合并用户数据
orders_user = pd.merge(
    orders, 
    users, 
    on='user_id', 
    how='left'
)

# Step 2: 合并商品数据
full_data = pd.merge(
    orders_user, 
    products, 
    on='product_id', 
    how='left'
)

# 计算业务指标
full_data['order_amount'] = full_data['quantity'] * full_data['price']
full_data['order_profit'] = full_data['quantity'] * (full_data['price'] - full_data['cost'])

# 提取时间特征
full_data['month'] = full_data['order_date'].dt.to_period('M')
full_data['week'] = full_data['order_date'].dt.to_period('W')
full_data['hour'] = full_data['order_date'].dt.hour
full_data['weekday'] = full_data['order_date'].dt.weekday
full_data['is_weekend'] = full_data['weekday'].isin([5, 6])

print(f"整合完成，数据形状：{full_data.shape}")
print(f"\n内存使用：{full_data.memory_usage(deep=True).sum() / 1024**2:.2f} MB")

8.4 多维度销售分析

# 分析1：品类销售分析
category_analysis = full_data.groupby('category').agg(
    total_sales=('order_amount', 'sum'),
    total_profit=('order_profit', 'sum'),
    order_count=('order_id', 'count'),
    unique_users=('user_id', 'nunique'),
    avg_order_value=('order_amount', 'mean')
).round(2)

category_analysis['profit_rate'] = (
    category_analysis['total_profit'] / category_analysis['total_sales']
).round(4)

category_analysis = category_analysis.sort_values('total_sales', ascending=False)
print("品类销售分析：")
print(category_analysis)

# 分析2：城市销售趋势（透视表）
city_monthly = pd.pivot_table(
    full_data,
    values='order_amount',
    index='city',
    columns='month',
    aggfunc='sum',
    fill_value=0
)
print("\n城市月度销售趋势：")
print(city_monthly)

# 分析3：时段分析
hourly_analysis = full_data.groupby('hour').agg(
    sales=('order_amount', 'sum'),
    orders=('order_id', 'count')
)
print("\n时段销售分析：")
print(hourly_analysis.head(10))

8.5 时间序列分析

# 设置时间索引
ts_data = full_data.set_index('order_date')

# 日销售额趋势
daily_sales = ts_data.resample('D')['order_amount'].sum()

# 计算移动平均线
daily_sales_df = daily_sales.to_frame()
daily_sales_df['ma_7'] = daily_sales_df['order_amount'].rolling(window=7).mean()
daily_sales_df['ma_30'] = daily_sales_df['order_amount'].rolling(window=30).mean()

# 计算同比增长（需要两年数据，这里用环比）
daily_sales_df['sales_prev'] = daily_sales_df['order_amount'].shift(1)
daily_sales_df['growth_rate'] = (
    (daily_sales_df['order_amount'] - daily_sales_df['sales_prev']) 
    / daily_sales_df['sales_prev'] * 100
).round(2)

print("日销售趋势（前20天）：")
print(daily_sales_df.head(20))

# 周同比分析
weekly_sales = ts_data.resample('W')['order_amount'].sum()
weekly_sales_df = weekly_sales.to_frame()
weekly_sales_df['yoy'] = weekly_sales_df['order_amount'].shift(52)  # 52周前
weekly_sales_df['yoy_growth'] = (
    (weekly_sales_df['order_amount'] - weekly_sales_df['yoy']) 
    / weekly_sales_df['yoy'] * 100
).round(2)

print("\n周销售趋势：")
print(weekly_sales_df.tail(10))

8.6 RFM用户价值分析

# RFM分析
analysis_date = full_data['order_date'].max()

rfm = full_data.groupby('user_id').agg(
    recency=('order_date', lambda x: (analysis_date - x.max()).days),
    frequency=('order_id', 'count'),
    monetary=('order_amount', 'sum')
)

# RFM评分（使用分位数）
rfm['r_score'] = pd.qcut(rfm['recency'], 5, labels=[5, 4, 3, 2, 1])  # 越小越好
rfm['f_score'] = pd.qcut(rfm['frequency'].rank(method='first'), 5, labels=[1, 2, 3, 4, 5])
rfm['m_score'] = pd.qcut(rfm['monetary'], 5, labels=[1, 2, 3, 4, 5])

# 组合RFM得分
rfm['rfm_score'] = rfm['r_score'].astype(str) + rfm['f_score'].astype(str) + rfm['m_score'].astype(str)

# 用户分群
def segment_user(row):
    if row['rfm_score'] in ['555', '554', '544', '545', '454', '455', '445']:
        return '重要价值客户'
    elif row['rfm_score'] in ['543', '444', '435', '355', '354', '345', '344', '335']:
        return '重要保持客户'
    elif row['rfm_score'] in ['553', '551', '552', '541', '542', '533', '532', '531', '452', '451']:
        return '重要发展客户'
    elif row['rfm_score'] in ['512', '511', '422', '421', '412', '411', '311']:
        return '重要挽留客户'
    elif row['rfm_score'] in ['155', '154', '144', '214', '215', '115', '114']:
        return '一般价值客户'
    else:
        return '低价值客户'

rfm['segment'] = rfm.apply(segment_user, axis=1)

# 分群统计
segment_summary = rfm.groupby('segment').agg(
    user_count=('recency', 'count'),
    avg_recency=('recency', 'mean'),
    avg_frequency=('frequency', 'mean'),
    avg_monetary=('monetary', 'mean')
).round(2)

print("RFM用户分群统计：")
print(segment_summary)

8.7 库存预警模型

# 计算每个商品的日均销量
product_daily_sales = full_data.groupby(['product_id', 'date']).agg(
    daily_sales=('quantity', 'sum')
).reset_index()

product_stats = product_daily_sales.groupby('product_id').agg(
    avg_daily_sales=('daily_sales', 'mean'),
    std_daily_sales=('daily_sales', 'std'),
    max_daily_sales=('daily_sales', 'max')
).reset_index()

# 合并库存信息
inventory = products[['product_id', 'stock', 'category']].merge(
    product_stats, 
    on='product_id', 
    how='left'
)

# 计算库存天数（安全库存 = 7天平均销量）
inventory['stock_days'] = (inventory['stock'] / inventory['avg_daily_sales']).round(1)
inventory['safety_stock'] = inventory['avg_daily_sales'] * 7
inventory['reorder_point'] = inventory['avg_daily_sales'] * 14  # 14天补货点

# 库存预警
def stock_status(row):
    if row['stock'] == 0:
        return '缺货'
    elif row['stock'] <= row['safety_stock']:
        return '库存不足'
    elif row['stock'] <= row['reorder_point']:
        return '需要补货'
    else:
        return '库存充足'

inventory['status'] = inventory.apply(stock_status, axis=1)

# 预警统计
warning_summary = inventory.groupby('status').agg(
    product_count=('product_id', 'count'),
    total_stock=('stock', 'sum')
)

print("库存预警统计：")
print(warning_summary)

print("\n缺货商品列表（前10个）：")
out_of_stock = inventory[inventory['status'] == '缺货'].head(10)
print(out_of_stock[['product_id', 'category', 'stock', 'avg_daily_sales']])

8.8 性能优化实践

# 优化前：内存占用大
print("优化前内存使用：")
print(f"full_data: {full_data.memory_usage(deep=True).sum() / 1024**2:.2f} MB")

# 优化数据类型
optimized_data = full_data.copy()

# 1. 类别型数据优化
optimized_data['gender'] = optimized_data['gender'].astype('category')
optimized_data['age_group'] = optimized_data['age_group'].astype('category')
optimized_data['city'] = optimized_data['city'].astype('category')
optimized_data['category'] = optimized_data['category'].astype('category')
optimized_data['sub_category'] = optimized_data['sub_category'].astype('category')

# 2. 数值型数据优化
optimized_data['quantity'] = optimized_data['quantity'].astype('int8')
optimized_data['hour'] = optimized_data['hour'].astype('int8')
optimized_data['weekday'] = optimized_data['weekday'].astype('int8')

# 3. 布尔型
optimized_data['is_weekend'] = optimized_data['is_weekend'].astype('bool')

print("\n优化后内存使用：")
print(f"optimized_data: {optimized_data.memory_usage(deep=True).sum() / 1024**2:.2f} MB")
print(f"节省内存: {(1 - optimized_data.memory_usage(deep=True).sum() / full_data.memory_usage(deep=True).sum()) * 100:.1f}%")

# 使用query进行高效筛选
high_value_orders = optimized_data.query(
    'order_amount > 1000 and category == "电子产品" and is_weekend'
)
print(f"\n高价值订单数: {len(high_value_orders):,}")

9. 本章小结

核心知识点回顾

主题 关键概念 最佳实践
数据合并 merge/join/concat 根据场景选择，注意连接键类型
分组聚合 split-apply-combine 使用向量化操作，避免循环
透视表 pivot_table/crosstab 合理使用margins和fill_value
时间序列 resample/rolling/ewm 注意数据排序，选择合适窗口
性能优化 数据类型/向量化 先测量后优化，分块处理大文件

进阶学习路径

  1. 深入学习：阅读Pandas官方文档的Categorical和Internals部分
  2. 性能进阶：学习Dask、Polars等大数据处理工具
  3. 实战项目：参与Kaggle竞赛，处理真实数据集
  4. 源码阅读：阅读Pandas源码，理解底层实现

参考资源

  1. Pandas官方文档：https://pandas.pydata.org/docs/
  2. Python for Data Analysis（Wes McKinney著）
  3. Pandas Cookbook（Theodore Petrou著）
  4. Stack Overflow Pandas标签：https://stackoverflow.com/questions/tagged/pandas

学习建议：本章内容较多，建议分多次学习。每学完一个小节，都动手运行示例代码，并尝试修改参数观察结果变化。实战练习部分建议独立完成，再对照参考答案。遇到问题时，善用Pandas官方文档和搜索引擎，培养解决问题的能力。

本教程持续更新中，如有疑问或建议，欢迎在评论区留言交流。

# python # pandas # 开发语言
Logo
AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

2026最新AI Agent高频面试题（分类型精简详解）

本文梳理AI Agent高频面试考点，覆盖基础概念、架构、推理规划、工具调用、记忆、RAG融合、多智能体与工程落地。区分Agent、Workflow与普通工具调用，讲解ReAct、Plan‑and‑Execute等主流范式，分析三级记忆机制与上下文溢出的解决办法，对比传统RAG与Agent‑RAG差异，介绍多智能体协作模式。同时总结落地痛点、优化手段和安全风控要点，划分初/中/高级面试复习重点，可

avatar AI Agent技术社区
cover

深入理解 AI Agent · AGENT #03：从单 Agent 到多 Agent

avatar AI Agent技术社区
cover

AI Agent安全风险启示：工业巡检机器人的AI安全管控

avatar AI Agent技术社区