fix(forecasting): resolve regressor source collisions deterministically

documentation/changelog.rst

@@ -14,6 +14,7 @@ New features
 -------------
 
 * Forecasting regressors can filter their input beliefs by data source, source type, excluded source type, or source organisation [see `PR #2347 <https://github.com/FlexMeasures/flexmeasures/pull/2347>`_]
+* When multiple selected sources record a belief about the same event at the same belief time, forecasting pipelines now resolve the collision deterministically: the order of an explicit ``sources`` list decides precedence (first listed wins), and otherwise the latest source version wins [see `PR #2347 <https://github.com/FlexMeasures/flexmeasures/pull/2347>`_]
 * Filter organisations by account role in the Accounts API and organisation list UI [see `PR #2353 <https://www.github.com/FlexMeasures/flexmeasures/pull/2353>`_]
 * The flex-context editor now also shows the fields that scheduling the asset would inherit from parent assets — uneditable, with buttons to jump to the editor of the defining parent asset or to override the field on the asset itself [see `PR #2346 <https://www.github.com/FlexMeasures/flexmeasures/pull/2346>`_]
 * Show asset annotations in asset charts: a lightly shaded time band across all subcharts, darkening with a tooltip (showing the annotation text and source) when hovered in a subchart; alerts get a warning hue, and instant annotations render as a vertical rule with a top marker; also adds a ``GET /api/v3_0/assets/<id>/chart_annotations`` endpoint [see `PR #2312 <https://www.github.com/FlexMeasures/flexmeasures/pull/2312>`_]

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

@@ -10,6 +10,7 @@
 from darts.dataprocessing.transformers import MissingValuesFiller
 from timely_beliefs import utils as tb_utils
 
+from flexmeasures.data.models.data_sources import source_priorities
 from flexmeasures.data.models.time_series import Sensor
 from flexmeasures.data.models.forecasting.exceptions import NotEnoughDataException
 from flexmeasures.data.schemas.sensors import SensorReference
@@ -54,6 +55,51 @@ def _regressor_sensor_and_source_filters(
     }
 
 
+def _resolve_source_collisions(
+    df: pd.DataFrame,
+    regressor: Sensor | SensorReference,
+) -> pd.DataFrame:
+    """Keep a single belief per (event_start, belief_time) pair.
+
+    Several selected sources may record a belief about the same event at the same
+    belief time. Which belief is kept is decided as follows:
+
+    - If the regressor reference lists explicit ``sources``, their list order decides:
+      the first listed source wins.
+    - Otherwise, the source ranked highest by
+      :func:`~flexmeasures.data.models.data_sources.source_priorities` wins,
+      i.e. the latest source version, with the highest source ID as the final tie-break.
+
+    Deduplicating here, before the source column is dropped, also prevents duplicated
+    (event_start, belief_time) keys from multiplying rows through the outer join over
+    regressors in ``load_data_all_beliefs``.
+    """
+    if not df.duplicated(subset=["event_start", "belief_time"]).any():
+        return df
+    explicit_sources = (
+        regressor.sources if isinstance(regressor, SensorReference) else None
+    )
+    if explicit_sources:
+        rank = {source.id: position for position, source in enumerate(explicit_sources)}
+        precedence = df["source"].map(lambda s: (rank.get(s.id, len(rank)), s.id))
+        first_wins = True
+    else:
+        precedence = df["source"].map(source_priorities(df["source"].unique()))
+        first_wins = False  # sort descending: a higher priority means a more recent source version
+    return (
+        df.assign(_precedence=precedence)
+        .sort_values(
+            by=["event_start", "belief_time", "_precedence"],
+            ascending=[True, True, first_wins],
+            kind="mergesort",
+        )
+        .drop_duplicates(subset=["event_start", "belief_time"], keep="first")
+        .drop(columns=["_precedence"])
+        .sort_values(by=["event_start", "belief_time"], kind="mergesort")
+        .reset_index(drop=True)
+    )
+
+
 class BasePipeline:
     """
     Base class for Train and Predict pipelines.
@@ -260,6 +306,7 @@ def load_data_all_beliefs(self) -> pd.DataFrame:
                 logging.warning(f"Error during custom resample for {name}: {e}")
 
             df = df.reset_index()
+            df = _resolve_source_collisions(df, regressor_or_sensor)
             df_filtered = df[["event_start", "belief_time", "event_value"]].copy()
             df_filtered.rename(columns={"event_value": name}, inplace=True)
 

flexmeasures/data/schemas/sensors.py

@@ -1024,7 +1024,7 @@ class Meta:
         DataSourceIdField(),
         load_default=None,
         metadata=dict(
-            description="Only use beliefs from these data source IDs.",
+            description="Only use beliefs from these data source IDs. The list order determines precedence: if several listed sources recorded a belief about the same event at the same belief time, the belief from the first listed source is used.",
         ),
     )
     source_account = fields.List(

flexmeasures/data/tests/test_forecasting_pipeline.py

@@ -222,6 +222,95 @@ def test_load_data_all_beliefs_applies_regressor_source_filters(
     assert loaded_data[pipeline.future_regressors[0]].notna().any()
 
 
+def _add_colliding_beliefs(db, sensor, sources_and_values):
+    """Record one belief per source about the same event, all with the same belief time."""
+    db.session.add_all(
+        [
+            TimedBelief(
+                sensor=sensor,
+                event_start=as_server_time(datetime(2025, 1, 2)),
+                event_value=value,
+                belief_horizon=timedelta(hours=6),
+                source=source,
+            )
+            for source, value in sources_and_values
+        ]
+    )
+    db.session.commit()
+
+
+def _load_regressor_values(target_sensor, regressor) -> pd.Series:
+    pipeline = BasePipeline(
+        target_sensor=target_sensor,
+        future_regressors=[regressor],
+        past_regressors=[],
+        n_steps_to_predict=1,
+        max_forecast_horizon=1,
+        forecast_frequency=1,
+        event_starts_after=as_server_time(datetime(2025, 1, 1)),
+        event_ends_before=as_server_time(datetime(2025, 1, 3)),
+    )
+    loaded_data = pipeline.load_data_all_beliefs()
+    assert not loaded_data.duplicated(subset=["event_start", "belief_time"]).any()
+    return loaded_data[pipeline.future_regressors[0]]
+
+
+def test_load_data_all_beliefs_resolves_source_collisions_by_list_order(
+    setup_fresh_test_forecast_data,
+    fresh_db,
+):
+    """The order of an explicit sources list decides which equally-timed belief wins."""
+    target_sensor = setup_fresh_test_forecast_data["solar-sensor"]
+    regressor_sensor = setup_fresh_test_forecast_data["irradiance-sensor"]
+    source_a = DataSource(name="colliding-source-a", type="forecaster")
+    source_b = DataSource(name="colliding-source-b", type="forecaster")
+    value_a = -111.0
+    value_b = -222.0
+    fresh_db.session.add_all([source_a, source_b])
+    _add_colliding_beliefs(
+        fresh_db, regressor_sensor, [(source_a, value_a), (source_b, value_b)]
+    )
+
+    values = _load_regressor_values(
+        target_sensor,
+        SensorReference(sensor=regressor_sensor, sources=[source_a, source_b]),
+    )
+    assert value_a in values.values
+    assert value_b not in values.values
+
+    values = _load_regressor_values(
+        target_sensor,
+        SensorReference(sensor=regressor_sensor, sources=[source_b, source_a]),
+    )
+    assert value_b in values.values
+    assert value_a not in values.values
+
+
+def test_load_data_all_beliefs_resolves_source_collisions_by_source_priority(
+    setup_fresh_test_forecast_data,
+    fresh_db,
+):
+    """Without an explicit sources list, the latest source version wins on collisions."""
+    target_sensor = setup_fresh_test_forecast_data["solar-sensor"]
+    regressor_sensor = setup_fresh_test_forecast_data["irradiance-sensor"]
+    old_source = DataSource(
+        name="versioned-source", type="forecaster", model="test-model", version="1.0.0"
+    )
+    new_source = DataSource(
+        name="versioned-source", type="forecaster", model="test-model", version="2.0.0"
+    )
+    old_value = -111.0
+    new_value = -222.0
+    fresh_db.session.add_all([old_source, new_source])
+    _add_colliding_beliefs(
+        fresh_db, regressor_sensor, [(new_source, new_value), (old_source, old_value)]
+    )
+
+    values = _load_regressor_values(target_sensor, regressor_sensor)
+    assert new_value in values.values
+    assert old_value not in values.values
+
+
 def test_train_predict_job_parameters_payload_preserves_plain_fields(
     setup_fresh_test_forecast_data,
 ):

flexmeasures/ui/static/openapi-specs.json

@@ -4661,7 +4661,7 @@
               "null"
             ],
             "default": null,
-            "description": "Only use beliefs from these data source IDs.",
+            "description": "Only use beliefs from these data source IDs. The list order determines precedence: if several listed sources recorded a belief about the same event at the same belief time, the belief from the first listed source is used.",
             "items": {
               "type": "integer"
             }