Source code for openstef_models.explainability.plotters.model_contributions_plotter
# SPDX-FileCopyrightText: 2026 Contributors to the OpenSTEF project <openstef@lfenergy.org>
#
# SPDX-License-Identifier: MPL-2.0
"""Visualizations for ensemble model-level contributions."""
from __future__ import annotations
from typing import TYPE_CHECKING
import plotly.graph_objects as go
from openstef_core.datasets.validated_datasets import ENSEMBLE_COLUMN_SEP
from openstef_core.types import Q, Quantile
from openstef_models.explainability.plotters.common import (
VerticalTwoPanelPlotLayout,
)
if TYPE_CHECKING:
from openstef_core.datasets import ForecastDataset, TimeSeriesDataset
[docs]
class ModelContributionsPlotter:
"""Visualize learned per-model weights alongside the ensemble forecast."""
[docs]
@staticmethod
def plot_stacked_area(
contributions: TimeSeriesDataset,
forecast: ForecastDataset,
quantile: Quantile = Q(0.5),
target_column: str = "load",
) -> go.Figure:
"""Create a stacked-area plot of per-model weights with the ensemble prediction.
The top panel shows the ensemble prediction for the requested quantile.
The bottom panel shows each base model's selection weight as a stacked
area (summing to 1.0), so you can see how the combiner distributes
weight across models at each timestep.
Args:
contributions: Per-model selection weights from a learned-weights
``EnsembleForecastingModel``. Use the result of
``predict_contributions()`` or the second element of
``predict_with_contributions()``. Columns contain model weights
for each forecast quantile.
forecast: Output of
``EnsembleForecastingModel.predict()`` or the first element
of ``EnsembleForecastingModel.predict_with_contributions()``.
Must contain a column for the requested quantile.
quantile: Forecast quantile to show in both panels. Default ``Q(0.5)``.
target_column: Name of the target column to exclude from
contributions. Default ``"load"``.
Returns:
Plotly Figure with a prediction line (top) and stacked-area model
weights (bottom).
Raises:
ValueError: If no contribution columns match the requested quantile.
KeyError: If the forecast dataset has no column for the requested
quantile.
"""
df = contributions.data.drop(columns=[target_column], errors="ignore")
available_columns: dict[tuple[str, str], str] = {}
for col in df.columns:
model, separator, model_quantile = col.rpartition(ENSEMBLE_COLUMN_SEP)
if separator:
available_columns[(model, model_quantile)] = col
selected_columns = [
col for (model, model_quantile), col in available_columns.items() if model_quantile == quantile.format()
]
trace_names = [col[: -len(ENSEMBLE_COLUMN_SEP + quantile.format())] for col in selected_columns]
if not selected_columns:
available = sorted({q for _, q in available_columns})
msg = f"No columns found for quantile {quantile.format()!r}. Available quantiles: {available}."
raise ValueError(msg)
selected = df[selected_columns]
forecast_series = forecast.data[quantile.format()]
fig = VerticalTwoPanelPlotLayout.make_two_panel_time_figure()
VerticalTwoPanelPlotLayout.add_top_panel_line_plot(fig, forecast_series.index, forecast_series)
for col, name in zip(selected_columns, trace_names, strict=True):
fig.add_trace(
go.Scatter(
x=df.index,
y=selected[col],
mode="lines",
stackgroup="contrib",
name=name,
),
row=2,
col=1,
)
fig.update_layout(
yaxis_title=f"Prediction ({quantile.format().removeprefix('quantile_')})",
yaxis2_title=f"Model weight ({quantile.format().removeprefix('quantile_')})",
xaxis2_title="Time",
margin={"t": 30, "r": 10, "b": 40, "l": 60},
)
return fig