Feat: add configurable forecast post processing

documentation/changelog.rst

@@ -13,6 +13,7 @@ v1.0.0 | July XX, 2026
 New features
 -------------
 * In the UI, asset and sensor lists can be filtered by ID prefix through API-backed search fields [see `PR #2231 <https://www.github.com/FlexMeasures/flexmeasures/pull/2231>`_]
+* Support configurable lower and upper bounds and snapping for forecast post-processing [see `PR #2273 <https://www.github.com/FlexMeasures/flexmeasures/pull/2273>`_]
 * Floor off-clock API datetimes to a non-instantaneous sensor's resolution by default when ingesting sensor data, uploading sensor data, and handling scheduler flex-model timed events; configurable with the ``floor_datetimes_to_resolution`` sensor attribute [see `PR #2146 <https://www.github.com/FlexMeasures/flexmeasures/pull/2146>`_]
 * Sensor references in flex-model and flex-context support various ways of filtering by source [see `PR #2209 <https://www.github.com/FlexMeasures/flexmeasures/pull/2209>`_]
 * Let storage scheduling infer missing ``power-capacity`` from directional device capacities before falling back to site capacity, and default the missing opposite capacity to zero when only a non-zero ``consumption-capacity`` or ``production-capacity`` is configured [see `PR #2222 <https://www.github.com/FlexMeasures/flexmeasures/pull/2222>`_]

documentation/features/forecasting.rst

@@ -62,6 +62,27 @@ Note that:
 ``forecast-frequency`` together with ``max-forecast-horizon`` determine how the forecasting cycles advance through time.
 ``train-period``, ``from-date`` and ``to-date`` allow precise control over the training and prediction windows in each cycle.
 
+Forecast post-processing
+--------------------------------
+
+Forecaster configuration can clip and snap forecast values before they are stored.
+Use ``lower`` and ``upper`` to clip values to bounds, and ``snap`` to replace values in a configured interval with a target value.
+Units are optional; unitless values are interpreted in the output sensor unit.
+
+For example, this configuration clips forecasts to the 0-20 kW range and snaps values between 0 kW and 4 kW to 0 kW:
+
+.. code-block:: json
+
+    {
+      "lower": "0 kW",
+      "snap": {
+        "0 kW": ["0 kW", "4 kW"]
+      },
+      "upper": "20 kW"
+    }
+
+Pass the same object as the API ``config`` payload, or place it in a JSON or YAML file and pass it to ``flexmeasures add forecasts`` with ``--config``.
+
 Forecasting via the UI
 -----------------------
 
@@ -142,4 +163,3 @@ If you want to take regressors into account, in addition to merely past measurem
 
 Including regressors can significantly improve forecasting accuracy, especially when they are highly correlated with the target variable. For example, using irradiation forecasts as regressors can substantially improve solar production predictions.
 In `this weather forecast plugin <https://github.com/flexmeasures/flexmeasures-weather>`_, we enable you to collect regressor data for ``["temperature", "wind speed", "cloud cover", "irradiance"]``, at a location you select.
-

flexmeasures/data/models/forecasting/pipelines/predict.py

@@ -13,7 +13,10 @@
 
 from flexmeasures import Sensor, Source
 from flexmeasures.data import db
-from flexmeasures.data.models.forecasting.utils import data_to_bdf
+from flexmeasures.data.models.forecasting.utils import (
+    apply_forecast_post_processing,
+    data_to_bdf,
+)
 from flexmeasures.data.models.forecasting.pipelines.base import BasePipeline
 from flexmeasures.data.utils import save_to_db
 
@@ -40,6 +43,7 @@ def __init__(
         predict_end: datetime | None = None,
         data_source: Source = None,
         missing_threshold: float = 1.0,
+        post_processing_config: dict | None = None,
     ) -> None:
         """
         Initialize the PredictPipeline.
@@ -63,6 +67,7 @@ def __init__(
         :param probabilistic: Whether to use a probabilistic model.
         :param sensor_to_save: Sensor to which the predictions will be attributed.
         :param missing_threshold: Max fraction of missing data allowed before failure. Missing data under the threshold will be filled with our interpolation methods.
+        :param post_processing_config: Optional clipping and snapping configuration for forecast values.
         """
         super().__init__(
             future_regressors=future_regressors,
@@ -88,6 +93,7 @@ def __init__(
         self.sensor_to_save = sensor_to_save
         self.predict_start = predict_start
         self.predict_end = predict_end
+        self.post_processing_config = post_processing_config or {}
 
         self.sensor_resolution = self.target_sensor.event_resolution
         self.readable_resolution = duration_isoformat(self.sensor_resolution)
@@ -266,6 +272,12 @@ def run(self, delete_model: bool = False) -> BeliefsDataFrame:
             y_list=y_list,
             belief_timestamps_list=belief_timestamps_list,
         )
+        df_pred = apply_forecast_post_processing(
+            data=df_pred,
+            horizon=self.max_forecast_horizon,
+            config=self.post_processing_config,
+            sensor_unit=self.sensor_to_save.unit,
+        )
         logging.debug("Predictions ready to be saved")
 
         # todo: it looks like data_to_bdf should become a class method

flexmeasures/data/models/forecasting/pipelines/train_predict.py

@@ -164,6 +164,11 @@ def run_cycle(
             sensor_to_save=self._parameters["sensor_to_save"],
             data_source=self.data_source,
             missing_threshold=self._config.get("missing_threshold"),
+            post_processing_config={
+                "lower": self._config.get("lower"),
+                "upper": self._config.get("upper"),
+                "snap": self._config.get("snap", {}),
+            },
         )
         logging.info(
             f"Prediction cycle from {predict_start} to {predict_end} started ..."

flexmeasures/data/models/forecasting/utils.py

@@ -1,5 +1,8 @@
 from __future__ import annotations
 
+import numbers
+from typing import Any
+
 import numpy as np
 import pandas as pd
 import timely_beliefs as tb
@@ -9,12 +12,121 @@
 from datetime import datetime, timedelta
 
 from flexmeasures.data import db
+from flexmeasures.utils.unit_utils import ur
 
 
 def negative_to_zero(x: np.ndarray) -> np.ndarray:
     return np.where(x < 0, 0, x)
 
 
+def _normalize_unit(unit: str | None) -> str:
+    """Normalize empty units for pint."""
+    return unit or "dimensionless"
+
+
+def _is_unitless(unit: str | None) -> bool:
+    """Check whether a parsed quantity carries no physical unit."""
+    return unit in (None, "", "dimensionless")
+
+
+def _quantity_to_sensor_value(value: Any, sensor_unit: str) -> float:
+    """Parse a configured quantity and return its magnitude in the sensor unit."""
+    to_unit = _normalize_unit(sensor_unit)
+
+    if isinstance(value, numbers.Real):
+        return float(value)
+
+    if not isinstance(value, str):
+        raise ValueError(
+            f"Forecast post-processing values must be numbers or quantity strings, not {type(value).__name__}."
+        )
+
+    try:
+        quantity = ur.Quantity(value)
+    except Exception as exc:
+        raise ValueError(
+            f"Could not parse forecast post-processing value '{value}'."
+        ) from exc
+
+    if _is_unitless(f"{quantity.units:~P}"):
+        return float(quantity.magnitude)
+
+    try:
+        return float(quantity.to(ur.Quantity(to_unit)).magnitude)
+    except Exception as exc:
+        raise ValueError(
+            f"Could not convert forecast post-processing value '{value}' to '{sensor_unit}'."
+        ) from exc
+
+
+def apply_forecast_post_processing(
+    data: pd.DataFrame,
+    horizon: int,
+    config: dict,
+    sensor_unit: str,
+) -> pd.DataFrame:
+    """Apply configured clipping and snapping to forecast horizon columns.
+
+    :param data:        DataFrame containing one column per forecast horizon.
+    :param horizon:     Maximum forecast horizon in time-steps.
+    :param config:      Forecaster config with optional ``lower``, ``upper`` and ``snap`` fields.
+    :param sensor_unit: Unit of the sensor to which forecasts will be saved.
+    :returns:           A DataFrame with post-processed forecast values.
+    """
+    lower = config.get("lower")
+    upper = config.get("upper")
+    snap = config.get("snap", {})
+
+    if lower is None and upper is None and not snap:
+        return data
+
+    processed = data.copy()
+    forecast_columns = [f"{h}h" for h in range(1, horizon + 1)]
+    lower_value = (
+        _quantity_to_sensor_value(lower, sensor_unit) if lower is not None else None
+    )
+    upper_value = (
+        _quantity_to_sensor_value(upper, sensor_unit) if upper is not None else None
+    )
+
+    if (
+        lower_value is not None
+        and upper_value is not None
+        and lower_value > upper_value
+    ):
+        raise ValueError(
+            "Forecast post-processing lower bound cannot be greater than upper bound."
+        )
+
+    for target, interval in snap.items():
+        if not isinstance(interval, (list, tuple)) or len(interval) != 2:
+            raise ValueError(
+                "Forecast post-processing snap intervals must contain exactly two bounds."
+            )
+
+        target_value = _quantity_to_sensor_value(target, sensor_unit)
+        interval_lower = _quantity_to_sensor_value(interval[0], sensor_unit)
+        interval_upper = _quantity_to_sensor_value(interval[1], sensor_unit)
+        if interval_lower > interval_upper:
+            raise ValueError(
+                "Forecast post-processing snap interval lower bound cannot be greater than upper bound."
+            )
+
+        for column in forecast_columns:
+            original_values = processed[column].copy()
+            mask = original_values.between(interval_lower, interval_upper)
+            if lower_value is not None:
+                mask &= original_values >= lower_value
+            if upper_value is not None:
+                mask &= original_values <= upper_value
+            processed.loc[mask, column] = target_value
+
+    processed[forecast_columns] = processed[forecast_columns].clip(
+        lower=lower_value, upper=upper_value, axis=None
+    )
+    return processed
+
+
 def data_to_bdf(
     data: pd.DataFrame,
     horizon: int,

flexmeasures/data/schemas/forecasting/pipeline.py

@@ -9,6 +9,7 @@
 from marshmallow import (
     fields,
     Schema,
+    validate,
     validates_schema,
     pre_load,
     post_load,
@@ -98,6 +99,34 @@ class TrainPredictPipelineConfigSchema(Schema):
             },
         },
     )
+    lower = fields.Raw(
+        required=False,
+        allow_none=True,
+        load_default=None,
+        metadata={
+            "description": "Optional lower bound for forecast values after prediction. Unitless values are interpreted in the output sensor unit.",
+            "example": "0 kW",
+        },
+    )
+    upper = fields.Raw(
+        required=False,
+        allow_none=True,
+        load_default=None,
+        metadata={
+            "description": "Optional upper bound for forecast values after prediction. Unitless values are interpreted in the output sensor unit.",
+            "example": "20 kW",
+        },
+    )
+    snap = fields.Dict(
+        keys=fields.Raw(),
+        values=fields.List(fields.Raw(), validate=validate.Length(equal=2)),
+        required=False,
+        load_default={},
+        metadata={
+            "description": "Optional mapping from snap targets to [lower, upper] intervals. Values inside an interval are replaced by the snap target before storage.",
+            "example": {"0 kW": ["0 kW", "4 kW"]},
+        },
+    )
     train_start = AwareDateTimeOrDateField(
         data_key="train-start",
         required=False,

flexmeasures/data/schemas/tests/test_forecasting.py

@@ -643,3 +643,36 @@ def test_timing_parameters_of_forecaster_config_schema(
         # Convert kebab-case key to snake_case to match data dictionary keys returned by schema
         snake_key = kebab_to_snake(k)
         assert data[snake_key] == v, f"{k} did not match expectations."
+
+
+def test_forecaster_config_schema_loads_forecast_post_processing_options():
+    data = TrainPredictPipelineConfigSchema().load(
+        {
+            "lower": "0 kW",
+            "upper": 20,
+            "snap": {"0 kW": ["0 kW", "4 kW"]},
+        }
+    )
+
+    assert data["lower"] == "0 kW"
+    assert data["upper"] == 20
+    assert data["snap"] == {"0 kW": ["0 kW", "4 kW"]}
+
+
+def test_forecaster_config_schema_defaults_forecast_post_processing_to_disabled():
+    data = TrainPredictPipelineConfigSchema().load({})
+
+    assert data["lower"] is None
+    assert data["upper"] is None
+    assert data["snap"] == {}
+
+
+def test_forecaster_config_schema_rejects_invalid_snap_interval_shape():
+    with pytest.raises(ValidationError) as exc:
+        TrainPredictPipelineConfigSchema().load(
+            {
+                "snap": {"0 kW": ["0 kW"]},
+            }
+        )
+
+    assert "snap" in exc.value.messages