Source code for openstef_models.models.forecasting.flatliner_forecaster

# SPDX-FileCopyrightText: 2025 Contributors to the OpenSTEF project <openstef@lfenergy.org>
#
# SPDX-License-Identifier: MPL-2.0

"""Simple constant zero forecasting model.

Provides basic forecasting model that predict constant flatliner zero values. It can be used
when a flatline (non-)zero measurement is observed in the past and expected in the future.
"""

from datetime import timedelta
from typing import cast, override

import pandas as pd
from pydantic import Field, PrivateAttr

from openstef_core.datasets.validated_datasets import ForecastDataset, ForecastInputDataset, TimeSeriesDataset
from openstef_core.mixins.predictor import HyperParams
from openstef_models.explainability.mixins import ContributionsMixin, ExplainableForecaster
from openstef_models.models.forecasting.forecaster import Forecaster

MODEL_CODE_VERSION = 1




class FlatlinerForecaster(Forecaster, ExplainableForecaster, ContributionsMixin):
    """Flatliner forecaster that predicts a flatline of zeros or median.

    A simple forecasting model that always predicts zero (or a median load value if
    configured) for all horizons and quantiles. The median can be calculated over
    a trailing window when the fallback is responding to a recent flatline.

    Invariants:
        - Configuration quantiles determine the number of prediction outputs
        - Zeros (or median values) are predicted for all horizons and quantiles

    Example:
        >>> from openstef_core.types import LeadTime, Quantile
        >>> from datetime import timedelta
        >>> forecaster = FlatlinerForecaster(
        ...     quantiles=[Quantile(0.5), Quantile(0.1), Quantile(0.9)],
        ...     horizons=[LeadTime(timedelta(hours=1)), LeadTime(timedelta(hours=2))],
        ... )
        >>> forecaster.fit(training_data)  # doctest: +SKIP
        >>> predictions = forecaster.predict(test_data)  # doctest: +SKIP

    See Also:
        Forecaster: Base class for forecasting models that predict multiple horizons.
    """

    predict_median: bool = Field(
        default=False,
        description="If True, predict the median of load measurements instead of zero.",
    )
    median_window: timedelta | None = Field(
        default=None,
        description="Optional trailing window for the median load estimate; None uses all available measurements.",
    )

    hyperparams: HyperParams = Field(
        default_factory=HyperParams,
        description="Model hyperparameters (no tuning parameters for flatliner).",
    )

    _median_value: float | None = PrivateAttr(default=None)

    @property
    @override
    def hparams(self) -> HyperParams:
        return self.hyperparams

    @property
    @override
    def is_fitted(self) -> bool:
        if self.predict_median:
            return self._median_value is not None
        return True



    @override
    def fit(
        self,
        data: ForecastInputDataset,
        data_val: ForecastInputDataset | None = None,
    ) -> None:
        if self.predict_median:
            target_series = data.target_series
            if self.median_window is not None:
                last_valid_index = cast(pd.Timestamp | None, target_series.last_valid_index())
                if last_valid_index is not None:
                    window_start = last_valid_index - self.median_window
                    target_series = target_series.loc[window_start:last_valid_index]
            self._median_value = float(target_series.median())




    @override
    def predict(self, data: ForecastInputDataset) -> ForecastDataset:
        forecast_index = data.create_forecast_range(horizon=self.max_horizon)

        prediction_value = self._median_value if self.predict_median else 0.0

        return ForecastDataset(
            data=pd.DataFrame(
                data={quantile.format(): prediction_value for quantile in self.quantiles},
                index=forecast_index,
            ),
            sample_interval=data.sample_interval,
            target_column=data.target_column,
        )


    @property
    @override
    def feature_importances(self) -> pd.DataFrame:
        columns = [quantile.format() for quantile in self.quantiles]
        return pd.DataFrame(
            data=[[0.0] * len(columns)],
            index=["load"],
            columns=columns,
        )



    @override
    def predict_contributions(self, data: ForecastInputDataset) -> TimeSeriesDataset:
        """Return zero contributions since flatliner has no features."""
        input_data = data.input_data(start=data.forecast_start)
        contribs_df = pd.DataFrame(0.0, index=input_data.index, columns=["bias"])
        return TimeSeriesDataset(data=contribs_df, sample_interval=data.sample_interval)