From e5671261d3b9003f3b93ea8b4aa48055c5011ed5 Mon Sep 17 00:00:00 2001 From: "F.N. Claessen" Date: Fri, 24 Jul 2026 00:12:01 +0200 Subject: [PATCH] fix(forecasting): resolve regressor source collisions deterministically When several selected sources record a belief about the same event at the same belief time, the pipeline used to keep whichever row it happened to see first, making forecasts dependent on row order. Now the collision is resolved deterministically: the order of an explicit sources list decides precedence (first listed wins), and otherwise the source ranked highest by source_priorities wins (latest version, then highest ID). Deduplicating right after loading also prevents duplicated (event_start, belief_time) keys from multiplying rows through the outer join over regressors. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01WF4PYUDxbBQyPf37PAUKBC Signed-off-by: F.N. Claessen --- documentation/changelog.rst | 1 + .../data/models/forecasting/pipelines/base.py | 47 ++++++++++ flexmeasures/data/schemas/sensors.py | 2 +- .../data/tests/test_forecasting_pipeline.py | 89 +++++++++++++++++++ flexmeasures/ui/static/openapi-specs.json | 2 +- 5 files changed, 139 insertions(+), 2 deletions(-) diff --git a/documentation/changelog.rst b/documentation/changelog.rst index da693d2b0c..276c29e8ff 100644 --- a/documentation/changelog.rst +++ b/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 `_] +* 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 `_] * Filter organisations by account role in the Accounts API and organisation list UI [see `PR #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 `_] * 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//chart_annotations`` endpoint [see `PR #2312 `_] diff --git a/flexmeasures/data/models/forecasting/pipelines/base.py b/flexmeasures/data/models/forecasting/pipelines/base.py index 89062b0e80..6b1bb519dc 100644 --- a/flexmeasures/data/models/forecasting/pipelines/base.py +++ b/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) diff --git a/flexmeasures/data/schemas/sensors.py b/flexmeasures/data/schemas/sensors.py index 78683615a9..9bee134701 100644 --- a/flexmeasures/data/schemas/sensors.py +++ b/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( diff --git a/flexmeasures/data/tests/test_forecasting_pipeline.py b/flexmeasures/data/tests/test_forecasting_pipeline.py index a1afbd4224..98dc6cab5d 100644 --- a/flexmeasures/data/tests/test_forecasting_pipeline.py +++ b/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, ): diff --git a/flexmeasures/ui/static/openapi-specs.json b/flexmeasures/ui/static/openapi-specs.json index 014e4ea368..a42d279b87 100644 --- a/flexmeasures/ui/static/openapi-specs.json +++ b/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" }