Pandas 滑動視窗

滑動視窗（Window）操作可以對資料的子集進行計算，常用於時間序列分析，例如計算移動平均、累計值等。

rolling()：滑動視窗

基本用法

import pandas as pd

df = pd.DataFrame({
    'date': pd.date_range('2024-01-01', periods=7),
    'value': [10, 15, 12, 18, 20, 16, 22]
})

# 3 天移動平均
df['ma_3'] = df['value'].rolling(window=3).mean()
print(df)

        date  value      ma_3
0 2024-01-01     10       NaN
1 2024-01-02     15       NaN
2 2024-01-03     12  12.333333
3 2024-01-04     18  15.000000
4 2024-01-05     20  16.666667
5 2024-01-06     16  18.000000
6 2024-01-07     22  19.333333

前兩個值是 NaN，因為不足 3 個資料點。

常用聚合方法

# 移動總和
df['sum_3'] = df['value'].rolling(3).sum()

# 移動最大值
df['max_3'] = df['value'].rolling(3).max()

# 移動最小值
df['min_3'] = df['value'].rolling(3).min()

# 移動標準差
df['std_3'] = df['value'].rolling(3).std()

min_periods 參數

設定最小資料點數：

# 至少有 1 個資料點就計算
df['ma_3'] = df['value'].rolling(window=3, min_periods=1).mean()

        date  value       ma_3
0 2024-01-01     10  10.000000
1 2024-01-02     15  12.500000
2 2024-01-03     12  12.333333
...

center 參數

將結果放在視窗中心：

df['ma_3_center'] = df['value'].rolling(window=3, center=True).mean()

基於時間的視窗

df = df.set_index('date')

# 7 天視窗
df['ma_7d'] = df['value'].rolling('7D').mean()

expanding()：擴展視窗

從第一筆資料開始累計計算：

df = pd.DataFrame({
    'value': [10, 15, 12, 18, 20]
})

# 累計平均
df['expanding_mean'] = df['value'].expanding().mean()
print(df)

   value  expanding_mean
0     10           10.00
1     15           12.50
2     12           12.33
3     18           13.75
4     20           15.00

常用累計方法

df['cumsum'] = df['value'].expanding().sum()   # 累計總和
df['cummax'] = df['value'].expanding().max()   # 累計最大值
df['cummin'] = df['value'].expanding().min()   # 累計最小值
df['cumcount'] = df['value'].expanding().count()  # 累計計數

也可以用內建方法：

df['cumsum'] = df['value'].cumsum()
df['cummax'] = df['value'].cummax()
df['cummin'] = df['value'].cummin()

ewm()：指數加權移動

指數加權移動（Exponentially Weighted Moving）給予近期資料更大的權重：

df = pd.DataFrame({
    'value': [10, 15, 12, 18, 20]
})

# 指數加權移動平均
df['ewm'] = df['value'].ewm(span=3).mean()
print(df)

ewm 參數

# span：指定衰減期
df['ewm'] = df['value'].ewm(span=3).mean()

# alpha：直接指定平滑係數
df['ewm'] = df['value'].ewm(alpha=0.5).mean()

# halflife：半衰期
df['ewm'] = df['value'].ewm(halflife=2).mean()

自訂視窗函數

使用 apply() 套用自訂函數：

def custom_func(x):
    return x.max() - x.min()

df['range'] = df['value'].rolling(3).apply(custom_func)

多欄位操作

df = pd.DataFrame({
    'A': [1, 2, 3, 4, 5],
    'B': [10, 20, 30, 40, 50]
})

# 對所有欄位計算
result = df.rolling(3).mean()

按群組的滑動視窗

df = pd.DataFrame({
    'city': ['Taipei', 'Taipei', 'Taipei', 'Tokyo', 'Tokyo', 'Tokyo'],
    'value': [10, 15, 12, 20, 25, 22]
})

# 每個城市分別計算移動平均
df['ma'] = df.groupby('city')['value'].transform(lambda x: x.rolling(2).mean())

實際應用

股票技術指標

df = pd.DataFrame({
    'date': pd.date_range('2024-01-01', periods=20),
    'close': [100, 102, 101, 103, 105, 104, 106, 108, 107, 110,
              109, 111, 113, 112, 114, 116, 115, 117, 119, 118]
})

# 移動平均線
df['MA5'] = df['close'].rolling(5).mean()
df['MA10'] = df['close'].rolling(10).mean()

# 布林通道
df['MA20'] = df['close'].rolling(20).mean()
df['std20'] = df['close'].rolling(20).std()
df['upper'] = df['MA20'] + 2 * df['std20']
df['lower'] = df['MA20'] - 2 * df['std20']

計算變化率

# 日變化率
df['daily_change'] = df['close'].pct_change()

# 5 日累計變化率
df['change_5d'] = df['close'].pct_change(periods=5)

平滑資料

# 使用指數加權移動平均平滑雜訊
df['smoothed'] = df['value'].ewm(span=5).mean()

異常值偵測

# 使用滑動標準差偵測異常
df['rolling_mean'] = df['value'].rolling(10).mean()
df['rolling_std'] = df['value'].rolling(10).std()
df['z_score'] = (df['value'] - df['rolling_mean']) / df['rolling_std']
df['is_anomaly'] = abs(df['z_score']) > 3