⏱️python时序处理

🧰 一、核心库概览（按用途分类）

🚀 二、重点库详解 + 实战代码

1. 🐼 pandas —— 时序处理基石

几乎所有时序任务的第一步都用 pandas。

import pandas as pd
import numpy as np

# 创建带时间索引的 Series
dates = pd.date_range('2023-01-01', periods=100, freq='D')
ts = pd.Series(np.random.randn(100).cumsum(), index=dates)

# 重采样（日 → 周）
weekly = ts.resample('W').mean()

# 滑动窗口（7日移动平均）
rolling_mean = ts.rolling(window=7).mean()

# 时序对齐（两个不同频率序列）
ts1 = pd.Series([1,2,3], index=pd.to_datetime(['2023-1-1','2023-1-3','2023-1-5']))
ts2 = pd.Series([10,20], index=pd.to_datetime(['2023-1-2','2023-1-4']))
aligned = pd.concat([ts1, ts2], axis=1).fillna(method='ffill')

# 时区处理
ts_utc = ts.tz_localize('UTC').tz_convert('Asia/Shanghai')

✅ 优势：API 直观，社区支持极强。
⚠️ 注意：大数据（>1亿行）时性能下降，可考虑 Polars。

2. 📈 statsmodels —— 统计建模

from statsmodels.tsa.seasonal import seasonal_decompose
from statsmodels.tsa.arima.model import ARIMA

# 季节分解（加法模型）
result = seasonal_decompose(ts, model='additive', period=7)
result.plot()

# 拟合 ARIMA(1,1,1)
model = ARIMA(ts, order=(1,1,1))
fitted = model.fit()
forecast = fitted.forecast(steps=7)

✅ 适合：小数据、可解释性强、需统计检验（如 ADF 平稳性检验）。

3. 🤖 sktime —— 时序机器学习统一框架

类似 scikit-learn，但专为时序设计。

from sktime.forecasting.naive import NaiveForecaster
from sktime.forecasting.arima import AutoARIMA
from sktime.transformations.series.detrend import Detrender
from sktime.forecasting.compose import TransformedTargetForecaster
from sktime.performance_metrics.forecasting import mean_absolute_percentage_error

# 加载示例数据
from sktime.datasets import load_airline
y = load_airline()  # 月度航空乘客数

# 构建 pipeline：去趋势 + AutoARIMA
forecaster = TransformedTargetForecaster([
    ("deseasonalize", Detrender()),
    ("arima", AutoARIMA())
])

# 拟合与预测
forecaster.fit(y)
y_pred = forecaster.predict(fh=[1,2,3])  # 预测未来3个月

# 评估
mape = mean_absolute_percentage_error(y[-3:], y_pred)

✅ 优势：统一接口，支持交叉验证（SlidingWindowSplitter），内置 50+ 模型。

4. 🧠 darts —— 深度 + 传统模型一体化

支持 PyTorch/TensorFlow，API 简洁。

from darts import TimeSeries
from darts.models import ExponentialSmoothing, RNNModel, Prophet
from darts.dataprocessing.transformers import Scaler

# 从 pandas 转为 Darts TimeSeries
series = TimeSeries.from_series(ts)

# 划分训练/测试
train, val = series.split_before(pd.Timestamp('2023-03-01'))

# 标准化
scaler = Scaler()
train_scaled = scaler.fit_transform(train)

# 训练多种模型
models = {
    'ARIMA': ExponentialSmoothing(),
    'RNN': RNNModel(input_chunk_length=24, training_length=48, n_epochs=20),
    'Prophet': Prophet()
}

for name, model in models.items():
    model.fit(train_scaled)
    pred = model.predict(n=len(val))
    pred = scaler.inverse_transform(pred)  # 反标准化
    print(f"{name} MAPE: {mape(val, pred)}")

✅ 亮点：

• 自动处理日期索引
• 内置 Prophet、Theta、N-BEATS、TCN 等 SOTA 模型
• 支持多变量、协变量（如天气、促销）

5. ⚡ polars —— 高性能替代 pandas

import polars as pl

# 创建时序 DataFrame
df = pl.DataFrame({
    "time": pl.date_range("2023-01-01", "2023-04-01", "1d"),
    "value": np.random.randn(91).cumsum()
}).with_columns(pl.col("time").dt.weekday().alias("weekday"))

# 重采样（日 → 周）
weekly = df.groupby_dynamic("time", every="1w").agg(pl.col("value").mean())

# 滑动窗口
df = df.with_columns(
    pl.col("value").rolling_mean(window_size=7).alias("rolling_7d")
)

✅ 性能：在 10M+ 行数据上，比 pandas 快 5~10 倍，内存更低。

6. 🧪 tsfresh —— 自动特征提取

from tsfresh import extract_features
from tsfresh.utilities.dataframe_functions import impute

# 假设 df 有 ['id', 'time', 'value'] 列（多条时序）
df = pd.DataFrame({
    'id': [1]*100 + [2]*100,
    'time': list(range(100)) + list(range(100)),
    'value': np.random.randn(200)
})

# 自动提取 1000+ 特征（均值、峰度、FFT 系数等）
features = extract_features(df, column_id="id", column_sort="time")
features = impute(features)  # 填充 NaN

✅ 适合：将时序转为静态特征，输入 XGBoost/LightGBM。

🔍 三、如何选择？

📚 四、学习资源推荐

• 官方文档：

  • pandas Time Series
  • sktime Tutorials
  • darts Examples

• 书籍：

  • 《Practical Time Series Forecasting with R/Python》
  • 《Time Series Analysis and Its Applications》（Shumway & Stoffer）

• 数据集：

  • sktime.datasets（内置）
  • UCI Time Series Archive
  • Kaggle（如 Store Sales, Web Traffic）

✅ 总结

• 起步：用 pandas 处理 + statsmodels 建模。
• 进阶 ML：切换到 sktime 或 darts。
• 性能瓶颈：尝试 polars。
• 自动特征：用 tsfresh。

💡 黄金法则：不要重复造轮子！优先使用这些成熟库，而非手写 for 循环处理时间戳。

附录

"""
Darts实战：用LSTM预测销售额
场景：预测某商品未来7天的销售额
数据：模拟的每日销售额（包含趋势、季节性、噪声）
模型：Darts + PyTorch LSTM
"""
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from datetime import datetime, timedelta
import warnings
warnings.filterwarnings('ignore')

# 1. 安装 Darts（如果未安装）
# pip install darts

from darts import TimeSeries
from darts.models import RNNModel, ExponentialSmoothing, Prophet
from darts.dataprocessing.transformers import Scaler
from darts.metrics import mape, mae
from darts.utils.timeseries_generation import datetime_attribute_timeseries

# 2. 生成模拟销售额数据
def generate_sales_data(start_date='2022-01-01', n_days=730):  # 两年数据
    dates = pd.date_range(start=start_date, periods=n_days, freq='D')
    
    # 模拟趋势（缓慢增长）
    trend = np.linspace(1000, 1500, n_days)
    
    # 模拟周季节性（周末销量更高）
    weekly_seasonality = 100 * np.sin(2 * np.pi * np.arange(n_days) / 7)
    
    # 模拟月季节性（月末促销）
    monthly_seasonality = 50 * np.sin(2 * np.pi * (np.arange(n_days) % 30) / 30)
    
    # 添加噪声
    noise = np.random.normal(0, 50, n_days)
    
    sales = trend + weekly_seasonality + monthly_seasonality + noise
    sales = np.clip(sales, 0, None)  # 销售额不能为负
    
    return pd.DataFrame({'date': dates, 'sales': sales})

df = generate_sales_data()
print("数据概览:")
print(df.head())
print(f"总数据点: {len(df)}")

# 3. 数据预处理
# 转换为 Darts TimeSeries
ts = TimeSeries.from_dataframe(df, time_col='date', value_cols='sales')
print(f"TimeSeries shape: {ts.pd_series().shape}")

# 划分训练/测试集（最后30天作为测试）
train, test = ts.split_before(ts.time_index[-30])
print(f"训练集: {len(train)}, 测试集: {len(test)}")

# 数据标准化（对LSTM很重要）
scaler = Scaler()
train_scaled = scaler.fit_transform(train)
test_scaled = scaler.transform(test)

# 4. 构建LSTM模型
lstm_model = RNNModel(
    model='LSTM',
    input_chunk_length=30,  # 用过去30天的数据预测未来
    training_length=100,    # 每次训练用100个时间步
    n_epochs=50,            # 训练轮数
    random_state=42,
    model_name='LSTM_Sales',
    log_tensorboard=False,
    save_checkpoints=True,
)

# 5. 训练模型
print("\n开始训练LSTM模型...")
lstm_model.fit(train_scaled)

# 6. 预测
print("\n进行预测...")
# 预测未来7天
forecast_horizon = 7
lstm_forecast_scaled = lstm_model.predict(n=forecast_horizon)

# 反标准化
lstm_forecast = scaler.inverse_transform(lstm_forecast_scaled)

# 7. 评估模型（在测试集上）
historical_forecast = lstm_model.historical_forecasts(
    train_scaled, 
    start=0.7,  # 从70%处开始预测
    forecast_horizon=1,
    stride=1,
    retrain=False
)
historical_forecast = scaler.inverse_transform(historical_forecast)

mape_score = mape(train[int(len(train)*0.7):], historical_forecast)
print(f"历史预测MAPE: {mape_score:.2f}%")

# 8. 对比其他模型
print("\n训练基准模型...")
baseline_model = ExponentialSmoothing()
baseline_model.fit(train_scaled)
baseline_forecast_scaled = baseline_model.predict(n=forecast_horizon)
baseline_forecast = scaler.inverse_transform(baseline_forecast_scaled)

# 9. 可视化结果
plt.figure(figsize=(14, 8))
train.plot(label='训练数据', lw=1.5)
test.plot(label='真实测试数据', lw=1.5)
lstm_forecast.plot(label='LSTM预测', lw=2, marker='o')
baseline_forecast.plot(label='指数平滑预测', lw=2, marker='s')

plt.title('销售额预测对比', fontsize=16)
plt.xlabel('日期')
plt.ylabel('销售额')
plt.legend()
plt.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()

# 10. 输出预测结果
print("\n未来7天销售额预测:")
forecast_df = pd.DataFrame({
    'Date': lstm_forecast.time_index,
    'LSTM_Forecast': lstm_forecast.pd_series().values,
    'Baseline_Forecast': baseline_forecast.pd_series().values
})
forecast_df['LSTM_Forecast'] = forecast_df['LSTM_Forecast'].round(2)
forecast_df['Baseline_Forecast'] = forecast_df['Baseline_Forecast'].round(2)
print(forecast_df)

# 11. 模型性能总结
print(f"\n模型性能总结:")
print(f"LSTM预测未来7天的平均销售额: {lstm_forecast.mean().values[0]:.2f}")
print(f"基准模型预测未来7天的平均销售额: {baseline_forecast.mean().values[0]:.2f}")
print(f"历史同期平均销售额: {train[-30:].mean().values[0]:.2f}")

# 12. 预测置信区间（可选）
print("\n生成预测置信区间...")
lstm_forecast_with_ci = lstm_model.predict(n=forecast_horizon, num_samples=200)
lower_quantile = lstm_forecast_with_ci.quantile(0.05)
upper_quantile = lstm_forecast_with_ci.quantile(0.95)

plt.figure(figsize=(12, 6))
train[-60:].plot(label='近期实际数据', lw=2)
lstm_forecast.plot(label='LSTM预测', lw=2, marker='o')
plt.fill_between(
    lstm_forecast.time_index,
    lower_quantile.pd_series().values,
    upper_quantile.pd_series().values,
    color='blue', alpha=0.2, label='90%置信区间'
)
plt.title('销售额预测及置信区间', fontsize=16)
plt.xlabel('日期')
plt.ylabel('销售额')
plt.legend()
plt.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()

print("\n预测置信区间:")
for i, (date, low, high) in enumerate(zip(
    lstm_forecast.time_index,
    lower_quantile.pd_series().values,
    upper_quantile.pd_series().values
)):
    print(f"{date.strftime('%Y-%m-%d')}: {low:.2f} - {high:.2f}")

代码说明:

1. 数据生成: 模拟了包含趋势、周季节性、月季节性和噪声的销售数据。
2. 数据预处理: 使用 Darts 的 TimeSeries 格式，并应用 Scaler 进行标准化，这对神经网络模型至关重要。
3. 模型训练: 配置并训练一个 LSTM 模型，使用过去30天的数据预测未来一天，通过滑动窗口训练。
4. 预测: 生成未来7天的预测值，并与简单的指数平滑模型进行对比。
5. 评估: 使用 MAPE 等指标评估模型性能。
6. 可视化: 绘制训练数据、测试数据、LSTM预测和基准模型预测的对比图。
7. 置信区间: 通过蒙特卡洛采样生成预测的不确定性区间，提供风险评估。