Python案例如何用Pandas做数据窗口函数

wen python案例 2026-07-15 1

本文目录导读：

基础窗口函数
扩展窗口 (expanding)
指数加权窗口 (ewm)
分组窗口函数
实际案例：股票分析
高级窗口函数技巧
性能优化技巧
常见应用场景总结
关键参数说明

我来详细介绍Pandas中窗口函数的用法,包含多个实用案例：

基础窗口函数

1 数据准备

import pandas as pd
import numpy as np
# 创建示例数据
df = pd.DataFrame({
    'date': pd.date_range('2023-01-01', periods=10, freq='D'),
    'sales': [100, 120, 110, 130, 150, 140, 160, 180, 170, 190],
    'product': ['A']*5 + ['B']*5,
    'store': ['X']*10
})
print("原始数据:")
print(df)

2 滑动窗口 (rolling)

# 3天滑动平均
df['rolling_mean_3'] = df['sales'].rolling(window=3).mean()
# 3天滚动求和
df['rolling_sum_3'] = df['sales'].rolling(window=3).sum()
# 自定义窗口函数
df['rolling_custom'] = df['sales'].rolling(window=3).apply(
    lambda x: x.max() - x.min(), 
    raw=False
)
print("滑动窗口结果:")
print(df[['date', 'sales', 'rolling_mean_3', 'rolling_sum_3', 'rolling_custom']])

扩展窗口 (expanding)

# 累计求和
df['cumulative_sum'] = df['sales'].expanding().sum()
# 累计平均值
df['cumulative_mean'] = df['sales'].expanding().mean()
# 累计最大值
df['cumulative_max'] = df['sales'].expanding().max()
# 累计统计量组合
df['cumulative_stats'] = df['sales'].expanding().agg(['mean', 'std', 'min', 'max'])
print("扩展窗口结果:")
print(df[['date', 'sales', 'cumulative_sum', 'cumulative_mean', 'cumulative_max']])

指数加权窗口 (ewm)

# 指数加权移动平均
df['ewm_mean'] = df['sales'].ewm(span=3).mean()
# 指数加权移动标准差
df['ewm_std'] = df['sales'].ewm(span=3).std()
# 调整参数
df['ewm_adjusted'] = df['sales'].ewm(alpha=0.3, adjust=True).mean()
print("指数加权窗口结果:")
print(df[['date', 'sales', 'ewm_mean', 'ewm_std', 'ewm_adjusted']])

分组窗口函数

# 按产品分组计算滚动平均
df_grouped = df.copy()
df_grouped['group_rolling_mean'] = df_grouped.groupby('product')['sales'].transform(
    lambda x: x.rolling(window=3, min_periods=1).mean()
)
# 分组累计求和
df_grouped['group_cumsum'] = df_grouped.groupby('product')['sales'].transform(
    lambda x: x.expanding().sum()
)
print("分组窗口结果:")
print(df_grouped[['date', 'product', 'sales', 'group_rolling_mean', 'group_cumsum']])

实际案例：股票分析

# 创建股票数据
np.random.seed(42)
dates = pd.date_range('2023-01-01', periods=30, freq='D')
stock_data = pd.DataFrame({
    'date': dates,
    'price': np.random.normal(100, 10, 30).cumsum() + 100,
    'volume': np.random.randint(1000, 5000, 30)
})
# 5日均线
stock_data['MA5'] = stock_data['price'].rolling(window=5).mean()
# 20日均线
stock_data['MA20'] = stock_data['price'].rolling(window=20).mean()
# 成交量移动平均
stock_data['volume_MA5'] = stock_data['volume'].rolling(window=5).mean()
# 布林带
stock_data['BB_middle'] = stock_data['price'].rolling(window=20).mean()
stock_data['BB_upper'] = stock_data['BB_middle'] + 2 * stock_data['price'].rolling(window=20).std()
stock_data['BB_lower'] = stock_data['BB_middle'] - 2 * stock_data['price'].rolling(window=20).std()
# 价格变化率
stock_data['pct_change'] = stock_data['price'].pct_change()
# 波动率（20天标准差）
stock_data['volatility'] = stock_data['pct_change'].rolling(window=20).std() * np.sqrt(252)
print("股票分析结果:")
print(stock_data.tail(10))

高级窗口函数技巧

# 创建更多数据
df_advanced = pd.DataFrame({
    'date': pd.date_range('2023-01-01', periods=100, freq='D'),
    'value': np.random.randn(100).cumsum(),
    'category': np.random.choice(['A', 'B', 'C'], 100)
})
# 多窗口统计
windows = [3, 5, 10, 20]
for w in windows:
    df_advanced[f'MA_{w}'] = df_advanced['value'].rolling(window=w).mean()
    df_advanced[f'STD_{w}'] = df_advanced['value'].rolling(window=w).std()
# 条件窗口函数
def conditional_window(series, condition, window=5):
    """根据条件计算窗口统计"""
    mask = condition
    result = pd.Series(index=series.index, dtype=float)
    for i in range(len(series)):
        if mask.iloc[i]:
            start = max(0, i - window + 1)
            result.iloc[i] = series.iloc[start:i+1].mean()
        else:
            result.iloc[i] = np.nan
    return result
# 应用条件窗口
condition = df_advanced['value'] > 0
df_advanced['conditional_mean'] = conditional_window(
    df_advanced['value'], 
    condition
)
print("高级窗口函数结果:")
print(df_advanced.head(20))

性能优化技巧

import time
# 大数据集
big_df = pd.DataFrame({
    'value': np.random.randn(100000),
    'group': np.random.choice(list('ABCDEFGH'), 100000)
})
# 方法1: 直接使用rolling
start = time.time()
result1 = big_df.groupby('group')['value'].transform(
    lambda x: x.rolling(20).mean()
)
print(f"方法1耗时: {time.time() - start:.2f}秒")
# 方法2: 使用numpy加速
start = time.time()
def fast_rolling_mean(series, window):
    """使用numpy实现快速滚动均值"""
    arr = series.to_numpy()
    cumsum = np.cumsum(np.insert(arr, 0, 0))
    result = (cumsum[window:] - cumsum[:-window]) / window
    return np.concatenate([np.full(window-1, np.nan), result])
result2 = big_df.groupby('group')['value'].transform(
    lambda x: fast_rolling_mean(x, 20)
)
print(f"方法2耗时: {time.time() - start:.2f}秒")

常见应用场景总结

窗口类型	常用函数	典型应用
rolling	.mean(), .std(), .sum()	移动平均线，波动率计算
expanding	.sum(), .max(), .min()	累计收益，历史最大值
ewm	.mean(), .std()	指数平滑，动态加权
groupby+rolling	组合使用	分组分析，如按产品、地区