diff --git a/documentation/changelog.rst b/documentation/changelog.rst index 0f41c0a1d4..1b5e5f8e47 100644 --- a/documentation/changelog.rst +++ b/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 `_] +* Support configurable lower and upper bounds and snapping for forecast post-processing [see `PR #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 `_] * Sensor references in flex-model and flex-context support various ways of filtering by source [see `PR #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 `_] diff --git a/documentation/features/forecasting.rst b/documentation/features/forecasting.rst index 8839bf34c0..5d9eb73822 100644 --- a/documentation/features/forecasting.rst +++ b/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 `_, we enable you to collect regressor data for ``["temperature", "wind speed", "cloud cover", "irradiance"]``, at a location you select. - diff --git a/flexmeasures/data/models/forecasting/pipelines/predict.py b/flexmeasures/data/models/forecasting/pipelines/predict.py index 54ae993059..011444fcf0 100644 --- a/flexmeasures/data/models/forecasting/pipelines/predict.py +++ b/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 diff --git a/flexmeasures/data/models/forecasting/pipelines/train_predict.py b/flexmeasures/data/models/forecasting/pipelines/train_predict.py index 88883e132c..cac519e6ff 100644 --- a/flexmeasures/data/models/forecasting/pipelines/train_predict.py +++ b/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 ..." diff --git a/flexmeasures/data/models/forecasting/utils.py b/flexmeasures/data/models/forecasting/utils.py index 0b474e5694..477c09db3d 100644 --- a/flexmeasures/data/models/forecasting/utils.py +++ b/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, diff --git a/flexmeasures/data/schemas/forecasting/pipeline.py b/flexmeasures/data/schemas/forecasting/pipeline.py index 1d4fe62d7d..39adc94771 100644 --- a/flexmeasures/data/schemas/forecasting/pipeline.py +++ b/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, diff --git a/flexmeasures/data/schemas/tests/test_forecasting.py b/flexmeasures/data/schemas/tests/test_forecasting.py index cdd7686f6e..31e6281170 100644 --- a/flexmeasures/data/schemas/tests/test_forecasting.py +++ b/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 diff --git a/flexmeasures/data/tests/test_forecasting_pipeline.py b/flexmeasures/data/tests/test_forecasting_pipeline.py index ac63643606..dad836ecf6 100644 --- a/flexmeasures/data/tests/test_forecasting_pipeline.py +++ b/flexmeasures/data/tests/test_forecasting_pipeline.py @@ -11,6 +11,9 @@ from flexmeasures.data.models.forecasting.custom_models.lgbm_model import CustomLGBM from flexmeasures.data.models.data_sources import DataSource from flexmeasures.data.models.forecasting.exceptions import NotEnoughDataException +from flexmeasures.data.models.forecasting.utils import ( + apply_forecast_post_processing, +) from flexmeasures.data.models.forecasting.pipelines.base import BasePipeline from flexmeasures.data.models.forecasting.pipelines.train import derive_daily_lag_steps from flexmeasures.data.models.generic_assets import ( @@ -89,6 +92,104 @@ def test_derive_daily_lag_steps_requires_divisible_resolution(caplog): ) +def test_forecast_post_processing_clips_and_snaps_values(): + df = pd.DataFrame( + { + "1h": [-2.0, 2.0, 8.5, 11.5, 21.0], + "2h": [0.5, 3.5, 10.5, 12.5, 25.0], + } + ) + + processed = apply_forecast_post_processing( + data=df, + horizon=2, + config={ + "lower": "0 kW", + "snap": { + "0 kW": ["0 kW", "4 kW"], + "8 kW": ["8 kW", "11 kW"], + "12 kW": ["11 kW", "12 kW"], + }, + "upper": "20 kW", + }, + sensor_unit="kW", + ) + + pd.testing.assert_frame_equal( + processed, + pd.DataFrame( + { + "1h": [0.0, 0.0, 8.0, 12.0, 20.0], + "2h": [0.0, 0.0, 8.0, 12.5, 20.0], + } + ), + ) + + +def test_forecast_post_processing_interprets_unitless_values_in_sensor_unit(): + df = pd.DataFrame({"1h": [0.5, 1.3, 2.5]}) + + processed = apply_forecast_post_processing( + data=df, + horizon=1, + config={ + "lower": 1, + "snap": {"1500 W": ["1.2 kW", "1.8 kW"]}, + "upper": "2 kW", + }, + sensor_unit="kW", + ) + + pd.testing.assert_frame_equal( + processed, + pd.DataFrame({"1h": [1.0, 1.5, 2.0]}), + ) + + +def test_forecast_post_processing_interprets_unitless_snap_in_sensor_unit(): + df = pd.DataFrame({"1h": [0.5, 2.5, 5.0]}) + + processed = apply_forecast_post_processing( + data=df, + horizon=1, + config={"snap": {"0": [0, 4]}}, + sensor_unit="kW", + ) + + pd.testing.assert_frame_equal( + processed, + pd.DataFrame({"1h": [0.0, 0.0, 5.0]}), + ) + + +def test_forecast_post_processing_converts_snap_units_to_sensor_unit(): + df = pd.DataFrame({"1h": [1.1, 1.5, 1.9]}) + + processed = apply_forecast_post_processing( + data=df, + horizon=1, + config={"snap": {"1000 W": ["1200 W", "1800 W"]}}, + sensor_unit="kW", + ) + + pd.testing.assert_frame_equal( + processed, + pd.DataFrame({"1h": [1.1, 1.0, 1.9]}), + ) + + +def test_forecast_post_processing_rejects_inconsistent_bounds(): + df = pd.DataFrame({"1h": [1.0]}) + + with pytest.raises(ValueError, match="lower bound cannot be greater"): + apply_forecast_post_processing( + data=df, + horizon=1, + config={"lower": "2 kW", "upper": "1 kW"}, + sensor_unit="kW", + ) + + @pytest.mark.parametrize( ["config", "params", "as_job", "expected_error"], [ diff --git a/flexmeasures/ui/static/openapi-specs.json b/flexmeasures/ui/static/openapi-specs.json index 599e06d878..34ffe9686f 100644 --- a/flexmeasures/ui/static/openapi-specs.json +++ b/flexmeasures/ui/static/openapi-specs.json @@ -4767,6 +4767,33 @@ "default": false, "description": "Whether to clip negative values in forecasts. Defaults to None (disabled).", "example": true + }, + "lower": { + "default": null, + "description": "Optional lower bound for forecast values after prediction. Unitless values are interpreted in the output sensor unit.", + "example": "0 kW" + }, + "upper": { + "default": null, + "description": "Optional upper bound for forecast values after prediction. Unitless values are interpreted in the output sensor unit.", + "example": "20 kW" + }, + "snap": { + "type": "object", + "default": {}, + "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" + ] + }, + "additionalProperties": { + "type": "array", + "minItems": 2, + "maxItems": 2, + "items": {} + } } }, "additionalProperties": false