Time Series Forecasting at Scale
Prophet + LightGBM + Airflow | Automated ML Pipeline
What You'll Learn
- 1Decompose a series and decide when Prophet, LightGBM, or a blend is the right tool
- 2Build lag, rolling, and calendar features without leaking the target into them
- 3Evaluate forecasts with walk-forward validation against a seasonal naive baseline
- 4Gate model registration on a measured improvement instead of on the schedule alone
Project Overview
Problem Statement
Businesses need accurate demand forecasting, revenue predictions, and capacity planning. Time series data has unique patterns (trends, seasonality, holidays) that require specialized models. This pipeline combines Prophet for trend/seasonality with LightGBM for feature-driven forecasting, orchestrated by Airflow.
Objectives
- Build ensemble forecasting with Prophet + LightGBM
- Implement automated feature engineering for time series
- Create Airflow DAGs for scheduled retraining
- Deploy with A/B testing between model versions
- Monitor for concept drift and data quality
| Component | Technology |
|---|---|
| Statistical Model | Facebook Prophet |
| ML Model | LightGBM + XGBoost |
| Orchestration | Apache Airflow |
| Data Storage | BigQuery + Parquet |
| Feature Store | Feast |
| API Framework | FastAPI |
| Monitoring | Evidently AI + Grafana |
| Experiment Tracking | MLflow |
Design Rationale
The pipeline splits the signal because Prophet and LightGBM fail in opposite directions. Prophet fits an additive decomposed model:
where g is a trend with changepoints, s is yearly and weekly seasonality, and h is a holiday regression term. It extrapolates cleanly but ignores exogenous drivers, so it cannot know that a promotion doubled demand. LightGBM consumes lag and rolling features and learns exactly those interactions, but it cannot extrapolate past the range of its labels, which is why the ensemble feeds the Prophet prediction in as a feature. The blend keeps both strengths:
Weights are fit on a validation window rather than guessed. If the blend never beats the better single model, the features are not adding signal and the extra component should be removed.
Accuracy is judged against a baseline and reported with error measures that behave well near zero:
The floor to beat is the seasonal naive forecast, which simply repeats the value from one season ago:
For daily data with weekly seasonality that means reusing last Monday's value for this Monday. If the ensemble cannot beat seasonal naive by at least 20 percent in RMSE on a walk-forward split, the added machinery is decoration.
Walk-forward validation trains on a window, predicts the next h steps, then slides the window forward and repeats. It mirrors how the model will actually be used, produces an error estimate per horizon so short-term and 30 day accuracy can be reported separately, and avoids the future-into-past leakage a random split would create.
Architecture Diagram
Step-by-Step Implementation
Step 1: Environment Setup
mkdir ts-forecasting && cd ts-forecasting
pip install prophet lightgbm xgboost
pip install apache-airflow airflow-providers-google
pip install feast mlflow evidently
pip install fastapi uvicorn pandas numpy
pip install statsmodels scipy
Step 2: Feature Engineering for Time Series
Build comprehensive feature engineering that captures temporal patterns, lag features, and external regressors.
# src/features/time_features.py
import pandas as pd
import numpy as np
from typing import List, Optional
from dataclasses import dataclass
@dataclass
class FeatureConfig:
target_col: str
date_col: str
lag_periods: List[int] = None
rolling_windows: List[int] = None
holiday_dates: Optional[List[str]] = None
def __post_init__(self):
if self.lag_periods is None:
self.lag_periods = [1, 7, 14, 28, 90]
if self.rolling_windows is None:
self.rolling_windows = [7, 14, 28, 90]
class TimeSeriesFeatureEngineer:
def __init__(self, config: FeatureConfig):
self.config = config
def create_features(self, df: pd.DataFrame) -> pd.DataFrame:
df = df.copy()
df[self.config.date_col] = pd.to_datetime(df[self.config.date_col])
df = df.sort_values(self.config.date_col).reset_index(drop=True)
# Calendar features
df["day_of_week"] = df[self.config.date_col].dt.dayofweek
df["month"] = df[self.config.date_col].dt.month
df["quarter"] = df[self.config.date_col].dt.quarter
df["year"] = df[self.config.date_col].dt.year
df["is_weekend"] = (df["day_of_week"] >= 5).astype(int)
df["day_of_month"] = df[self.config.date_col].dt.day
df["week_of_year"] = df[self.config.date_col].dt.isocalendar().week.astype(int)
# Lag features
for lag in self.config.lag_periods:
df[f"lag_{lag}"] = df[self.config.target_col].shift(lag)
# Rolling statistics
for window in self.config.rolling_windows:
df[f"rolling_mean_{window}"] = df[self.config.target_col].rolling(window=window).mean()
df[f"rolling_std_{window}"] = df[self.config.target_col].rolling(window=window).std()
df[f"rolling_min_{window}"] = df[self.config.target_col].rolling(window=window).min()
df[f"rolling_max_{window}"] = df[self.config.target_col].rolling(window=window).max()
df[f"rolling_median_{window}"] = df[self.config.target_col].rolling(window=window).median()
# Exponentially weighted features
for span in [7, 14, 28]:
df[f"ewm_mean_{span}"] = df[self.config.target_col].ewm(span=span).mean()
# Diff features
df["diff_1"] = df[self.config.target_col].diff(1)
df["diff_7"] = df[self.config.target_col].diff(7)
# Percentage change
df["pct_change_1"] = df[self.config.target_col].pct_change(1)
df["pct_change_7"] = df[self.config.target_col].pct_change(7)
# Holiday features
if self.config.holiday_dates:
holidays = pd.to_datetime(self.config.holiday_dates)
df["is_holiday"] = df[self.config.date_col].isin(holidays).astype(int)
df["days_to_holiday"] = df[self.config.date_col].apply(
lambda x: min([abs((x - h).days) for h in holidays]) if holidays.any() else 0
)
return df.dropna()
def create_prophet_features(self, df: pd.DataFrame) -> pd.DataFrame:
prophet_df = df[[self.config.date_col, self.config.target_col]].copy()
prophet_df.columns = ["ds", "y"]
return prophet_df
Step 3: Prophet + LightGBM Ensemble
Combine Prophet for trend/seasonality with LightGBM for feature-driven corrections.
# src/models/ensemble.py
import pandas as pd
import numpy as np
from prophet import Prophet
import lightgbm as lgb
from typing import Dict, Tuple, Optional
import mlflow
from dataclasses import dataclass
@dataclass
class EnsembleConfig:
prophet_seasonality: str = "monthly"
lgbm_params: Dict = None
ensemble_method: str = "weighted" # weighted, stacking
prophet_weight: float = 0.4
lgbm_weight: float = 0.6
def __post_init__(self):
if self.lgbm_params is None:
self.lgbm_params = {
"objective": "regression",
"metric": "rmse",
"num_leaves": 31,
"learning_rate": 0.05,
"feature_fraction": 0.8,
"bagging_fraction": 0.8,
"bagging_freq": 5,
"verbose": -1,
}
class ProphetLightGBMEnsemble:
def __init__(self, config: EnsembleConfig = None):
self.config = config or EnsembleConfig()
self.prophet_model = None
self.lgbm_model = None
def fit(self, df: pd.DataFrame, target_col: str, date_col: str):
# Split for LGBM feature-based predictions
feature_cols = [c for c in df.columns if c not in [target_col, date_col, "ds", "y"]]
# Fit Prophet
prophet_df = df[[date_col, target_col]].rename(columns={date_col: "ds", target_col: "y"})
self.prophet_model = Prophet(
yearly_seasonality=True,
weekly_seasonality=True,
daily_seasonality=False,
changepoint_prior_scale=0.05,
)
self.prophet_model.fit(prophet_df)
# Get Prophet predictions as feature for LGBM
prophet_pred = self.prophet_model.predict(prophet_df)["yhat"].values
# Fit LightGBM with Prophet predictions as additional feature
X = df[feature_cols].copy()
X["prophet_pred"] = prophet_pred
y = df[target_col].values
train_data = lgb.Dataset(X, label=y)
self.lgbm_model = lgb.train(
self.config.lgbm_params, train_data, num_boost_round=1000,
valid_sets=[train_data], callbacks=[lgb.log_evaluation(100)],
)
self.feature_cols = feature_cols
def predict(self, df: pd.DataFrame) -> pd.DataFrame:
# Prophet forecast
future = self.prophet_model.make_future_dataframe(periods=len(df))
prophet_forecast = self.prophet_model.predict(future)
# LGBM prediction
X = df[self.feature_cols].copy()
X["prophet_pred"] = prophet_forecast["yhat"].values[-len(df):]
lgbm_pred = self.lgbm_model.predict(X)
# Ensemble
ensemble_pred = (
self.config.prophet_weight * prophet_forecast["yhat"].values[-len(df):]
+ self.config.lgbm_weight * lgbm_pred
)
return pd.DataFrame({
"date": df["date"].values,
"prophet_pred": prophet_forecast["yhat"].values[-len(df):],
"lgbm_pred": lgbm_pred,
"ensemble_pred": ensemble_pred,
"prophet_lower": prophet_forecast["yhat_lower"].values[-len(df):],
"prophet_upper": prophet_forecast["yhat_upper"].values[-len(df):],
})
Step 4: Airflow DAG for Automated Retraining
# dags/forecast_pipeline.py
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator
from airflow.sensors.external_task import ExternalTaskSensor
from datetime import datetime, timedelta
default_args = {
"owner": "ml-team",
"depends_on_past": False,
"email_on_failure": True,
"email": ["[email protected]"],
"retries": 2,
"retry_delay": timedelta(minutes=10),
}
def extract_data(**context):
"""Extract latest data from BigQuery."""
from google.cloud import bigquery
client = bigquery.Client()
query = "SELECT * FROM `project.dataset.time_series_data` WHERE date >= DATE_SUB(CURRENT_DATE(), INTERVAL 365 DAY)"
df = client.query(query).to_dataframe()
df.to_parquet("/tmp/ts_data.parquet")
return f"Extracted {len(df)} rows"
def train_models(**context):
"""Train Prophet + LightGBM ensemble."""
import pandas as pd
from src.models.ensemble import ProphetLightGBMEnsemble, EnsembleConfig
import mlflow
df = pd.read_parquet("/tmp/ts_data.parquet")
config = EnsembleConfig()
ensemble = ProphetLightGBMEnsemble(config)
with mlflow.start_run():
ensemble.fit(df, target_col="value", date_col="date")
mlflow.prophet.log_model(ensemble.prophet_model, "prophet_model")
mlflow.lightgbm.log_model(ensemble.lgbm_model, "lgbm_model")
return "Training complete"
def evaluate_and_register(**context):
"""Evaluate and register model if better than current."""
from src.monitoring.drift import DriftDetector
detector = DriftDetector()
is_better = detector.compare_models("current", "new")
if is_better:
detector.register_model("new")
return "Model registered"
return "Model not registered - no improvement"
with DAG(
"time_series_forecast",
default_args=default_args,
description="Automated time series forecasting pipeline",
schedule_interval="0 2 * * 1", # Weekly on Monday at 2 AM
start_date=datetime(2024, 1, 1),
catchup=False,
) as dag:
extract = PythonOperator(task_id="extract_data", python_callable=extract_data)
train = PythonOperator(task_id="train_models", python_callable=train_models)
evaluate = PythonOperator(task_id="evaluate_register", python_callable=evaluate_and_register)
deploy = BashOperator(task_id="deploy_model", bash_command="python -m src.deploy.push_model")
extract >> train >> evaluate >> deploy
Step 5: FastAPI Serving
# src/api/main.py
from fastapi import FastAPI
from pydantic import BaseModel
from typing import List, Optional
from contextlib import asynccontextmanager
import pandas as pd
@asynccontextmanager
async def lifespan(app: FastAPI):
# Load models at startup
app.state.ensemble = load_ensemble("production")
yield
app = FastAPI(title="Forecast API", lifespan=lifespan)
class ForecastRequest(BaseModel):
historical_data: List[dict]
horizon: int = 30
include_confidence: bool = True
class ForecastResponse(BaseModel):
forecasts: List[dict]
model_version: str
@app.post("/api/v1/forecast", response_model=ForecastResponse)
async def forecast(request: ForecastRequest):
df = pd.DataFrame(request.historical_data)
predictions = app.state.ensemble.predict(df)
return ForecastResponse(
forecasts=predictions.to_dict(orient="records"),
model_version="v1.0.0",
)
âšī¸
Always include confidence intervals in forecasts. Point predictions alone are insufficient for business decision-making. Prophet provides natural uncertainty intervals.
đĄ
Monitor forecast accuracy by horizon length. Short-term forecasts (1-7 days) should be much more accurate than long-term (30+ days). Set different alerting thresholds per horizon.
Performance Metrics
| Metric | Target | Description |
|---|---|---|
| MAPE | < 10% | Mean Absolute Percentage Error |
| RMSE | Baseline -20% | vs simple baseline |
| Coverage | > 90% | 95% CI actual coverage |
| Retraining Latency | < 30min | Full pipeline |
| Inference Latency | < 100ms | Per forecast request |
What Breaks in Production
Forecasting pipelines break on data plumbing more often than on modeling. Late-arriving rows create silent gaps: an export that lands at 03:00 while the DAG runs at 02:00 trains without the most recent day, which is precisely the signal that matters most. Assert row count and max(date) before training instead of trusting the schedule.
Second, leakage through rolling features. A rolling mean computed over a window that includes the current row hands the model the answer, cuts backtest RMSE roughly in half, and then performs no better in production. Build every rolling statistic strictly from past rows and unit test the property directly: perturbing today's target must leave today's features unchanged.
Third, holiday and changepoint overfit. Handing Prophet 40 holiday dates on two years of data makes it fit noise, intervals collapse, and error on the next quarter grows. Select holidays by cross-validated error rather than by completeness, and raise changepoint_prior_scale only when the trend is visibly unstable.
Finally, unconditional retraining. A weekly DAG that always registers its newest model will eventually promote a regression, so gate registration on a measured margin:
A new model must beat the incumbent by at least 5 percent, and the previous version stays one rollback away so a bad promotion costs minutes, not a quarter.
Interview Talking Points
- Ensemble Strategy: Prophet captures trend and seasonality, LightGBM handles feature interactions and non-linear patterns.
- Feature Engineering: Lag features, rolling statistics, and calendar features are critical for forecasting accuracy.
- Airflow Orchestration: DAGs ensure reliable, auditable, and schedulable ML pipelines.
- Backtesting: Walk-forward validation simulates real-world forecasting conditions.
- Drift Detection: Monitor for distribution shifts in input data and forecast accuracy degradation.
- Business Impact: Connect forecast accuracy to business KPIs like inventory costs and revenue.
â ī¸
Prophet can overfit with too many holidays or changepoints. Cross-validate the number of changepoints and holiday effects before production deployment.
âšī¸
For high-frequency data (intraday), consider using NeuralProphet or DeepAR which handle sub-daily patterns better.
Common Mistakes
Random train and test splits on a time series
Fix: Use walk-forward validation; a random split puts future rows in training and flatters every metric.
Rolling windows that include the current row
Fix: Build rolling features from past rows only, and unit test that perturbing today's target leaves today's features unchanged.
Registering whichever model trained last
Fix: Require at least a 5 percent RMSE improvement over the incumbent before the model is promoted.
Best Practices
- Report MAPE and RMSE per horizon bucket instead of one blended number
- Return confidence intervals from the API so consumers can plan against the width, not the point
- Assert row count and max(date) before every training run so late data is caught early
Key Takeaways
- Prophet extrapolates trend and seasonality while LightGBM learns feature interactions, and the blend takes both
- Seasonal naive is the floor: beat it by 20 percent RMSE or simplify the pipeline
- Walk-forward validation mirrors production usage and avoids the leakage a random split creates
- A 5 percent RMSE gate stops a scheduled retrain from promoting a regression
Frequently Asked Questions
Why blend Prophet with LightGBM at all?
Prophet owns extrapolation: it continues a trend and repeats seasonality it has seen many times. LightGBM owns conditioning: it reacts to promotions, price changes, and weather that Prophet cannot represent. Blending them at 0.4 and 0.6 beats either alone precisely when the series has both a stable seasonal shape and volatile exogenous drivers.
How many lag features should you create?
Start with one full season of each granularity you trust: lag 1, 7, 14, and 28 for daily data, plus rolling mean and standard deviation over 7 and 28. Beyond a handful of seasons the feature count grows faster than the data supports, and LightGBM's feature_fraction sampling starts hiding useful columns rather than revealing them.
What do you do for a brand new series with no history?
Fall back to a hierarchical forecast: predict at the category level with the same model, then allocate down using whatever share information exists, or use a global model trained across all series so cold items inherit learned seasonality. Keep a separate error metric for cold series so the fallback's quality is visible.
Interview Questions
Q1: Explain walk-forward validation and why a random split is invalid here.
Train on rows up to time t, predict t+1 to t+h, then slide t forward and repeat, accumulating one error per horizon. A random split trains on future rows to predict past ones, which leaks autocorrelated signal and typically understates real error by 30 percent or more.
Q2: How do you prove your features do not leak the target?
Perturb a single row's target value and assert every feature for that row is bit-identical afterward. Rolling windows must be closed on the left, diffs must use periods of at least 1, and any feature computed from the target must be shifted before aggregation.
Q3: Why do you return confidence intervals instead of a single number?
Inventory decisions depend on the width of the uncertainty, not its center. A forecast of 100 units with an 80 to 120 interval implies different safety stock than the same 100 with a 40 to 160 interval, and Prophet's interval gives that width for free.