Python使用Darts进行预测数据的完整教学

import pandas as pd
from darts import TimeSeries

df = pd.read_csv('sales.csv', parse_dates=['date'])
series = TimeSeries.from_dataframe(
    df, 
    time_col='date', 
    value_cols='sales',
    fill_missing_dates=True,  # 自动补全缺失日期
    freq='D'                   # 设定频率
)

from darts.models import (
    NaiveSeasonal,    # 季节性基准
    ExponentialSmoothing,  # 指数平滑
    ARIMA,            # 自回归积分滑动平均
    Prophet           # Facebook Prophet
)
from darts.metrics import mape, rmse

# 1. 划分训练集和测试集（8:2）
train, test = series.split_after(pd.Timestamp('2025-01-01'))

# 2. 定义模型列表
models = [
    NaiveSeasonal(K=12),                    # 月度季节性基准
    ExponentialSmoothing(seasonal_periods=12),
    ARIMA(p=1, d=1, q=1),
    Prophet()
]

# 3. 训练并评估
for model in models:
    model.fit(train)
    forecast = model.predict(n=len(test))
    error = mape(test, forecast)
    print(f"{model.__class__.__name__:25s} MAPE: {error:.2f}%")

from darts.models import NBEATSModel

model = NBEATSModel(
    input_chunk_length=24,    # 回看24个时间步
    output_chunk_length=12,   # 预测未来12步
    n_epochs=50,
    batch_size=64,
    random_state=42
)

model.fit(train)
forecast = model.predict(n=len(test))

# 可视化
series.plot(label='Actual')
forecast.plot(label='Forecast')

from darts.models import TFTModel

# 假设有天气数据作为协变量
weather_series = TimeSeries.from_dataframe(weather_df, 'date', 'temp')

model = TFTModel(
    input_chunk_length=30,
    output_chunk_length=7,
    hidden_size=64,
    lstm_layers=2,
    num_attention_heads=4,
    add_relative_index=True
)

model.fit(
    train,
    future_covariates=weather_series,
    val_series=test
)

forecast = model.predict(n=7, future_covariates=weather_series)

from darts.models import NBEATSModel

# 假设有100家门店的销售数据
series_list = [TimeSeries.from_dataframe(df, 'date', f'store_{i}') 
               for i in range(100)]

model = NBEATSModel(input_chunk_length=24, output_chunk_length=7)
model.fit(series_list)  # 在100条序列上同时训练

# 预测时可以一次性处理多条序列
forecasts = model.predict(series=series_list, n=7)

from darts.metrics import smape

# 滑动窗口回测：从50%数据开始，每次预测12步
error = model.backtest(
    series=train,
    forecast_horizon=12,
    start=0.5,
    metric=smape
)
print(f"回测SMAPE: {error:.2f}%")

from darts.utils.model_selection import TimeSeriesCrossValidator

cv = TimeSeriesCrossValidator(cv=3, horizon=6, start=0.5)
for train_split, val_split in cv.split(series):
    model.fit(train_split)
    pred = model.predict(len(val_split))
    # 评估验证集表现

from darts.utils.model_selection import ModelComparison

models = [
    NBEATSModel(input_chunk_length=24, output_chunk_length=12),
    TCNModel(input_chunk_length=24, output_chunk_length=12),
    TransformerModel(input_chunk_length=24, output_chunk_length=12)
]

comparison = ModelComparison(models=models)
comparison.fit(train)
metrics = comparison.score(test, metrics=['mae', 'rmse', 'mape'])
print(metrics)

from darts.dataprocessing.transformers import Scaler, MissingValuesFiller
from darts.utils.statistics import check_seasonality

# 检查季节性
is_seasonal, period = check_seasonality(series, alpha=0.05)
print(f"是否存在季节性: {is_seasonal}, 周期: {period}")

# 填充 + 标准化
filler = MissingValuesFiller()
scaler = Scaler()
series_clean = filler.transform(series)
series_scaled = scaler.fit_transform(series_clean)

from darts.ad import ForecastingAnomalyModel, KMeansScorer

# 用Prophet作为底层预测模型
from darts.models import Prophet
predictor = Prophet()

# KMeans评分器检测异常
scorer = KMeansScorer(window=24)

anomaly_model = ForecastingAnomalyModel(
    predictor,
    scorer
)

anomaly_model.fit(train)
anomalies = anomaly_model.score(series)

# 基础安装
pip install darts
# 深度学习模型（需PyTorch）
pip install darts[u]
# 异常检测模块
pip install darts[ad]

# 保存
model.save("my_nbeats.pt")
# 加载
loaded_model = NBEATSModel.load("my_nbeats.pt")

pip install darts