Plotting

Requires either matplotlib or plotly to be installed, e.g. via pip install "tirex-2[examples]".

tirex2.plotting.plot_multivariate

plot_multivariate(input: TimeseriesType, forecast: Tensor | ndarray, ground_truth: Tensor | ndarray | None = None, x: Sequence | None = None, quantiles: tuple[float, float] = (0.1, 0.9), quantile_levels: tuple[float] = (0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9), max_context_to_show: int | None = None, target_index: int = 0, past_cov_labels: list[str] | None = None, future_cov_labels: list[str] | None = None, engine='plotly', title: str | None = None, subtitle: str | None = None)

Source code in src/tirex2/plotting.py

def plot_multivariate(
    input: TimeseriesType,
    forecast: torch.Tensor | np.ndarray,
    ground_truth: torch.Tensor | np.ndarray | None = None,
    x: Sequence | None = None,
    quantiles: tuple[float, float] = (0.1, 0.9),
    quantile_levels: tuple[float] = (0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9),
    max_context_to_show: int | None = None,
    target_index: int = 0,
    past_cov_labels: list[str] | None = None,
    future_cov_labels: list[str] | None = None,
    engine="plotly",
    title: str | None = None,
    subtitle: str | None = None,
):
    # determine sizes of displayed time series
    forecast_length = forecast.shape[-1] if forecast is not None else 0
    ground_truth_length = len(ground_truth) if ground_truth is not None else 0
    max_future_length = max(forecast_length, ground_truth_length, input.future_length)
    full_size = input.past_length + max_future_length

    if x is None:
        x = np.arange(full_size)
        # x = np.arange(-input.past_length, max_future_length) + 1
    elif len(x) < full_size:
        raise ValueError(
            "Not enough 'x' values provided to have one for every timestep in context, forecast, and ground truth window."
        )

    # prepare covariate slices and labels
    cov_slice_start = 0 if max_context_to_show is None else input.past_length - max_context_to_show
    past_covariates = input.past_covariates if input.past_covariates is not None else []
    future_covariates = input.future_covariates if input.future_covariates is not None else []
    if len(past_covariates) > 0:
        if past_cov_labels is not None and len(past_cov_labels) != len(past_covariates):
            raise ValueError("Length of 'past_cov_labels' must match the number of past covariates.")
        elif past_cov_labels is None:
            past_cov_labels = []
            for i, cov in enumerate(past_covariates):
                past_cov_labels.append(f"Past Covariate {i + 1}")
    else:
        past_cov_labels = []

    if len(future_covariates) > 0:
        if future_cov_labels is not None and len(future_cov_labels) != len(future_covariates):
            raise ValueError("Length of 'future_cov_labels' must match the number of future covariates.")
        elif future_cov_labels is None:
            future_cov_labels = []
            for i, cov in enumerate(future_covariates):
                future_cov_labels.append(f"Future Covariate {i + 1}")
    else:
        future_cov_labels = []

    cov_lookup = {}
    for i, (lbl, cov) in enumerate(
        chain(
            zip(past_cov_labels, past_covariates),
            zip(future_cov_labels, future_covariates),
        )
    ):
        cov_lookup[lbl] = (
            x[cov_slice_start : cov_slice_start + len(cov)],
            cov[cov_slice_start : cov_slice_start + len(cov)],
        )

    # prepare plot parameters for the plotting functions
    plot_params = dict(
        context=input.target[target_index, cov_slice_start:],
        x=x[cov_slice_start:],
        quantiles=quantiles,
        quantile_levels=quantile_levels,
        forecast=forecast[target_index],
        ground_truth=ground_truth,
        cov_lookup=cov_lookup,
        title=title,
        subtitle=subtitle,
    )

    # call appropriate plotting function based on the specified engine
    match engine:
        case "matplotlib":
            return _plot_multivariate_matplotlib(**plot_params)
        case "plotly":
            return _plot_multivariate_plotly(**plot_params)
        case _:
            raise ValueError(f"Drawing {engine=} not supported.")

tirex2.plotting.plot_forecast

plot_forecast(context: Tensor | ndarray | None = None, forecasts: Tensor | ndarray | None = None, ground_truth: Tensor | ndarray | None = None, x: Sequence | None = None, quantiles: tuple[float, float] = (0.1, 0.9), quantile_levels: tuple[float] = (0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9), engine='plotly', max_context_to_show: int | None = None, ax=None, fig=None, **kwargs)

Plots the historical context, optional ground-truth future, and forecast.

Parameters:

Name	Type	Description	Default
context	Tensor or ndarray	The historical time series data to be plotted.	None
forecasts	Tensor or ndarray	The forecasts data including quantiles, of shape [Q, N], where Q=9 quantiles are required, and N is the number of forecast timesteps.	None
ground_truth	Tensor or ndarray	The actual future data to compare the forecast against.	None
x	Sequence	X-axis values (e.g., timestamps or indices) for the data. The sequence must be slicable.	None
quantiles	tuple[float]	A tuple indicating the quantile levels to use to plot as shaded areas around the median forecast. Set to None to deactivate. Default is (0.1, 0.9).	(0.1, 0.9)
quantile_levels	tuple[float]	A tuple indicating the quantile levels.	(0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9)
engine	str	What framework to use for rendering the plots.	'plotly'
max_context_to_show	int	If set, limits the number of context points to show for better visibility of forecasts.	None
ax	Axes or Figure	The matplotlib axes / plotly figure object to plot on.	None
**kwargs		Additional keyword arguments to pass to the plotting functions.	{}

Returns:

Type	Description
Axes	The Axes object with the plotted forecast, if engine="matplotlib"
Figure	The Figure object with the plotted forecast, if engine="plotly"

Source code in src/tirex2/plotting.py

def plot_forecast(
    context: torch.Tensor | np.ndarray | None = None,
    forecasts: torch.Tensor | np.ndarray | None = None,
    ground_truth: torch.Tensor | np.ndarray | None = None,
    x: Sequence | None = None,
    quantiles: tuple[float, float] = (0.1, 0.9),
    quantile_levels: tuple[float] = (0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9),
    engine="plotly",
    max_context_to_show: int | None = None,
    ax=None,
    fig=None,
    **kwargs,
):
    """
    Plots the historical context, optional ground-truth future, and forecast.

    Parameters
    ----------
    context : torch.Tensor or np.ndarray
        The historical time series data to be plotted.
    forecasts : torch.Tensor or np.ndarray, optional
        The forecasts data including quantiles, of shape [Q, N],
        where Q=9 quantiles are required, and N is the number of forecast timesteps.
    ground_truth : torch.Tensor or np.ndarray, optional
        The actual future data to compare the forecast against.
    x : Sequence, optional
        X-axis values (e.g., timestamps or indices) for the data. The sequence must be slicable.
    quantiles : tuple[float], optional
        A tuple indicating the quantile levels to use to plot as shaded areas
        around the median forecast. Set to None to deactivate. Default is (0.1, 0.9).
    quantile_levels : tuple[float], optional
        A tuple indicating the quantile levels.
    engine : str, optional
        What framework to use for rendering the plots.
    max_context_to_show : int, optional
        If set, limits the number of context points to show for better visibility of forecasts.
    ax : matplotlib.Axes or plotly.go.Figure, optional
        The matplotlib axes / plotly figure object to plot on.
    **kwargs
        Additional keyword arguments to pass to the plotting functions.
    Returns
    -------
    matplotlib.Axes
        The Axes object with the plotted forecast, if engine="matplotlib"
    plotly.go.Figure
        The Figure object with the plotted forecast, if engine="plotly"
    """

    if context is None and forecasts is None and ground_truth is None:
        raise ValueError("At least one of context, forecasts, or ground_truth must be provided for plotting.")

    if quantiles is not None and len(quantiles) != 2:
        raise ValueError(
            "quantiles must either be a collection of two values for min- and max quantile, respectively, or None."
        )

    # determine all lenghts for clarity
    context_size = len(context) if context is not None else 0
    forecast_size = forecasts.shape[-1] if forecasts is not None else 0
    ground_truth_size = len(ground_truth) if ground_truth is not None else 0
    full_size = context_size + max(forecast_size, ground_truth_size)

    if x is None:
        x = np.arange(full_size)
    elif len(x) < full_size:
        raise ValueError(
            "Not enough 'x' values provided to have one for every timestep in context, forecast, and ground truth window."
        )

    x_context = x[:context_size] if context is not None else None
    if max_context_to_show is not None:
        x_context = x_context[-max_context_to_show:]
        context = context[-max_context_to_show:]

    def connect_to_context(v, is_x_axis=False):
        if CONNECT_FORECAST_TO_CONTEXT and context is not None and len(context) > 0:
            context_data = x_context if is_x_axis else context
            return np.hstack([np.array(context_data)[-1:], v])
        return v

    plot_params = dict(
        context=context,
        x_context=x_context,
        label_context="Context",
        label_ground_truth="Ground Truth Future",
        label_forecast="Forecast (Median)",
    )

    if ground_truth is not None:
        plot_params.update(
            dict(
                ground_truth=connect_to_context(ground_truth, is_x_axis=False),
                x_ground_truth=connect_to_context(x[context_size : context_size + ground_truth_size], is_x_axis=True),
            )
        )

    # plot forecasts if supplied
    # forecasts are a 2D array with quantiles as rows, and data for each timestep as columns
    if forecasts is not None:
        median_index = quantile_levels.index(0.5)
        plot_params["point_forecast"] = connect_to_context(forecasts[median_index, :], is_x_axis=False)
        if quantiles is not None:
            min_quantile, max_quantile = quantiles
            min_quantile_index, max_quantile_index = (quantile_levels.index(q) for q in quantiles)
            plot_params.update(
                dict(
                    x_forecast=connect_to_context(x[context_size : context_size + forecast_size], is_x_axis=True),
                    lower_quantile=connect_to_context(forecasts[min_quantile_index, :], is_x_axis=False),
                    upper_quantile=connect_to_context(forecasts[max_quantile_index, :], is_x_axis=False),
                    label_quantile=f"Forecast {min_quantile * 100:.0f}% - {max_quantile * 100:.0f}% Quantiles",
                )
            )

    match engine:
        case "matplotlib":
            return _plot_forecast_matplotlib(**plot_params, ax=ax, **kwargs)
        case "plotly":
            return _plot_forecast_plotly(**plot_params, fig=fig, **kwargs)
        case _:
            raise ValueError(f"Drawing {engine=} not supported.")

tirex2.plotting.plot_covariate

plot_covariate(covariate: Tensor | ndarray | None, label: str | None = None, color: str = COVARIATE_COLORS[0], x: Sequence | None = None, engine='plotly', ax=None, fig=None, **kwargs)

Source code in src/tirex2/plotting.py

def plot_covariate(
    covariate: torch.Tensor | np.ndarray | None,
    label: str | None = None,
    color: str = COVARIATE_COLORS[0],
    x: Sequence | None = None,
    engine="plotly",
    ax=None,
    fig=None,
    **kwargs,
):
    covariate_size = len(covariate)
    if x is None:
        x = np.arange(covariate_size)
    elif len(x) < covariate_size:
        raise ValueError("Not enough 'x' values provided to have one for every timestep of the covariate.")
    x = x[:covariate_size]

    plot_params = dict(
        x=x,
        covariate=covariate,
        label=label,
        color=color,
    )

    match engine:
        case "matplotlib":
            return _plot_covariate_matplotlib(**plot_params, ax=ax, **kwargs)
        case "plotly":
            return _plot_covariate_plotly(**plot_params, fig=fig, **kwargs)
        case _:
            raise ValueError(f"Drawing {engine=} not supported.")