探索Python Pandas时间处理函数的强大功能从基础到进阶的时间序列数据分析技巧

引言

时间序列数据是数据分析中最常见的数据类型之一，无论是金融市场的股票价格、气象数据的温度变化，还是网站访问量的日统计，都涉及到时间序列的分析。Python Pandas库提供了强大而灵活的时间处理功能，使得时间序列数据的操作和分析变得简单高效。本文将带领读者从基础到进阶，全面探索Pandas中时间处理函数的强大功能，并通过实际案例展示如何应用这些技巧解决实际问题。

1. 基础时间数据处理

1.1 时间数据类型的创建和转换

Pandas提供了两种主要的时间数据类型：Timestamp（表示单个时间点）和DatetimeIndex（表示时间序列）。让我们首先了解如何创建和转换这些时间数据类型。

import pandas as pd import numpy as np # 创建单个时间点 ts = pd.Timestamp('2023-01-01') print("单个时间点:", ts) print("时间类型:", type(ts)) # 从字符串创建时间序列 dates = ['2023-01-01', '2023-01-02', '2023-01-03', '2023-01-04'] dt_index = pd.DatetimeIndex(dates) print("n时间序列索引:", dt_index) # 将DataFrame的列转换为时间类型 df = pd.DataFrame({'date': ['2023-01-01', '2023-01-02', '2023-01-03'], 'value': [10, 20, 30]}) df['date'] = pd.to_datetime(df['date']) print("n转换后的DataFrame:") print(df) print("日期列的类型:", df['date'].dtype) # 从其他格式转换时间 # 从Unix时间戳转换 unix_timestamp = 1672531200 # 2023-01-01 00:00:00 UTC ts_from_unix = pd.to_datetime(unix_timestamp, unit='s') print("n从Unix时间戳转换:", ts_from_unix) # 从Excel日期序列号转换 excel_date = 44927 # 2023-01-01 in Excel date system ts_from_excel = pd.to_datetime(excel_date, unit='D', origin='1899-12-30') print("从Excel日期序列号转换:", ts_from_excel)

1.2 时间索引的设置和使用

在时间序列分析中，将时间列设置为索引是非常常见的操作，这样可以方便地进行时间切片和聚合。

# 创建带有时间索引的DataFrame date_rng = pd.date_range(start='2023-01-01', end='2023-01-10', freq='D') df = pd.DataFrame(date_rng, columns=['date']) df['data'] = np.random.randint(0, 100, size=(len(date_rng))) df.set_index('date', inplace=True) print("带有时间索引的DataFrame:") print(df) # 获取时间索引的各个部分 df['year'] = df.index.year df['month'] = df.index.month df['day'] = df.index.day df['dayofweek'] = df.index.dayofweek # 周一为0，周日为6 df['hour'] = df.index.hour df['minute'] = df.index.minute df['second'] = df.index.second df['quarter'] = df.index.quarter # 季度 print("n添加时间组件后的DataFrame:") print(df) # 使用时间组件进行筛选 print("n筛选工作日（周一到周五）:") print(df[df['dayofweek'] < 5]) # 0-4为周一到周五

1.3 基本的时间筛选和切片操作

Pandas提供了多种方式来筛选和切片时间序列数据，这些操作在时间序列分析中非常实用。

# 扩展数据集以包含更多时间点 extended_date_rng = pd.date_range(start='2023-01-01', end='2023-03-31', freq='D') extended_df = pd.DataFrame(extended_date_rng, columns=['date']) extended_df['value'] = np.random.randint(0, 100, size=(len(extended_date_rng))) extended_df.set_index('date', inplace=True) # 按日期范围切片 print("2023年1月的数据:") print(extended_df['2023-01']) print("n2023年1月15日到2月15日的数据:") print(extended_df['2023-01-15':'2023-02-15']) # 使用truncate方法进行切片 print("n使用truncate方法获取2023年2月的数据:") print(extended_df.truncate(before='2023-02-01', after='2023-02-28')) # 按条件筛选 print("n筛选所有周一的数据:") print(extended_df[extended_df.index.dayofweek == 0]) # 0表示周一 # 使用between_time方法筛选一天中的特定时间段 # 首先创建一个包含小时和分钟信息的DataFrame hourly_df = pd.DataFrame({ 'time': pd.date_range(start='2023-01-01', end='2023-01-03', freq='H'), 'value': np.random.randint(0, 100, size=(48)) }) hourly_df.set_index('time', inplace=True) print("n筛选每天9点到17点之间的数据:") print(hourly_df.between_time('09:00', '17:00'))

2. 中级时间序列操作

2.1 时间重采样（Resample）

时间重采样是将时间序列数据从一个频率转换到另一个频率的过程，例如从日数据转换为月数据。Pandas的resample()方法使这一过程变得简单。

# 创建一个包含分钟级数据的DataFrame minute_df = pd.DataFrame({ 'time': pd.date_range(start='2023-01-01', end='2023-01-10', freq='T'), 'value': np.random.randint(0, 100, size=(60*24*10)) # 10天的分钟数据 }) minute_df.set_index('time', inplace=True) # 将分钟数据重采样为小时数据 hourly_resampled = minute_df.resample('H').mean() # 计算每小时的平均值 print("重采样为小时数据（平均值）:") print(hourly_resampled.head()) # 使用不同的聚合函数 print("n重采样为小时数据（总和）:") print(minute_df.resample('H').sum().head()) print("n重采样为小时数据（最大值）:") print(minute_df.resample('H').max().head()) # 重采样为日数据 daily_resampled = minute_df.resample('D').agg(['mean', 'std', 'min', 'max']) print("n重采样为日数据（多种统计量）:") print(daily_resampled.head()) # OHLC重采样（常用于金融数据） ohlc_resampled = minute_df['value'].resample('D').agg( open='first', high='max', low='min', close='last' ) print("nOHLC重采样:") print(ohlc_resampled.head()) # 上采样和填充 # 从日数据上采样到小时数据 daily_sample_df = pd.DataFrame({ 'date': pd.date_range(start='2023-01-01', end='2023-01-10', freq='D'), 'value': np.random.randint(0, 100, size=(10)) }) daily_sample_df.set_index('date', inplace=True) # 上采样到小时数据并填充缺失值 upsampled = daily_sample_df.resample('H').asfreq() # 默认填充NaN print("n上采样到小时数据（不填充）:") print(upsampled.head(24)) # 使用不同的填充方法 ffilled = daily_sample_df.resample('H').ffill() # 前向填充 bfilled = daily_sample_df.resample('H').bfill() # 后向填充 interpolated = daily_sample_df.resample('H').interpolate() # 插值 print("n前向填充:") print(ffilled.head(24)) print("n插值填充:") print(interpolated.head(24))

2.2 时间窗口操作（Rolling）

时间窗口操作允许我们在一个滑动的时间窗口上计算统计量，这对于平滑时间序列数据和计算移动平均等指标非常有用。

# 创建一个包含随机波动的时间序列 np.random.seed(42) window_df = pd.DataFrame({ 'date': pd.date_range(start='2023-01-01', end='2023-03-31', freq='D'), 'value': np.random.normal(50, 10, size=(90)) # 均值为50，标准差为10 }) window_df.set_index('date', inplace=True) # 计算7天移动平均 window_df['7_day_mean'] = window_df['value'].rolling(window=7).mean() window_df['7_day_std'] = window_df['value'].rolling(window=7).std() window_df['7_day_min'] = window_df['value'].rolling(window=7).min() window_df['7_day_max'] = window_df['value'].rolling(window=7).max() print("带有移动统计量的DataFrame:") print(window_df.head(10)) # 计算指数加权移动平均 window_df['ewm_mean'] = window_df['value'].ewm(span=7).mean() print("n带有指数加权移动平均的DataFrame:") print(window_df[['value', '7_day_mean', 'ewm_mean']].head(10)) # 使用rolling进行自定义计算 # 例如，计算窗口内的最大值和最小值的差值 window_df['range'] = window_df['value'].rolling(window=7).apply(lambda x: x.max() - x.min()) print("n带有自定义计算的DataFrame:") print(window_df[['value', 'range']].head(10)) # 使用不同类型的窗口 # 扩展窗口（expanding）：从开始到当前点的所有数据 window_df['expanding_mean'] = window_df['value'].expanding().mean() print("n带有扩展窗口平均的DataFrame:") print(window_df[['value', 'expanding_mean']].head(10)) # 累积求和 window_df['cumsum'] = window_df['value'].cumsum() print("n带有累积求和的DataFrame:") print(window_df[['value', 'cumsum']].head(10))

2.3 时间Shift和差分操作

Shift和差分操作是时间序列分析中的重要工具，它们可以帮助我们比较不同时间点的数据，以及计算数据的变化率。

# 创建一个示例DataFrame shift_df = pd.DataFrame({ 'date': pd.date_range(start='2023-01-01', end='2023-01-31', freq='D'), 'value': np.random.randint(0, 100, size=(31)) }) shift_df.set_index('date', inplace=True) # 向前和向后shift shift_df['shifted_forward'] = shift_df['value'].shift(1) # 向前移动一天 shift_df['shifted_backward'] = shift_df['value'].shift(-1) # 向后移动一天 print("带有shift操作的DataFrame:") print(shift_df.head()) # 计算日变化量 shift_df['daily_change'] = shift_df['value'] - shift_df['shifted_forward'] print("n带有日变化量的DataFrame:") print(shift_df.head()) # 计算百分比变化 shift_df['pct_change'] = shift_df['value'].pct_change() * 100 print("n带有百分比变化的DataFrame:") print(shift_df.head()) # 计算差分 shift_df['diff'] = shift_df['value'].diff() # 等同于 daily_change print("n带有差分的DataFrame:") print(shift_df.head()) # 计算多阶差分 shift_df['second_diff'] = shift_df['value'].diff(periods=2) print("n带有二阶差分的DataFrame:") print(shift_df.head()) # 计算对数收益率（常用于金融分析） shift_df['log_return'] = np.log(shift_df['value'] / shift_df['value'].shift(1)) print("n带有对数收益率的DataFrame:") print(shift_df.head()) # 计算移动窗口的百分比变化 shift_df['rolling_pct_change'] = shift_df['value'].pct_change(periods=7) * 100 print("n带有7天窗口百分比变化的DataFrame:") print(shift_df.head(10))

3. 高级时间序列处理

3.1 时区处理

在处理全球数据时，时区处理是一个重要的问题。Pandas提供了强大的时区处理功能，可以轻松地在不同时区之间转换。

# 创建一个无时区的时间序列 naive_dates = pd.date_range(start='2023-01-01', periods=10, freq='D') print("无时区的时间序列:") print(naive_dates) # 本地化时区 localized_dates = naive_dates.tz_localize('UTC') # 指定为UTC时区 print("n本地化为UTC时区:") print(localized_dates) # 转换为其他时区 us_eastern = localized_dates.tz_convert('US/Eastern') print("n转换为美国东部时区:") print(us_eastern) # 创建一个带有时区的DataFrame tz_df = pd.DataFrame({ 'time': pd.date_range(start='2023-01-01', periods=24, freq='H', tz='UTC'), 'value': np.random.randint(0, 100, size=(24)) }) print("n带有时区的DataFrame:") print(tz_df.head()) # 转换整个DataFrame的时区 tz_df['time'] = tz_df['time'].dt.tz_convert('Asia/Shanghai') print("n转换为上海时区:") print(tz_df.head()) # 处理不同时区的数据 # 创建两个不同时区的时间序列 utc_times = pd.date_range(start='2023-01-01', periods=5, freq='D', tz='UTC') eastern_times = pd.date_range(start='2023-01-01', periods=5, freq='D', tz='US/Eastern') # 转换为同一时区进行比较 eastern_to_utc = eastern_times.tz_convert('UTC') print("n美国东部时间转换为UTC:") print(eastern_to_utc) # 处理夏令时 dst_dates = pd.date_range(start='2023-03-10', periods=4, freq='D', tz='US/Eastern') print("n包含夏令时转换的日期:") print(dst_dates) # 创建一个跨越夏令时转换的时间序列 dst_hourly = pd.date_range(start='2023-03-11 22:00', periods=6, freq='H', tz='US/Eastern') print("n跨越夏令时转换的小时数据:") print(dst_hourly)

3.2 时间周期和时期

Pandas中的Period对象表示固定的时间段，如某一天、某个月或某一年，这在处理周期性数据时非常有用。

# 创建Period对象 p1 = pd.Period('2023-01', freq='M') # 2023年1月 p2 = pd.Period('2023-01-01', freq='D') # 2023年1月1日 print("月度Period:", p1) print("日度Period:", p2) # Period的运算 print("nPeriod运算:") print("下一个月:", p1 + 1) print("上一个月:", p1 - 1) # 创建PeriodIndex periods = pd.period_range(start='2023-01', end='2023-12', freq='M') print("n月度PeriodIndex:") print(periods) # 创建带有PeriodIndex的DataFrame period_df = pd.DataFrame({ 'period': pd.period_range(start='2023-01', end='2023-06', freq='M'), 'value': np.random.randint(0, 100, size=(6)) }) period_df.set_index('period', inplace=True) print("n带有PeriodIndex的DataFrame:") print(period_df) # 将时间戳转换为Period timestamp_df = pd.DataFrame({ 'date': pd.date_range(start='2023-01-01', end='2023-06-30', freq='M'), 'value': np.random.randint(0, 100, size=(6)) }) timestamp_df['period'] = timestamp_df['date'].dt.to_period('M') print("n将时间戳转换为Period:") print(timestamp_df) # 将Period转换为时间戳 print("n将Period转换为时间戳:") print(period_df.index.to_timestamp()) # 使用Period进行时间范围查询 print("n查询2023年第一季度的数据:") print(period_df['2023-Q1']) # Q1表示第一季度 # 按季度聚合数据 quarterly_df = period_df.resample('Q').sum() print("n按季度聚合的数据:") print(quarterly_df)

3.3 自定义时间频率和偏移量

Pandas提供了灵活的方式来定义自定义的时间频率和偏移量，这对于处理非标准时间间隔的数据非常有用。

# 使用标准频率 print("标准频率示例:") print(pd.date_range(start='2023-01-01', periods=5, freq='D')) # 日 print(pd.date_range(start='2023-01-01', periods=5, freq='W')) # 周 print(pd.date_range(start='2023-01-01', periods=5, freq='M')) # 月 print(pd.date_range(start='2023-01-01', periods=5, freq='Q')) # 季度 print(pd.date_range(start='2023-01-01', periods=5, freq='A')) # 年 # 使用自定义频率 print("n自定义频率示例:") print(pd.date_range(start='2023-01-01', periods=5, freq='2D')) # 每2天 print(pd.date_range(start='2023-01-01', periods=5, freq='3H')) # 每3小时 print(pd.date_range(start='2023-01-01', periods=5, freq='2W')) # 每2周 # 使用组合频率 print("n组合频率示例:") print(pd.date_range(start='2023-01-01', periods=5, freq='W-MON')) # 每周一 print(pd.date_range(start='2023-01-01', periods=5, freq='MS')) # 每月第一天 print(pd.date_range(start='2023-01-01', periods=5, freq='BMS')) # 每月第一个工作日 # 使用DateOffset对象 print("nDateOffset示例:") print(pd.date_range(start='2023-01-01', periods=5, freq=pd.DateOffset(days=2))) print(pd.date_range(start='2023-01-01', periods=5, freq=pd.DateOffset(weeks=1))) print(pd.date_range(start='2023-01-01', periods=5, freq=pd.DateOffset(months=1))) # 使用自定义业务日历 from pandas.tseries.offsets import CustomBusinessDay # 创建一个自定义业务日历（排除周末） bday = CustomBusinessDay(weekmask='Mon Tue Wed Thu Fri') print("n自定义业务日（排除周末）:") print(pd.date_range(start='2023-01-01', periods=10, freq=bday)) # 创建一个排除特定假日的业务日历 from pandas.tseries.holiday import AbstractHolidayCalendar, Holiday, nearest_workday class MyCustomCalendar(AbstractHolidayCalendar): rules = [ Holiday('New Year', month=1, day=1, observance=nearest_workday), Holiday('My Custom Holiday', month=1, day=15) ] my_bday = CustomBusinessDay(calendar=MyCustomCalendar()) print("n自定义业务日（排除周末和特定假日）:") print(pd.date_range(start='2023-01-01', periods=10, freq=my_bday)) # 使用锚定偏移量 from pandas.tseries.offsets import MonthBegin, MonthEnd, QuarterBegin, QuarterEnd print("n锚定偏移量示例:") print("月初:", pd.date_range(start='2023-01-01', periods=5, freq=MonthBegin())) print("月末:", pd.date_range(start='2023-01-01', periods=5, freq=MonthEnd())) print("季初:", pd.date_range(start='2023-01-01', periods=5, freq=QuarterBegin())) print("季末:", pd.date_range(start='2023-01-01', periods=5, freq=QuarterEnd()))

4. 实战案例：使用Pandas进行时间序列数据分析

让我们通过一个实际案例来综合运用前面学到的Pandas时间处理技巧。假设我们有一组销售数据，需要进行分析和可视化。

import matplotlib.pyplot as plt import seaborn as sns # 设置可视化风格 sns.set(style="whitegrid") plt.rcParams['figure.figsize'] = (12, 6) # 创建示例销售数据 np.random.seed(42) date_rng = pd.date_range(start='2021-01-01', end='2023-12-31', freq='D') sales_data = pd.DataFrame(date_rng, columns=['date']) sales_data['sales'] = np.random.randint(100, 1000, size=(len(date_rng))) sales_data.set_index('date', inplace=True) # 添加季节性模式（周末销售额更高） sales_data['day_of_week'] = sales_data.index.dayofweek sales_data['weekend'] = sales_data['day_of_week'].isin([5, 6]) # 5=周六, 6=周日 sales_data.loc[sales_data['weekend'], 'sales'] *= 1.5 # 周末销售额增加50% # 添加月度趋势（月末销售额更高） sales_data['day_of_month'] = sales_data.index.day sales_data.loc[sales_data['day_of_month'] > 25, 'sales'] *= 1.2 # 月末销售额增加20% # 添加一些随机噪声 sales_data['sales'] = sales_data['sales'] * np.random.normal(1, 0.1, size=(len(sales_data))) # 转换为整数 sales_data['sales'] = sales_data['sales'].astype(int) print("销售数据示例:") print(sales_data.head()) # 1. 基本时间序列分析 # 按月重采样 monthly_sales = sales_data['sales'].resample('M').sum() print("n月度销售额:") print(monthly_sales.head()) # 按季度重采样 quarterly_sales = sales_data['sales'].resample('Q').sum() print("n季度销售额:") print(quarterly_sales.head()) # 2. 计算移动平均 sales_data['7_day_ma'] = sales_data['sales'].rolling(window=7).mean() sales_data['30_day_ma'] = sales_data['sales'].rolling(window=30).mean() sales_data['90_day_ma'] = sales_data['sales'].rolling(window=90).mean() # 可视化移动平均 plt.figure(figsize=(14, 7)) plt.plot(sales_data.index, sales_data['sales'], alpha=0.3, label='日销售额') plt.plot(sales_data.index, sales_data['7_day_ma'], label='7天移动平均') plt.plot(sales_data.index, sales_data['30_day_ma'], label='30天移动平均') plt.plot(sales_data.index, sales_data['90_day_ma'], label='90天移动平均') plt.title('销售额及移动平均趋势') plt.xlabel('日期') plt.ylabel('销售额') plt.legend() plt.tight_layout() plt.show() # 3. 分析季节性模式 # 按星期几分析 weekday_sales = sales_data.groupby('day_of_week')['sales'].mean() weekday_sales.index = ['周一', '周二', '周三', '周四', '周五', '周六', '周日'] plt.figure(figsize=(10, 5)) weekday_sales.plot(kind='bar') plt.title('按星期几的平均销售额') plt.xlabel('星期') plt.ylabel('平均销售额') plt.tight_layout() plt.show() # 按月份分析 monthly_avg = sales_data.groupby(sales_data.index.month)['sales'].mean() monthly_avg.index = ['1月', '2月', '3月', '4月', '5月', '6月', '7月', '8月', '9月', '10月', '11月', '12月'] plt.figure(figsize=(10, 5)) monthly_avg.plot(kind='bar') plt.title('按月份的平均销售额') plt.xlabel('月份') plt.ylabel('平均销售额') plt.tight_layout() plt.show() # 4. 年度同比分析 # 按年月分组 year_month_sales = sales_data.groupby([sales_data.index.year, sales_data.index.month])['sales'].sum() year_month_sales.index.names = ['年', '月'] # 转换为DataFrame以便于比较 year_month_df = year_month_sales.unstack(level=0) # 计算同比增长率 year_month_df['YoY_growth'] = (year_month_df[2023] / year_month_df[2022] - 1) * 100 print("n年度同比分析:") print(year_month_df) # 可视化年度同比 plt.figure(figsize=(10, 5)) year_month_df['YoY_growth'].plot(kind='bar') plt.title('2023年相对于2022年的同比增长率') plt.xlabel('月份') plt.ylabel('增长率 (%)') plt.axhline(y=0, color='r', linestyle='-') plt.tight_layout() plt.show() # 5. 预测未来销售额（简单移动平均法） # 使用最后30天的平均销售额作为预测基准 last_30_days_avg = sales_data['sales'].iloc[-30:].mean() # 创建未来30天的日期 future_dates = pd.date_range(start=sales_data.index[-1] + pd.Timedelta(days=1), periods=30, freq='D') # 创建预测DataFrame forecast_df = pd.DataFrame(index=future_dates) forecast_df['forecast'] = last_30_days_avg # 添加季节性调整 forecast_df['day_of_week'] = forecast_df.index.dayofweek forecast_df['weekend'] = forecast_df['day_of_week'].isin([5, 6]) forecast_df.loc[forecast_df['weekend'], 'forecast'] *= 1.5 # 周末预测值增加50% # 合并历史数据和预测数据 combined_df = pd.concat([sales_data['sales'], forecast_df['forecast']], axis=1) combined_df.columns = ['历史销售额', '预测销售额'] # 可视化预测结果 plt.figure(figsize=(14, 7)) plt.plot(combined_df.index, combined_df['历史销售额'], label='历史销售额') plt.plot(combined_df.index, combined_df['预测销售额'], label='预测销售额', linestyle='--') plt.title('销售额历史数据与预测') plt.xlabel('日期') plt.ylabel('销售额') plt.legend() plt.tight_layout() plt.show()