Small forecasting fixes: forecast-only future regressors and dict-typed CLI options

documentation/changelog.rst

@@ -108,6 +108,8 @@ Infrastructure / Support
 
 Bugfixes
 -----------
+* Future regressors from sensors that only ever record forecasts (belief time never after the event start, e.g. day-ahead market fundamentals) no longer drop out of the forecasting pipeline's training window entirely; the training window now falls back to the latest forecast per event where no realized belief exists [see `PR #2373 <https://www.github.com/FlexMeasures/flexmeasures/pull/2373>`_]
+* Dict-typed CLI options such as ``flexmeasures add forecasts --model-params`` now parse their JSON (or Python-literal) argument instead of failing schema validation with "Not a valid mapping type"; an argument that parses to something other than a mapping is now reported as such by the option itself [see `PR #2373 <https://www.github.com/FlexMeasures/flexmeasures/pull/2373>`_]
 * Include the Excel reader in default installations so XLSX sensor-data uploads work outside test environments [see `PR #2376 <https://www.github.com/FlexMeasures/flexmeasures/pull/2376>`_]
 * In a multi-device flex-model, a device without a stock (e.g. a converter port or curtailable generator) silently disabled constraint validation for all devices after it; validation now covers every device, and also newly checks that each device's power bounds do not contradict each other, so a contradictory hard bound fails with a clear per-time-step message instead of a bare solver infeasibility [see `PR #2252 <https://www.github.com/FlexMeasures/flexmeasures/pull/2252>`_]
 * The scheduler now rejects a commitment that no constraint would bind — a stock commitment naming no device or known stock group, or a commodity commitment for a commodity that no commitment maps devices to — instead of silently dropping it from the problem, or letting a favourably priced deviation make the problem unbounded; the error names the commitment [see `PR #2410 <https://www.github.com/FlexMeasures/flexmeasures/pull/2410>`_ and `PR #2413 <https://www.github.com/FlexMeasures/flexmeasures/pull/2413>`_]

flexmeasures/cli/tests/test_utils.py

@@ -147,6 +147,59 @@ def cmd(name):
     assert result.output.endswith("foo\n")
 
 
+def test_add_cli_options_from_schema_parses_dict_fields():
+    """A plain ``fields.Dict`` option (e.g. ``--model-params``) must arrive as a dict,
+    in both JSON and Python-literal syntax, like nested list fields do.
+    """
+    from flexmeasures.cli.utils import add_cli_options_from_schema
+    from flexmeasures.data.schemas.forecasting.pipeline import (
+        TrainPredictPipelineConfigSchema,
+    )
+
+    captured = {}
+
+    @click.command()
+    @add_cli_options_from_schema(TrainPredictPipelineConfigSchema())
+    def cmd(**kwargs):
+        captured.update(kwargs)
+
+    result = CliRunner().invoke(cmd, ["--model-params", '{"min_child_samples": 5}'])
+    assert result.exit_code == 0, result.output
+    assert captured["model_params"] == {"min_child_samples": 5}
+    # The parsed dict must pass schema validation,
+    # which used to reject the raw string with "Not a valid mapping type."
+    config = TrainPredictPipelineConfigSchema().load(
+        {"model-params": captured["model_params"]}
+    )
+    assert config["model_params"] == {"min_child_samples": 5}
+
+    result = CliRunner().invoke(cmd, ["--model-params", "{'max_depth': 6}"])
+    assert result.exit_code == 0, result.output
+    assert captured["model_params"] == {"max_depth": 6}
+
+
+@pytest.mark.parametrize("bad_value", ["[1, 2]", "5", '"a string"'])
+def test_add_cli_options_from_schema_rejects_non_mapping_dict_fields(bad_value):
+    """A parsable but non-mapping argument must be rejected by Click itself.
+
+    Such a value would otherwise reach Marshmallow, which reports the unhelpful "Not a valid mapping type".
+    """
+    from flexmeasures.cli.utils import add_cli_options_from_schema
+    from flexmeasures.data.schemas.forecasting.pipeline import (
+        TrainPredictPipelineConfigSchema,
+    )
+
+    @click.command()
+    @add_cli_options_from_schema(TrainPredictPipelineConfigSchema())
+    def cmd(**kwargs):
+        pass
+
+    result = CliRunner().invoke(cmd, ["--model-params", bad_value])
+    assert result.exit_code == 2, result.output
+    assert "Expected a mapping" in result.output
+    assert "Not a valid mapping type" not in result.output
+
+
 @pytest.mark.xfail(
     strict=True,
     raises=RuntimeError,

flexmeasures/cli/utils.py

@@ -409,7 +409,8 @@ class NestedDictParamType(click.ParamType):
 
     Accepts both JSON double-quoted syntax (``{"key": "value"}``) and Python-literal
     single-quoted syntax (``{'key': 'value'}``).  Used for CLI options whose Marshmallow
-    field type is ``fields.List(fields.Nested(...))``.
+    field type is ``fields.List(fields.Nested(...))`` (one dict per occurrence) or
+    ``fields.Dict`` (a single dict).
     """
 
     name = "DICT"
@@ -418,16 +419,25 @@ def convert(self, value, param, ctx):
         if isinstance(value, dict):
             return value
         try:
-            return json.loads(value)
+            parsed = json.loads(value)
         except json.JSONDecodeError:
             try:
-                return ast.literal_eval(value)
+                parsed = ast.literal_eval(value)
             except (ValueError, SyntaxError):
                 self.fail(
                     f"Cannot parse as a JSON object or Python-literal dict: {value!r}",
                     param,
                     ctx,
                 )
+        # A parsable non-object (e.g. a list or a number) would otherwise travel on,
+        # only to be rejected further downstream by Marshmallow with "Not a valid mapping type".
+        if not isinstance(parsed, dict):
+            self.fail(
+                f'Expected a mapping such as \'{{"key": "value"}}\', but got {type(parsed).__name__}: {value!r}',
+                param,
+                ctx,
+            )
+        return parsed
 
 
 class JSONOrFile(click.ParamType):
@@ -535,6 +545,9 @@ def decorator(command):
                     kwargs["type"] = NestedDictParamType()
                 else:
                     kwargs["type"] = str
+            elif isinstance(field, fields.Dict):
+                # The value is a single dict string; parse it at the Click level.
+                kwargs["type"] = NestedDictParamType()
 
             command = click.option(*options, **kwargs)(command)
 

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

@@ -709,17 +709,30 @@ def _latest_known_per_regressor(
                 regressor_columns: list[str],
                 forecast_belief_time: pd.Timestamp,
                 realized_only: bool = False,
+                fall_back_to_forecast: bool = False,
             ) -> pd.DataFrame:
-                """Select latest regressor values known at forecast belief time."""
+                """Select latest regressor values known at forecast belief time.
+
+                :param realized_only:           Keep only ex-post beliefs (belief time after the event started);
+                                                otherwise, keep only ex-ante beliefs.
+                :param fall_back_to_forecast:   Also keep ex-ante beliefs for events without any ex-post belief,
+                                                which is ignored unless ``realized_only`` is set.
+                                                Some regressor sensors only ever record ex-ante beliefs (e.g. day-ahead market fundamentals),
+                                                and would otherwise yield no values at all.
+                """
                 keep = ["event_start", *regressor_columns]
                 if df_.empty:
                     return df_.iloc[0:0][keep].copy()
 
                 known = df_.loc[df_["belief_time"] <= forecast_belief_time].copy()
-                if realized_only:
+                if realized_only and not fall_back_to_forecast:
                     known = known.loc[known["belief_time"] > known["event_start"]]
-                else:
+                elif not realized_only:
                     known = known.loc[known["belief_time"] <= known["event_start"]]
+                # With realized_only and fall_back_to_forecast, both ex-post and ex-ante beliefs are kept:
+                # for any given event, an ex-post belief time necessarily exceeds every ex-ante belief time,
+                # so selecting the latest belief per event prefers realized values,
+                # and falls back to forecasts only where no realized value exists.
                 if known.empty:
                     return df_.iloc[0:0][keep].copy()
 
@@ -840,12 +853,15 @@ def _overlay_annotations(
                     # values would hide it from the training window entirely. Its
                     # visibility is governed solely by its own belief time, applied
                     # below once the sensor-based frame has been assembled.
+                    # Forecast-only sensors have no realized rows (belief time never after the event start, e.g. day-ahead fundamentals),
+                    # so the training window falls back to their latest forecasts.
                     future_regressor_columns = self.future_regressors
                     future_known = _latest_known_per_regressor(
                         X_future_regressors_df,
                         future_regressor_columns,
                         belief_time,
                         realized_only=True,
+                        fall_back_to_forecast=True,
                     )
                     realized_slice = _slice_closed(
                         future_known, target_start, target_end

flexmeasures/data/tests/test_forecasting_pipeline.py

@@ -1724,6 +1724,124 @@ def capture_frame(self, df, sensors, sensor_names, start, end, **kwargs):
     assert 77.0 not in set(values_by_event)
 
 
+def test_forecast_only_future_regressor_populates_training_window(monkeypatch):
+    """A future regressor may only ever record ex-ante beliefs (e.g. day-ahead market fundamentals, whose belief time never passes the event start).
+
+    The training window must then fall back to the latest forecasts instead of coming out empty,
+    while realized values still win where they exist.
+    """
+    target_sensor = type(
+        "SensorStub",
+        (),
+        {"name": "target", "id": 1, "event_resolution": timedelta(hours=1)},
+    )()
+    forecast_only_regressor = type(
+        "SensorStub",
+        (),
+        {"name": "residual-load", "id": 2, "event_resolution": timedelta(hours=1)},
+    )()
+    revised_regressor = type(
+        "SensorStub",
+        (),
+        {"name": "weather", "id": 3, "event_resolution": timedelta(hours=1)},
+    )()
+
+    pipeline = BasePipeline(
+        target_sensor=target_sensor,
+        future_regressors=[forecast_only_regressor, revised_regressor],
+        past_regressors=[],
+        n_steps_to_predict=1,
+        max_forecast_horizon=1,
+        forecast_frequency=1,
+        event_starts_after=datetime(2025, 1, 8, 6),
+        event_ends_before=datetime(2025, 1, 8, 10),
+    )
+    regressor_a, regressor_b = pipeline.future_regressors
+
+    day_ahead = pd.Timedelta(hours=21)
+    rows = []
+    for hour, value_a, value_b in [
+        (6, 1.0, 10.0),
+        (7, 2.0, 20.0),
+        (8, 3.0, 30.0),
+        (9, 4.0, 40.0),
+        (10, 5.0, 50.0),
+        (11, 6.0, 60.0),
+    ]:
+        event_start = pd.Timestamp(f"2025-01-08T{hour:02d}:00:00")
+        # Day-ahead beliefs only: each belief precedes its event start.
+        rows.append(
+            {
+                "event_start": event_start,
+                "belief_time": event_start - day_ahead,
+                pipeline.target: None,
+                regressor_a: value_a,
+                regressor_b: value_b,
+            }
+        )
+        if hour <= 9:
+            # The target realizes ex post, but neither regressor does here,
+            # so these rows must not shadow the day-ahead regressor beliefs.
+            rows.append(
+                {
+                    "event_start": event_start,
+                    "belief_time": event_start + pd.Timedelta(minutes=30),
+                    pipeline.target: 100.0 + hour,
+                    regressor_a: None,
+                    regressor_b: None,
+                }
+            )
+    # Regressor B alone gets one realized revision within the training window.
+    rows.append(
+        {
+            "event_start": pd.Timestamp("2025-01-08T07:00:00"),
+            "belief_time": pd.Timestamp("2025-01-08T08:00:00"),
+            pipeline.target: None,
+            regressor_a: None,
+            regressor_b: 25.0,
+        }
+    )
+    df = pd.DataFrame(rows)
+
+    captured_future_frames = []
+
+    # Capture the covariate frame before missing-value filling converts it to a Darts TimeSeries.
+    # This keeps the test focused on in-memory belief selection,
+    # instead of requiring database-backed sensor data.
+    def capture_frame(self, df, sensors, sensor_names, start, end, **kwargs):
+        if sensor_names == self.future_regressors:
+            captured_future_frames.append(df.copy())
+        return df
+
+    monkeypatch.setattr(BasePipeline, "detect_and_fill_missing_values", capture_frame)
+
+    pipeline.split_data_all_beliefs(df)
+
+    assert len(captured_future_frames) == 1, (
+        "Expected one future-covariate frame because this one-step pipeline "
+        "prepares exactly one split."
+    )
+    selected = captured_future_frames[0].set_index("event_start")
+    for hour, expected in [(6, 1.0), (7, 2.0), (8, 3.0), (9, 4.0)]:
+        assert (
+            selected.loc[pd.Timestamp(f"2025-01-08T{hour:02d}:00:00"), regressor_a]
+            == expected
+        ), (
+            "Expected the forecast-only regressor's day-ahead values to fill "
+            "the training window, because no realized beliefs exist to prefer."
+        )
+    assert selected.loc[pd.Timestamp("2025-01-08T10:00:00"), regressor_a] == 5.0
+    assert selected.loc[pd.Timestamp("2025-01-08T11:00:00"), regressor_a] == 6.0
+    assert selected.loc[pd.Timestamp("2025-01-08T07:00:00"), regressor_b] == 25.0, (
+        "Expected the realized revision to win over the day-ahead belief for "
+        "the same event, because its belief time is necessarily later."
+    )
+    assert selected.loc[pd.Timestamp("2025-01-08T06:00:00"), regressor_b] == 10.0, (
+        "Expected the day-ahead fall-back to apply per event, so a realized "
+        "revision for one event does not affect its neighbours."
+    )
+
+
 def test_annotation_regressor_split_preserves_annotation_columns(monkeypatch):
     target_sensor = type(
         "SensorStub",