Feat/optional training start

documentation/changelog.rst

@@ -12,7 +12,7 @@ v0.31.0 | February XX, 2026
 New features
 -------------
 * Improve CSV upload validation by inferring the intended base resolution even when data contains valid gaps, instead of requiring perfectly regular timestamps [see `PR #1918 <https://www.github.com/FlexMeasures/flexmeasures/pull/1918>`_]
-* New forecasting API endpoints `[POST] /sensors/(id)/forecasts/trigger <api/v3_0.html#post--api-v3_0-sensors-id-forecasts-trigger>`_ and `[GET] /sensors/(id)/forecasts/(uuid) <api/v3_0.html#get--api-v3_0-sensors-id-forecasts-uuid>`_ to forecast sensor data [see `PR #1813 <https://www.github.com/FlexMeasures/flexmeasures/pull/1813>`_ and `PR #1823 <https://www.github.com/FlexMeasures/flexmeasures/pull/1823>`_]
+* New forecasting API endpoints, and all timing parameters in forecasting CLI got sensible defaults for ease of use `[POST] /sensors/(id)/forecasts/trigger <api/v3_0.html#post--api-v3_0-sensors-id-forecasts-trigger>`_ and `[GET] /sensors/(id)/forecasts/(uuid) <api/v3_0.html#get--api-v3_0-sensors-id-forecasts-uuid>`_ to forecast sensor data [see `PR #1813 <https://www.github.com/FlexMeasures/flexmeasures/pull/1813>`_, `PR #1823 <https://www.github.com/FlexMeasures/flexmeasures/pull/1823>`_ and `PR #1917 <https://www.github.com/FlexMeasures/flexmeasures/pull/1917>`_]
 * Support setting a resolution when triggering a schedule via the API or CLI [see `PR #1857 <https://www.github.com/FlexMeasures/flexmeasures/pull/1857>`_]
 * Support variable peak pricing and changes in commitment baselines [see `PR #1835 <https://www.github.com/FlexMeasures/flexmeasures/pull/1835>`_]
 * Support storing the aggregate power schedule [see `PR #1736 <https://www.github.com/FlexMeasures/flexmeasures/pull/1736>`_]

documentation/cli/change_log.rst

@@ -10,6 +10,7 @@ since v0.31.0 | February XX, 2026
 * Fix ``delete-beliefs`` CLI command ignoring the ``--source`` filter during deletion, preventing unintended removal of beliefs from other sources.
 * Let ``flexmeasures add schedule`` create schedules with only information known prior to some time using the ``prior`` option.
 * New ``-dry-run`` flag for ``flexmeasures add schedule`` to avoid saving anything (printing out the results instead).
+* Streamlines option names for ``flexmeasures add forecasts`` with API usage (preserving backwards compatibility).
 * Return validation errors instead of database errors for fields that map to database objects.
 * Mutate job state when running ``flexmeasures jobs run-job <job ID>``, including updating metadata and moving between registries
 * Add ``flexmeasures jobs stats``, which shows queueing statistics to help evaluate the health of the queueing system.

documentation/features/forecasting.rst

@@ -55,12 +55,12 @@ The main CLI parameters that control this process are:
 - ``to-date``: The global cutoff point. Training and prediction cycles continue until the ``predict-end`` reaches this date.
 - ``max-forecast-horizon``: The maximum length of a forecast into the future.
 - ``forecast-frequency``: Determines the number of prediction cycles within the forecast period (e.g. daily, hourly).
-- ``start-date``: Define the start of historical data used for training.
+- ``train-period``: Define a window of historical data to use for training.
 
 Note that:
 
 ``forecast-frequency`` together with ``max-forecast-horizon`` determine how the forecasting cycles advance through time.
-``start-date`` / ``from-date`` and ``to-date`` allow precise control over the training and prediction windows in each cycle.
+``train-period``, ``from-date`` and ``to-date`` allow precise control over the training and prediction windows in each cycle.
 
 Forecasting via the API
 -----------------------

documentation/tut/forecasting_scheduling.rst

@@ -104,9 +104,8 @@ There are two ways to queue a forecasting job:
    .. code-block:: json
 
        {
-         "start_date": "2025-01-01T00:00:00+00:00",
-         "start_predict_date": "2025-01-04T00:00:00+00:00",
-         "end_date": "2025-01-04T04:00:00+00:00"
+         "start": "2025-01-04T00:00:00+00:00",
+         "duration": "PT4H"
        }
 
    Example response:

flexmeasures/api/common/responses.py

@@ -260,6 +260,11 @@ def unknown_schedule(message: str) -> ResponseTuple:
     return dict(result="Rejected", status="UNKNOWN_SCHEDULE", message=message), 400
 
 
+@BaseMessage("No known forecast for this time period.")
+def unknown_forecast(message: str) -> ResponseTuple:
+    return dict(result="Rejected", status="UNKNOWN_FORECAST", message=message), 400
+
+
 def fallback_schedule_redirect(message: str, location: str) -> ResponseTuple:
     return (
         dict(result="Rejected", status="UNKNOWN_SCHEDULE", message=message),

flexmeasures/api/common/schemas/utils.py

@@ -4,14 +4,18 @@
 from marshmallow import Schema, fields
 
 from flexmeasures.utils.doc_utils import rst_to_openapi
+from flexmeasures.data.schemas.forecasting.pipeline import (
+    ForecastingTriggerSchema,
+    TrainPredictPipelineConfigSchema,
+)
 from flexmeasures.data.schemas.sensors import (
     SensorReferenceSchema,
     VariableQuantityField,
     VariableQuantityOpenAPISchema,
 )
 
 
-def make_openapi_compatible(schema_cls: Type[Schema]) -> Type[Schema]:
+def make_openapi_compatible(schema_cls: Type[Schema]) -> Type[Schema]:  # noqa: C901
     """
     Create an OpenAPI-compatible version of a Marshmallow schema.
 
@@ -27,6 +31,25 @@ def make_openapi_compatible(schema_cls: Type[Schema]) -> Type[Schema]:
     new_fields = {}
     for name, field in schema_cls._declared_fields.items():
 
+        if schema_cls in (ForecastingTriggerSchema, TrainPredictPipelineConfigSchema):
+            if "cli" in field.metadata and field.metadata["cli"].get(
+                "cli-exclusive", False
+            ):
+                continue
+            if isinstance(field, fields.Nested):
+                nested_schema_cls = type(field.schema)
+                if nested_schema_cls is TrainPredictPipelineConfigSchema:
+                    field_copy = fields.Nested(
+                        make_openapi_compatible(nested_schema_cls),
+                        metadata=field.metadata,
+                        data_key=field.data_key,
+                        many=field.many,
+                        required=field.required,
+                        allow_none=field.allow_none,
+                    )
+                    new_fields[name] = field_copy
+                    continue
+
         # Copy metadata, but sanitize description for OpenAPI
         metadata = dict(getattr(field, "metadata", {}))
         if "description" in metadata:

flexmeasures/api/v3_0/__init__.py

@@ -15,7 +15,10 @@
 from marshmallow import Schema
 
 from flexmeasures import __version__ as fm_version
-from flexmeasures.api.v3_0.sensors import SensorAPI
+from flexmeasures.api.v3_0.sensors import (
+    SensorAPI,
+    forecasting_trigger_schema_openAPI,
+)
 from flexmeasures.api.v3_0.accounts import AccountAPI
 from flexmeasures.api.v3_0.users import UserAPI
 from flexmeasures.api.v3_0.assets import AssetAPI, AssetTypesAPI
@@ -137,6 +140,7 @@ def create_openapi_specs(app: Flask):
     # Explicitly register OpenAPI-compatible schemas
     schemas = [
         ("FlexContextOpenAPISchema", flex_context_schema_openAPI),
+        ("forecasting_trigger_schema_openAPI", forecasting_trigger_schema_openAPI),
         ("UserAPIQuerySchema", UserAPIQuerySchema),
         ("AssetAPIQuerySchema", AssetAPIQuerySchema),
         ("AssetSchema", AssetSchema),

flexmeasures/api/v3_0/sensors.py

@@ -22,10 +22,12 @@
 from flexmeasures.api.common.responses import (
     request_processed,
     unrecognized_event,
+    unknown_forecast,
     unknown_schedule,
     unprocessable_entity,
     fallback_schedule_redirect,
 )
+from flexmeasures.api.common.schemas.utils import make_openapi_compatible
 from flexmeasures.api.common.utils.validators import (
     optional_duration_accepted,
 )
@@ -71,14 +73,27 @@
 from flexmeasures.data.models.forecasting import Forecaster
 from flexmeasures.data.services.data_sources import get_data_generator
 from flexmeasures.data.schemas.forecasting.pipeline import (
-    ForecasterParametersSchema,
+    ForecastingTriggerSchema,
 )
 
 # Instantiate schemes outside of endpoint logic to minimize response time
 sensors_schema = SensorSchema(many=True)
 sensor_schema = SensorSchema()
 partial_sensor_schema = SensorSchema(partial=True, exclude=["generic_asset_id"])
 
+# Create ForecasterParametersSchema OpenAPI compatible schema
+EXCLUDED_FORECASTING_FIELDS = [
+    # todo: hide these in the config schema instead
+    # "train_period",
+    # "max_training_period",
+    "sensor_to_save",
+]
+forecasting_trigger_schema_openAPI = make_openapi_compatible(ForecastingTriggerSchema)(
+    # partial=True,
+    exclude=EXCLUDED_FORECASTING_FIELDS
+    + ["sensor"],
+)
+
 
 class SensorKwargsSchema(Schema):
     account = AccountIdField(data_key="account_id", required=False)
@@ -1519,7 +1534,10 @@ def get_status(self, id, sensor):
 
     @route("/<id>/forecasts/trigger", methods=["POST"])
     @use_args(
-        ForecasterParametersSchema(),
+        ForecastingTriggerSchema(
+            # partial=True,
+            exclude=EXCLUDED_FORECASTING_FIELDS,
+        ),
         location="combined_sensor_data_description",
         as_kwargs=True,
     )
@@ -1533,9 +1551,8 @@ def trigger_forecast(self, id: int, **params):
           description: |
             Trigger a forecasting job for a sensor.
 
-            This endpoint starts a forecasting job asynchronously and returns a
-            job UUID. The job will run in the background and generate forecast values
-            for the specified period.
+            This endpoint starts a forecasting job asynchronously and returns a job UUID.
+            The job will run in the background and generate forecasts for the specified period.
 
             Once triggered, the job status and results can be retrieved using the
             ``GET /api/v3_0/sensors/<id>/forecasts/<uuid>`` endpoint.
@@ -1554,25 +1571,10 @@ def trigger_forecast(self, id: int, **params):
             required: true
             content:
               application/json:
-                schema:
-                  type: object
-                  properties:
-                    start_date:
-                      type: string
-                      format: date-time
-                      description: Start date of the historical data used for training.
-                    start_predict_date:
-                      type: string
-                      format: date-time
-                      description: Start date of the forecast period.
-                    end_date:
-                      type: string
-                      format: date-time
-                      description: End date of the forecast period.
+                schema: forecasting_trigger_schema_openAPI
                 example:
-                  start_date: "2026-01-01T00:00:00+01:00"
-                  start_predict_date: "2026-01-15T00:00:00+01:00"
-                  end_date: "2026-01-17T00:00:00+01:00"
+                  start: "2026-01-15T00:00:00+01:00"
+                  duration: "P2D"
           responses:
             200:
               description: PROCESSED
@@ -1611,9 +1613,6 @@ def trigger_forecast(self, id: int, **params):
         # Put the sensor to save in the parameters
         parameters["sensor"] = params["sensor_to_save"].id
 
-        # Ensure the forecast is run as a job on a forecasting queue
-        parameters["as_job"] = True
-
         # Set forecaster model
         model = parameters.pop("model", "TrainPredictPipeline")
 
@@ -1622,7 +1621,7 @@ def trigger_forecast(self, id: int, **params):
             forecaster = get_data_generator(
                 source=None,
                 model=model,
-                config={},
+                config=parameters.pop("config", {}),
                 save_config=True,
                 data_generator_type=Forecaster,
             )
@@ -1631,7 +1630,7 @@ def trigger_forecast(self, id: int, **params):
 
         # Queue forecasting job
         try:
-            job_id = forecaster.compute(parameters=parameters)
+            job_id = forecaster.compute(parameters=parameters, as_job=True)
         except Exception as e:
             current_app.logger.exception("Forecast job failed to enqueue.")
             return unprocessable_entity(str(e))
@@ -1731,6 +1730,8 @@ def get_forecast(self, id: int, uuid: str, sensor: Sensor, job_id: str):
                       summary: Started forecasting job
                       value:
                         status: "STARTED"
+            400:
+              description: UNKNOWN_FORECAST
             401:
               description: UNAUTHORIZED
             403:
@@ -1770,7 +1771,9 @@ def get_forecast(self, id: int, uuid: str, sensor: Sensor, job_id: str):
             return dict(**response), s
 
         # Check job status
-        if not job.is_finished:
+        if job.is_finished:
+            message = "A forecasting job has been processed with your job ID"
+        else:
             job_status = job.get_status()
             job_status_name = (
                 job_status.upper() if isinstance(job_status, str) else job_status.name
@@ -1789,8 +1792,8 @@ def get_forecast(self, id: int, uuid: str, sensor: Sensor, job_id: str):
             data_source = get_data_source_for_job(job, type="forecasting")
 
             forecasts = sensor.search_beliefs(
-                event_starts_after=job.meta.get("start_predict_date"),
-                event_ends_before=job.meta.get("end_date"),
+                event_starts_after=job.meta.get("start"),
+                event_ends_before=job.meta.get("end"),
                 source=data_source,
                 most_recent_beliefs_only=True,
                 use_latest_version_per_event=True,
@@ -1800,6 +1803,10 @@ def get_forecast(self, id: int, uuid: str, sensor: Sensor, job_id: str):
             current_app.logger.exception("Failed to get forecast job status.")
             return unprocessable_entity(str(e))
 
+        if forecasts.empty:
+            return unknown_forecast(
+                f"{message}, but the forecast was not found in the database."
+            )
         start = forecasts["event_start"].min()
         last_event_start = forecasts["event_start"].max()
 

flexmeasures/api/v3_0/tests/test_forecasting_api.py

@@ -2,9 +2,7 @@
 import isodate
 import pytest
 from flask import url_for
-from flexmeasures.data.services.scheduling import (
-    get_data_source_for_job,
-)
+
 from rq.job import Job
 from flexmeasures.utils.job_utils import work_on_rq
 from flexmeasures.api.tests.utils import get_auth_token
@@ -35,18 +33,21 @@ def test_trigger_and_fetch_forecasts(
 
     # Trigger job
     payload = {
-        "start_date": "2025-01-01T00:00:00+00:00",
-        "start_predict_date": "2025-01-05T00:00:00+00:00",
-        "end_date": "2025-01-05T02:00:00+00:00",
-        "max_forecast_horizon": "PT1H",
-        "retrain_frequency": "PT1H",
+        "start": "2025-01-05T00:00:00+00:00",
+        "end": "2025-01-05T02:00:00+00:00",
+        "max-forecast-horizon": "PT1H",
+        "forecast-frequency": "PT1H",
+        "config": {
+            "train-start": "2025-01-01T00:00:00+00:00",
+            "retrain-frequency": "PT1H",
+        },
     }
 
     trigger_url = url_for("SensorAPI:trigger_forecast", id=sensor_0.id)
     trigger_res = client.post(
         trigger_url, json=payload, headers={"Authorization": token}
     )
-    assert trigger_res.status_code == 200
+    assert trigger_res.status_code == 200, trigger_res.json
 
     trigger_json = trigger_res.get_json()
     wrap_up_job = app.queues["forecasting"].fetch_job(trigger_json["forecast"])
@@ -77,15 +78,15 @@ def test_trigger_and_fetch_forecasts(
     payload["sensor"] = sensor_1.id
 
     # Run pipeline manually to compute expected forecasts
-    pipeline = TrainPredictPipeline()
+    pipeline = TrainPredictPipeline(config=payload.pop("config", {}))
     pipeline.compute(parameters=payload)
 
     # Fetch forecasts for each job
     for job_id in job_ids:
 
         fetch_url = url_for("SensorAPI:get_forecast", id=sensor_0.id, uuid=job_id)
         res = client.get(fetch_url, headers={"Authorization": token})
-        assert res.status_code == 200
+        assert res.status_code == 200, res.json
 
         data = res.get_json()
 
@@ -103,14 +104,11 @@ def test_trigger_and_fetch_forecasts(
         assert isinstance(api_forecasts, list)
         assert len(api_forecasts) > 0
 
-        # Identify which data source wrote these beliefs
-        data_source = get_data_source_for_job(job, type="forecasting")
-
         # Load only the latest belief per event_start
         forecasts_df = sensor_1.search_beliefs(
-            event_starts_after=job.meta.get("start_predict_date"),
-            event_ends_before=job.meta.get("end_date") + sensor_1.event_resolution,
-            source=data_source,
+            event_starts_after=job.meta.get("start"),
+            event_ends_before=job.meta.get("end"),
+            source_types=["forecaster"],
             most_recent_beliefs_only=True,
             use_latest_version_per_event=True,
         ).reset_index()