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" }