Pandas时间窗口函数使用技巧与实例解析助你轻松处理时间数据

引言

在数据分析领域，时间序列数据是一种常见且重要的数据类型。无论是金融市场的股票价格、气象数据中的温度变化，还是网站的用户访问量，这些数据都带有时间戳，并随时间变化。Pandas作为Python数据分析的核心库，提供了强大而灵活的时间窗口函数，使我们能够高效地处理和分析时间序列数据。

时间窗口函数允许我们在数据上应用移动窗口，计算窗口内的统计量，如移动平均、移动标准差等。这些函数在平滑数据、识别趋势、计算波动性等方面非常有用。本文将深入探讨Pandas中的时间窗口函数，通过详细的解释和实例，帮助读者掌握这些工具的使用技巧，轻松应对各种时间序列数据分析任务。

Pandas时间序列基础

在深入了解时间窗口函数之前，我们需要先掌握Pandas中处理时间序列的基础知识。

时间戳和周期

Pandas提供了两种主要的时间相关数据类型：Timestamp和Period。

import pandas as pd import numpy as np # 创建时间戳 timestamp = pd.Timestamp('2023-01-01') print(f"时间戳: {timestamp}") print(f"时间戳类型: {type(timestamp)}") # 创建周期 period = pd.Period('2023-01-01', freq='D') print(f"n周期: {period}") print(f"周期类型: {type(period)}") print(f"周期的开始时间: {period.start_time}") print(f"周期的结束时间: {period.end_time}")

时间索引的创建和操作

在Pandas中，我们可以使用DatetimeIndex或PeriodIndex作为数据的时间索引。

# 创建日期范围 dates = pd.date_range('2023-01-01', periods=10, freq='D') print(f"日期范围:n{dates}") # 创建带有时间索引的Series ts = pd.Series(np.random.randn(10), index=dates) print(f"n时间序列数据:n{ts}") # 时间索引的操作 print(f"n时间序列的年份:n{ts.index.year}") print(f"n时间序列的月份:n{ts.index.month}") print(f"n时间序列的星期几:n{ts.index.dayofweek}")

此外，Pandas还提供了时间重采样、时区转换等功能，这些都是处理时间序列数据的基础。

# 时间重采样 resampled = ts.resample('3D').mean() print(f"按3天重采样后的数据:n{resampled}") # 时区转换 ts_utc = ts.tz_localize('UTC') ts_est = ts_utc.tz_convert('US/Eastern') print(f"nUTC时间:n{ts_utc.index}") print(f"n美国东部时间:n{ts_est.index}")

时间窗口函数概述

Pandas提供了三种主要的时间窗口函数：滚动窗口（rolling）、扩展窗口（expanding）和指数加权窗口（ewm）。这些函数允许我们在数据上应用移动窗口，计算各种统计量。

滚动窗口（rolling）

滚动窗口函数创建一个固定大小的窗口，该窗口在数据上滑动，计算窗口内的统计量。这是最常用的窗口函数。

# 创建示例时间序列数据 np.random.seed(42) dates = pd.date_range('2023-01-01', periods=20, freq='D') data = pd.Series(np.random.randn(20), index=dates) # 计算3天滚动平均 rolling_mean = data.rolling(window=3).mean() print(f"原始数据:n{data.head()}") print(f"n3天滚动平均:n{rolling_mean.head()}")

扩展窗口（expanding）

扩展窗口函数从序列的起始点开始，窗口大小逐渐扩大，直到包含所有数据。

# 计算扩展平均 expanding_mean = data.expanding().mean() print(f"扩展平均:n{expanding_mean.head()}")

指数加权窗口（ewm）

指数加权窗口函数根据指数衰减权重计算统计量，最近的观测值获得更高的权重。

# 计算指数加权平均 ewm_mean = data.ewm(span=3).mean() print(f"指数加权平均:n{ewm_mean.head()}")

滚动窗口函数详解

滚动窗口函数是时间序列分析中最常用的工具之一，它允许我们在固定大小的窗口上计算各种统计量。

基本用法

rolling()函数创建一个滚动窗口对象，我们可以在这个对象上应用各种聚合函数。

# 创建示例数据 np.random.seed(42) dates = pd.date_range('2023-01-01', periods=20, freq='D') data = pd.Series(np.random.randn(20), index=dates) # 创建滚动窗口对象 rolling_obj = data.rolling(window=5) # 计算各种统计量 rolling_mean = rolling_obj.mean() rolling_std = rolling_obj.std() rolling_max = rolling_obj.max() rolling_min = rolling_obj.min() print(f"原始数据:n{data.head(10)}") print(f"n5天滚动平均:n{rolling_mean.head(10)}") print(f"n5天滚动标准差:n{rolling_std.head(10)}") print(f"n5天滚动最大值:n{rolling_max.head(10)}") print(f"n5天滚动最小值:n{rolling_min.head(10)}")

窗口大小设置

窗口大小可以通过window参数设置，它可以是一个整数（表示观测值的数量）或一个字符串（表示时间跨度）。

# 使用整数窗口大小 rolling_int = data.rolling(window=3).mean() # 使用时间跨度窗口大小 rolling_time = data.rolling(window='3D').mean() print(f"整数窗口大小 (3个观测值):n{rolling_int.head()}") print(f"n时间跨度窗口大小 (3天):n{rolling_time.head()}")

不同类型的滚动窗口

Pandas提供了几种不同类型的滚动窗口，可以通过win_type参数指定。

# 矩形窗口（默认） rectangular_window = data.rolling(window=5, win_type='boxcar').mean() # 三角窗口 triangular_window = data.rolling(window=5, win_type='triang').mean() # 高斯窗口 gaussian_window = data.rolling(window=5, win_type='gaussian').mean(std=1) print(f"矩形窗口:n{rectangular_window.head(10)}") print(f"n三角窗口:n{triangular_window.head(10)}") print(f"n高斯窗口:n{gaussian_window.head(10)}")

应用实例

让我们通过一个实际例子来展示滚动窗口函数的应用。假设我们有一组每日股票价格数据，我们想计算5日、10日和20日的移动平均线。

# 创建模拟股票价格数据 np.random.seed(42) dates = pd.date_range('2023-01-01', periods=100, freq='D') price = 100 + np.cumsum(np.random.randn(100) * 0.5) stock_data = pd.Series(price, index=dates) # 计算不同周期的移动平均 ma_5 = stock_data.rolling(window=5).mean() ma_10 = stock_data.rolling(window=10).mean() ma_20 = stock_data.rolling(window=20).mean() # 绘制结果 import matplotlib.pyplot as plt plt.figure(figsize=(12, 6)) plt.plot(stock_data, label='Stock Price') plt.plot(ma_5, label='5-day MA') plt.plot(ma_10, label='10-day MA') plt.plot(ma_20, label='20-day MA') plt.title('Stock Price with Moving Averages') plt.xlabel('Date') plt.ylabel('Price') plt.legend() plt.grid(True) plt.show()

扩展窗口函数详解

扩展窗口函数从序列的起始点开始，窗口大小逐渐扩大，直到包含所有数据。这种函数在计算累积统计量时非常有用。

基本用法

expanding()函数创建一个扩展窗口对象，我们可以在这个对象上应用各种聚合函数。

# 创建示例数据 np.random.seed(42) dates = pd.date_range('2023-01-01', periods=20, freq='D') data = pd.Series(np.random.randn(20), index=dates) # 创建扩展窗口对象 expanding_obj = data.expanding() # 计算各种统计量 expanding_mean = expanding_obj.mean() expanding_std = expanding_obj.std() expanding_max = expanding_obj.max() expanding_min = expanding_obj.min() print(f"原始数据:n{data.head(10)}") print(f"n扩展平均:n{expanding_mean.head(10)}") print(f"n扩展标准差:n{expanding_std.head(10)}") print(f"n扩展最大值:n{expanding_max.head(10)}") print(f"n扩展最小值:n{expanding_min.head(10)}")

应用实例

扩展窗口函数在金融分析中有很多应用，例如计算累计收益、最大回撤等。

# 创建模拟股票价格数据 np.random.seed(42) dates = pd.date_range('2023-01-01', periods=100, freq='D') returns = pd.Series(np.random.randn(100) * 0.02, index=dates) price = 100 * (1 + returns).cumprod() # 计算累计收益 cumulative_returns = (1 + returns).expanding().prod() - 1 # 计算最大回撤 running_max = price.expanding().max() drawdown = (price - running_max) / running_max max_drawdown = drawdown.expanding().min() # 绘制结果 fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(12, 10)) # 价格和累计收益 ax1.plot(price, label='Stock Price') ax1.set_title('Stock Price') ax1.set_ylabel('Price') ax1.grid(True) # 回撤 ax2.fill_between(drawdown.index, drawdown.values, 0, color='red', alpha=0.3) ax2.plot(max_drawdown, label='Maximum Drawdown', color='red') ax2.set_title('Drawdown') ax2.set_ylabel('Drawdown') ax2.grid(True) plt.tight_layout() plt.show()

指数加权窗口函数详解

指数加权窗口函数根据指数衰减权重计算统计量，最近的观测值获得更高的权重。这种函数在需要更多关注近期数据的场景中非常有用。

基本用法

ewm()函数创建一个指数加权窗口对象，我们可以在这个对象上应用各种聚合函数。

# 创建示例数据 np.random.seed(42) dates = pd.date_range('2023-01-01', periods=20, freq='D') data = pd.Series(np.random.randn(20), index=dates) # 创建指数加权窗口对象 ewm_obj = data.ewm(span=5) # 计算各种统计量 ewm_mean = ewm_obj.mean() ewm_std = ewm_obj.std() ewm_var = ewm_obj.var() print(f"原始数据:n{data.head(10)}") print(f"n指数加权平均 (span=5):n{ewm_mean.head(10)}") print(f"n指数加权标准差 (span=5):n{ewm_std.head(10)}") print(f"n指数加权方差 (span=5):n{ewm_var.head(10)}")

指数加权窗口函数有几个重要参数：

span：指定跨度的周期数，相当于窗口大小
com：指定中心质量的周期数，与span的关系为：span = 2*com + 1
halflife：指定权重减半的周期数
alpha：指定平滑因子，直接控制权重衰减速度

# 使用不同参数创建指数加权窗口 ewm_span = data.ewm(span=5).mean() ewm_com = data.ewm(com=2).mean() # 等同于span=5 ewm_halflife = data.ewm(halflife=2).mean() ewm_alpha = data.ewm(alpha=0.3).mean() print(f"使用span=5:n{ewm_span.head()}") print(f"n使用com=2 (等同于span=5):n{ewm_com.head()}") print(f"n使用halflife=2:n{ewm_halflife.head()}") print(f"n使用alpha=0.3:n{ewm_alpha.head()}")

应用实例

指数加权移动平均（EWMA）在金融和时间序列分析中广泛应用，特别是在需要更多关注近期数据的场景。

# 创建模拟股票价格数据 np.random.seed(42) dates = pd.date_range('2023-01-01', periods=100, freq='D') price = 100 + np.cumsum(np.random.randn(100) * 0.5) stock_data = pd.Series(price, index=dates) # 计算简单移动平均和指数加权移动平均 sma_20 = stock_data.rolling(window=20).mean() ewma_20 = stock_data.ewm(span=20).mean() # 绘制结果 plt.figure(figsize=(12, 6)) plt.plot(stock_data, label='Stock Price', alpha=0.5) plt.plot(sma_20, label='20-day SMA') plt.plot(ewma_20, label='20-day EWMA') plt.title('Stock Price with Simple and Exponential Weighted Moving Averages') plt.xlabel('Date') plt.ylabel('Price') plt.legend() plt.grid(True) plt.show()

从图中可以看出，指数加权移动平均（EWMA）比简单移动平均（SMA）对价格变化更敏感，能够更快地反映趋势的变化。

自定义窗口函数

除了使用内置的聚合函数外，Pandas还允许我们使用自定义函数来处理窗口数据。

使用apply方法

apply()方法允许我们对每个窗口应用自定义函数。

# 创建示例数据 np.random.seed(42) dates = pd.date_range('2023-01-01', periods=20, freq='D') data = pd.Series(np.random.randn(20), index=dates) # 定义自定义函数：计算窗口内数据的范围（最大值-最小值） def custom_range(window): return window.max() - window.min() # 应用自定义函数 rolling_range = data.rolling(window=5).apply(custom_range) expanding_range = data.expanding().apply(custom_range) print(f"原始数据:n{data.head(10)}") print(f"n5天滚动范围:n{rolling_range.head(10)}") print(f"n扩展范围:n{expanding_range.head(10)}")

应用实例

自定义窗口函数在实现复杂计算时非常有用。例如，我们可以计算窗口内的线性回归斜率，以识别趋势。

# 定义计算线性回归斜率的函数 def calc_slope(window): x = np.arange(len(window)) y = window.values slope = np.polyfit(x, y, 1)[0] return slope # 创建模拟股票价格数据 np.random.seed(42) dates = pd.date_range('2023-01-01', periods=100, freq='D') price = 100 + np.cumsum(np.random.randn(100) * 0.5) stock_data = pd.Series(price, index=dates) # 计算10天滚动斜率 rolling_slope = stock_data.rolling(window=10).apply(calc_slope) # 绘制结果 fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(12, 10)) # 价格 ax1.plot(stock_data, label='Stock Price') ax1.set_title('Stock Price') ax1.set_ylabel('Price') ax1.grid(True) # 斜率 ax2.plot(rolling_slope, label='10-day Slope', color='orange') ax2.axhline(y=0, color='r', linestyle='--') ax2.set_title('10-day Rolling Slope') ax2.set_ylabel('Slope') ax2.grid(True) plt.tight_layout() plt.show()

时间窗口函数的高级应用

掌握了基础用法后，我们可以探索一些时间窗口函数的高级应用。

多列操作

Pandas的窗口函数不仅适用于Series，也适用于DataFrame。我们可以同时对多列应用窗口操作。

# 创建示例DataFrame np.random.seed(42) dates = pd.date_range('2023-01-01', periods=20, freq='D') data = pd.DataFrame({ 'A': np.random.randn(20), 'B': np.random.randn(20) * 0.5, 'C': np.random.randn(20) * 2 }, index=dates) # 对所有列应用3天滚动平均 rolling_mean = data.rolling(window=3).mean() # 对特定列应用不同的窗口操作 result = pd.DataFrame({ 'A_mean': data['A'].rolling(window=3).mean(), 'B_max': data['B'].rolling(window=5).max(), 'C_std': data['C'].rolling(window=4).std() }) print(f"原始数据:n{data.head()}") print(f"n所有列的3天滚动平均:n{rolling_mean.head()}") print(f"n不同列的不同窗口操作:n{result.head()}")

聚合操作

我们可以使用agg()方法一次性应用多个聚合函数。

# 创建示例数据 np.random.seed(42) dates = pd.date_range('2023-01-01', periods=20, freq='D') data = pd.Series(np.random.randn(20), index=dates) # 应用多个聚合函数 rolling_stats = data.rolling(window=5).agg(['mean', 'std', 'min', 'max']) print(f"原始数据:n{data.head(10)}") print(f"n5天滚动统计量:n{rolling_stats.head(10)}")

时间重采样

时间重采样（resampling）是另一种处理时间序列数据的重要方法，它可以与窗口函数结合使用。

# 创建高频数据 np.random.seed(42) dates = pd.date_range('2023-01-01', periods=240, freq='H') data = pd.Series(np.random.randn(240), index=dates) # 按天重采样并计算每日统计量 daily_stats = data.resample('D').agg(['mean', 'std', 'min', 'max']) # 在重采样数据上应用窗口函数 weekly_mean = daily_stats['mean'].rolling(window=7).mean() print(f"原始数据 (前10小时):n{data.head(10)}") print(f"n每日统计量:n{daily_stats.head()}") print(f"n7天滚动平均 (基于每日数据):n{weekly_mean.head(10)}")

性能优化技巧

处理大规模时间序列数据时，性能优化非常重要。以下是一些优化技巧：

1. 避免循环，使用向量化操作

# 不好的做法：使用循环 def rolling_mean_loop(series, window): result = pd.Series(index=series.index, dtype=float) for i in range(window-1, len(series)): result.iloc[i] = series.iloc[i-window+1:i+1].mean() return result # 好的做法：使用内置的rolling函数 def rolling_mean_vectorized(series, window): return series.rolling(window=window).mean() # 测试性能 np.random.seed(42) large_data = pd.Series(np.random.randn(10000), index=pd.date_range('2023-01-01', periods=10000, freq='D')) import time start = time.time() result_loop = rolling_mean_loop(large_data, 20) time_loop = time.time() - start start = time.time() result_vectorized = rolling_mean_vectorized(large_data, 20) time_vectorized = time.time() - start print(f"循环方法耗时: {time_loop:.4f}秒") print(f"向量化方法耗时: {time_vectorized:.4f}秒") print(f"向量化方法比循环方法快 {time_loop/time_vectorized:.1f}倍")

2. 使用适当的数据类型

# 创建数据 np.random.seed(42) dates = pd.date_range('2023-01-01', periods=100000, freq='D') data_float64 = pd.Series(np.random.randn(100000), index=dates, dtype='float64') data_float32 = pd.Series(np.random.randn(100000), index=dates, dtype='float32') # 测试性能 start = time.time() result_float64 = data_float64.rolling(window=20).mean() time_float64 = time.time() - start start = time.time() result_float32 = data_float32.rolling(window=20).mean() time_float32 = time.time() - start print(f"float64数据类型耗时: {time_float64:.4f}秒") print(f"float32数据类型耗时: {time_float32:.4f}秒") print(f"使用float32比float64快 {time_float64/time_float32:.1f}倍")

3. 使用引擎参数

Pandas 1.0+版本允许我们通过engine参数选择不同的计算引擎。

# 创建数据 np.random.seed(42) large_data = pd.Series(np.random.randn(100000), index=pd.date_range('2023-01-01', periods=100000, freq='D')) # 使用默认引擎 start = time.time() result_default = large_data.rolling(window=20).mean() time_default = time.time() - start # 使用cython引擎 start = time.time() result_cython = large_data.rolling(window=20, engine='cython').mean() time_cython = time.time() - start # 使用numba引擎（如果已安装） try: start = time.time() result_numba = large_data.rolling(window=20, engine='numba').mean() time_numba = time.time() - start print(f"默认引擎耗时: {time_default:.4f}秒") print(f"Cython引擎耗时: {time_cython:.4f}秒") print(f"Numba引擎耗时: {time_numba:.4f}秒") except: print(f"默认引擎耗时: {time_default:.4f}秒") print(f"Cython引擎耗时: {time_cython:.4f}秒") print("Numba引擎不可用")

实战案例：金融数据分析

让我们通过一个完整的金融数据分析案例，综合运用所学的Pandas时间窗口函数。

数据准备

首先，我们获取一些股票数据。这里我们使用yfinance库获取真实的股票数据。

# 安装yfinance库（如果尚未安装） # !pip install yfinance import yfinance as yf # 获取苹果公司(AAPL)的股票数据 aapl_data = yf.download('AAPL', start='2020-01-01', end='2023-01-01') print(f"数据形状: {aapl_data.shape}") print(f"数据列: {aapl_data.columns.tolist()}") print(f"数据预览:n{aapl_data.head()}")

计算技术指标

我们将计算一些常见的技术指标，如移动平均、相对强弱指数(RSI)和布林带。

# 计算移动平均 aapl_data['MA_10'] = aapl_data['Close'].rolling(window=10).mean() aapl_data['MA_30'] = aapl_data['Close'].rolling(window=30).mean() aapl_data['MA_50'] = aapl_data['Close'].rolling(window=50).mean() # 计算相对强弱指数(RSI) def calculate_rsi(data, window=14): delta = data.diff() gain = (delta.where(delta > 0, 0)).rolling(window=window).mean() loss = (-delta.where(delta < 0, 0)).rolling(window=window).mean() rs = gain / loss rsi = 100 - (100 / (1 + rs)) return rsi aapl_data['RSI'] = calculate_rsi(aapl_data['Close']) # 计算布林带 def calculate_bollinger_bands(data, window=20, num_std=2): ma = data.rolling(window=window).mean() std = data.rolling(window=window).std() upper_band = ma + (std * num_std) lower_band = ma - (std * num_std) return upper_band, ma, lower_band aapl_data['BB_Upper'], aapl_data['BB_Middle'], aapl_data['BB_Lower'] = calculate_bollinger_bands(aapl_data['Close']) print(f"计算技术指标后的数据预览:n{aapl_data.tail()}")

生成交易信号

基于计算的技术指标，我们可以生成一些简单的交易信号。

# 生成交易信号 # 1. 移动平均交叉信号 aapl_data['MA_Signal'] = 0 aapl_data.loc[aapl_data['MA_10'] > aapl_data['MA_30'], 'MA_Signal'] = 1 # 买入信号 aapl_data.loc[aapl_data['MA_10'] < aapl_data['MA_30'], 'MA_Signal'] = -1 # 卖出信号 # 2. RSI信号 aapl_data['RSI_Signal'] = 0 aapl_data.loc[aapl_data['RSI'] < 30, 'RSI_Signal'] = 1 # 超卖，买入信号 aapl_data.loc[aapl_data['RSI'] > 70, 'RSI_Signal'] = -1 # 超买，卖出信号 # 3. 布林带信号 aapl_data['BB_Signal'] = 0 aapl_data.loc[aapl_data['Close'] < aapl_data['BB_Lower'], 'BB_Signal'] = 1 # 价格低于下轨，买入信号 aapl_data.loc[aapl_data['Close'] > aapl_data['BB_Upper'], 'BB_Signal'] = -1 # 价格高于上轨，卖出信号 # 综合信号（简单投票） aapl_data['Combined_Signal'] = aapl_data['MA_Signal'] + aapl_data['RSI_Signal'] + aapl_data['BB_Signal'] print(f"带有交易信号的数据预览:n{aapl_data.tail(10)}")

回测策略

让我们基于生成的信号进行简单的策略回测。

# 计算每日收益率 aapl_data['Daily_Return'] = aapl_data['Close'].pct_change() # 计算策略收益率（假设当天信号第二天执行） aapl_data['Strategy_Return'] = aapl_data['Daily_Return'] * aapl_data['Combined_Signal'].shift(1) # 计算累计收益率 aapl_data['Cumulative_Market_Return'] = (1 + aapl_data['Daily_Return']).cumprod() - 1 aapl_data['Cumulative_Strategy_Return'] = (1 + aapl_data['Strategy_Return']).cumprod() - 1 # 计算滚动波动率 aapl_data['Market_Volatility'] = aapl_data['Daily_Return'].rolling(window=30).std() * np.sqrt(252) aapl_data['Strategy_Volatility'] = aapl_data['Strategy_Return'].rolling(window=30).std() * np.sqrt(252) # 计算最大回撤 def calculate_max_drawdown(returns): cumulative = (1 + returns).cumprod() running_max = cumulative.expanding().max() drawdown = (cumulative - running_max) / running_max return drawdown.min() market_max_dd = calculate_max_drawdown(aapl_data['Daily_Return']) strategy_max_dd = calculate_max_drawdown(aapl_data['Strategy_Return']) print(f"市场最大回撤: {market_max_dd:.2%}") print(f"策略最大回撤: {strategy_max_dd:.2%}") # 计算年化收益率和夏普比率 market_annual_return = aapl_data['Cumulative_Market_Return'].iloc[-1] ** (252 / len(aapl_data)) - 1 strategy_annual_return = aapl_data['Cumulative_Strategy_Return'].iloc[-1] ** (252 / len(aapl_data)) - 1 market_sharpe = aapl_data['Daily_Return'].mean() / aapl_data['Daily_Return'].std() * np.sqrt(252) strategy_sharpe = aapl_data['Strategy_Return'].mean() / aapl_data['Strategy_Return'].std() * np.sqrt(252) print(f"n市场年化收益率: {market_annual_return:.2%}") print(f"策略年化收益率: {strategy_annual_return:.2%}") print(f"n市场夏普比率: {market_sharpe:.2f}") print(f"策略夏普比率: {strategy_sharpe:.2f}")

可视化结果

最后，让我们可视化回测结果。

# 绘制累计收益率 plt.figure(figsize=(12, 6)) plt.plot(aapl_data.index, aapl_data['Cumulative_Market_Return'], label='Market Return') plt.plot(aapl_data.index, aapl_data['Cumulative_Strategy_Return'], label='Strategy Return') plt.title('Cumulative Returns') plt.xlabel('Date') plt.ylabel('Cumulative Return') plt.legend() plt.grid(True) plt.show() # 绘制价格和移动平均 plt.figure(figsize=(12, 6)) plt.plot(aapl_data.index, aapl_data['Close'], label='Close Price') plt.plot(aapl_data.index, aapl_data['MA_10'], label='10-day MA') plt.plot(aapl_data.index, aapl_data['MA_30'], label='30-day MA') plt.plot(aapl_data.index, aapl_data['MA_50'], label='50-day MA') plt.title('Stock Price with Moving Averages') plt.xlabel('Date') plt.ylabel('Price') plt.legend() plt.grid(True) plt.show() # 绘制RSI plt.figure(figsize=(12, 6)) plt.plot(aapl_data.index, aapl_data['RSI'], label='RSI') plt.axhline(y=70, color='r', linestyle='--') plt.axhline(y=30, color='g', linestyle='--') plt.title('Relative Strength Index (RSI)') plt.xlabel('Date') plt.ylabel('RSI') plt.legend() plt.grid(True) plt.show() # 绘制布林带 plt.figure(figsize=(12, 6)) plt.plot(aapl_data.index, aapl_data['Close'], label='Close Price') plt.plot(aapl_data.index, aapl_data['BB_Upper'], label='Upper Band', color='r', linestyle='--') plt.plot(aapl_data.index, aapl_data['BB_Middle'], label='Middle Band', color='b') plt.plot(aapl_data.index, aapl_data['BB_Lower'], label='Lower Band', color='g', linestyle='--') plt.title('Bollinger Bands') plt.xlabel('Date') plt.ylabel('Price') plt.legend() plt.grid(True) plt.show()

常见问题与解决方案

在使用Pandas时间窗口函数时，我们可能会遇到一些常见问题。以下是一些问题及其解决方案：

1. 处理缺失值

时间序列数据中常常有缺失值，这可能会影响窗口函数的计算。

# 创建带有缺失值的数据 np.random.seed(42) dates = pd.date_range('2023-01-01', periods=20, freq='D') data = pd.Series(np.random.randn(20), index=dates) data.iloc[[3, 7, 12, 15]] = np.nan print(f"带有缺失值的数据:n{data}") # 尝试计算滚动平均 rolling_mean = data.rolling(window=5).mean() print(f"n直接计算滚动平均:n{rolling_mean}") # 使用min_periods参数 rolling_mean_min_periods = data.rolling(window=5, min_periods=3).mean() print(f"n使用min_periods=3计算滚动平均:n{rolling_mean_min_periods}") # 先填充缺失值再计算 filled_data = data.fillna(method='ffill') rolling_mean_filled = filled_data.rolling(window=5).mean() print(f"n填充缺失值后计算滚动平均:n{rolling_mean_filled}")

2. 处理不规则时间序列

有时候，我们的时间序列可能不是规则的，例如有缺失的日期。

# 创建不规则时间序列 np.random.seed(42) dates = pd.date_range('2023-01-01', periods=20, freq='D') # 随机删除一些日期 missing_dates = np.random.choice(dates, size=5, replace=False) dates = [d for d in dates if d not in missing_dates] data = pd.Series(np.random.randn(len(dates)), index=dates) print(f"不规则时间序列:n{data}") # 使用基于时间的窗口 rolling_time = data.rolling(window='3D').mean() print(f"n使用基于时间的窗口(3天):n{rolling_time}") # 先重新采样再计算 resampled = data.resample('D').mean() filled_resampled = resampled.fillna(method='ffill') rolling_resampled = filled_resampled.rolling(window=3).mean() print(f"n重新采样后计算滚动平均:n{rolling_resampled.head(10)}")

3. 处理不同频率的数据

当处理不同频率的数据时，我们需要确保它们对齐。

# 创建不同频率的数据 np.random.seed(42) daily_dates = pd.date_range('2023-01-01', periods=30, freq='D') daily_data = pd.Series(np.random.randn(30), index=daily_dates) weekly_dates = pd.date_range('2023-01-01', periods=5, freq='W') weekly_data = pd.Series(np.random.randn(5), index=weekly_dates) print(f"每日数据:n{daily_data.head()}") print(f"n每周数据:n{weekly_data}") # 将低频数据重采样到高频 weekly_to_daily = weekly_data.resample('D').ffill() aligned_daily = daily_data.to_frame('daily').join(weekly_to_daily.to_frame('weekly')) print(f"n对齐后的数据:n{aligned_daily.head(10)}") # 在对齐的数据上计算窗口函数 rolling_daily = aligned_daily['daily'].rolling(window=7).mean() rolling_weekly = aligned_daily['weekly'].rolling(window=7).mean() print(f"n每日数据的7天滚动平均:n{rolling_daily.head(10)}") print(f"n每周数据的7天滚动平均:n{rolling_weekly.head(10)}")

4. 处理大数据集

处理大型时间序列数据时，内存和性能可能成为问题。

# 创建大型数据集 np.random.seed(42) large_dates = pd.date_range('2000-01-01', periods=100000, freq='D') large_data = pd.Series(np.random.randn(100000), index=large_dates) # 使用分块处理 def chunked_rolling(data, window, func, chunk_size=10000): results = [] for i in range(0, len(data), chunk_size): chunk = data.iloc[i:i+chunk_size] result = func(chunk.rolling(window=window)) results.append(result) return pd.concat(results) # 测试分块处理 start = time.time() rolling_normal = large_data.rolling(window=30).mean() time_normal = time.time() - start start = time.time() rolling_chunked = chunked_rolling(large_data, window=30, func=lambda x: x.mean()) time_chunked = time.time() - start print(f"正常处理耗时: {time_normal:.4f}秒") print(f"分块处理耗时: {time_chunked:.4f}秒") print(f"结果是否相同: {np.allclose(rolling_normal, rolling_chunked, equal_nan=True)}")