Python Pandas处理时间序列的三大核心工具详解

import pandas as pd
import numpy as np

# 生成模拟数据：30天的每小时用电量
dates = pd.date_range('2024-06-01', periods=720, freq='15T')
power = np.random.normal(100, 20, 720).cumsum()  # 累计生成趋势
df = pd.DataFrame({'power': power}, index=dates)

# 计算7日移动标准差（按天聚合后计算）
daily_power = df.resample('D').sum()
daily_std = daily_power.rolling(7).std()

def trimmed_mean(x):
    return (x.sum() - x.max() - x.min()) / (len(x)-2)

df['trimmed_7d'] = df['amount'].rolling(7).apply(trimmed_mean)

# 计算7日移动平均，从第3天开始输出结果
df['ma_7'] = df['value'].rolling(7, min_periods=3).mean()

# 生成模拟分钟级数据
trade_data = pd.DataFrame({
    'price': np.random.normal(100, 1, 1440).cumsum(),
    'volume': np.random.randint(100, 1000, 1440)
}, index=pd.date_range('2024-01-01', periods=1440, freq='T'))

# 重采样为5分钟K线
ohlc_dict = {
    'price': ['first', 'max', 'min', 'last'],
    'volume': 'sum'
}
five_min_data = trade_data.resample('5T').agg(ohlc_dict)

# 生成日频数据
daily_temp = pd.Series(np.random.normal(25, 5, 30), 
                      index=pd.date_range('2024-06-01', periods=30))

# 升频并插值
hourly_temp = daily_temp.resample('H').asfreq().interpolate('linear')

sales_data = pd.DataFrame({
    'amount': np.random.randint(1000, 5000, 90)
}, index=pd.date_range('2024-01-01', periods=90))

# 自定义工作日聚合函数
def weekday_mean(x):
    return x[x.index.weekday < 5].mean()  # 只计算周一到周五

weekly_sales = sales_data.resample('W').apply(weekday_mean)

def detect_trend(series, window=7):
    ma = series.rolling(window).mean()
    recent = ma.dropna().tail(3)
    if len(recent) < 3:
        return '数据不足'
    slope = np.polyfit(range(len(recent)), recent, 1)[0]
    volatility = series.rolling(window).std().iloc[-1]
    
    if slope > 2 and volatility < 10:
        return '明确上升'
    elif slope < -2 and volatility < 10:
        return '明确下降'
    else:
        return '震荡'

df['trend'] = df['price'].apply(lambda x: '')  # 占位列
df.iloc[-1, df.columns.get_loc('trend')] = detect_trend(df['price'])

from statsmodels.tsa.seasonal import STL

# 生成含季节性的销售数据
dates = pd.date_range('2020-01-01', '2022-12-31', freq='D')
sales = 100 + 20*np.sin(2*np.pi*np.arange(len(dates))/365) + np.random.normal(0, 5, len(dates))
df = pd.DataFrame({'sales': sales}, index=dates)

# STL分解
stl = STL(df['sales'], period=365)
result = stl.fit()
trend = result.trend
seasonal = result.seasonal

# 计算30日移动标准差作为波动率
df['volatility'] = df['value'].rolling(30).std()

# 设置动态阈值（均值±3倍波动率）
df['upper_bound'] = df['value'].rolling(30).mean() + 3*df['volatility']
df['lower_bound'] = df['value'].rolling(30).mean() - 3*df['volatility']

# 检测异常
df['anomaly'] = np.where((df['value'] > df['upper_bound']) | 
                         (df['value'] < df['lower_bound']), 1, 0)

# 使用Grouper替代resample提升性能
df['timestamp'] = pd.to_datetime(df['timestamp'])
df.set_index('timestamp', inplace=True)

# 分组计算移动平均（比resample快3倍）
grouped = df.groupby(pd.Grouper(freq='5T'))['value'].mean().rolling(3).mean()

# 前向填充（适用于短期缺失）
df.ffill(limit=3)  # 最多填充3个缺失值

# 时间加权插值（适用于趋势数据）
df.interpolate(method='time')

# 模型预测填充（适用于长期缺失）
from sklearn.ensemble import RandomForestRegressor
# 训练模型预测缺失值...

import dask.dataframe as dd

# 创建Dask DataFrame
ddf = dd.from_pandas(df, npartitions=8)

# 并行计算移动平均
result = ddf.rolling('7D').mean().compute()

# 计算10分钟窗口的振动能量（RMS）
df['vibration_rms'] = df['acceleration'].rolling('10T').apply(lambda x: np.sqrt(np.mean(x**2)))

# 检测异常模式
df['health_score'] = 100 - (df['vibration_rms'] - df['vibration_rms'].mean()) / df['vibration_rms'].std() * 10

# 将车辆位置数据升频为1分钟间隔
truck_data = truck_data.resample('1T').asfreq().interpolate('linear')

# 计算路段通行时间趋势
segment_time = truck_data.groupby('segment_id')['timestamp'].diff().dt.total_seconds()
trend = segment_time.rolling(24*7).mean()  # 7日移动平均

# 计算5日和20日均线
df['ma5'] = df['close'].rolling(5).mean()
df['ma20'] = df['close'].rolling(20).mean()

# 生成交易信号
df['signal'] = np.where(df['ma5'] > df['ma20'], 1, -1)