Skip to content

Count an event once in a KPI, when several sources report it #2472

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Flix6x wants to merge 4 commits into
base: main
Choose a base branch
from
Open

Count an event once in a KPI, when several sources report it #2472

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
b4d0b30
Count an event once in a KPI, when several sources report it
Flix6x Sep 3, 2026
d9423d1
docs/changelog: record the KPI fix
Flix6x Sep 3, 2026
097070f
tests: cover the version priority a KPI applies, and say where ties stop
Flix6x Sep 3, 2026
da9ebb9
docs: reflow the tie-order docstring to break after punctuation
Flix6x Sep 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions documentation/changelog.rst
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@ Infrastructure / Support
Bugfixes
-----------

* A KPI on the asset page counted an event once per data source that reported it, so a total could come out higher than any source reported; it now reduces one value per event, from the latest source version and the most recent belief within it [see `PR #2472 <https://www.github.com/FlexMeasures/flexmeasures/pull/2472>`_]
* Sensor data ingestion now preserves ``null`` gaps when converting posted values to the sensor's unit, instead of failing the request [see `PR #2461 <https://www.github.com/FlexMeasures/flexmeasures/pull/2461>`_]
* KPIs on the asset page counted one day more than the selected time range [see `PR #2434 <https://www.github.com/FlexMeasures/flexmeasures/pull/2434>`_]
* KPIs on the asset page now total the values the chart beside them draws, counting each event under the day it starts in: a sensor reported by several sources counted only one of them, and a revised value was counted on top of the value it revised [see `PR #2434 <https://www.github.com/FlexMeasures/flexmeasures/pull/2434>`_]
Expand Down
4 changes: 4 additions & 0 deletions documentation/views/asset-data.rst
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -131,6 +131,10 @@ Currently, this supports only a daily resolution (which fits the date picker on
So you will need a sensor with daily resolution (probably generated with FlexMeasures' reporting tooling).
From this data, you can display summed totals, means, max or min values (the image above shows two KPIs with totals).

The function is applied to one value per event.
Where several data sources reported the same event, the value is the one from the latest source version, and from the most recent belief within that, rather than each source's value in turn.
The chart beside the KPI still draws every source, so it can show more points than the KPI counted.

We aim to support a graphical tool to edit these KPIs in the future.
For now, you can set them by editing the asset's `kpi_sensors_to_show` field in the properties page, which will validate that the format is correct and tell you what to change. Read more about the format below.

Expand Down
10 changes: 7 additions & 3 deletions flexmeasures/api/v3_0/assets.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2098,13 +2098,17 @@ def get_kpis(self, id: int, asset: GenericAsset, start, end):
kpis = []
for kpi in asset_kpis:
sensor = Sensor.query.get(kpi["sensor"])
# The beliefs the chart draws: one value per event, the most recent one.
# Aggregating belief rows instead would count a revision on top of what it revised,
# and would count each source separately when several report the same sensor.
# One value per event, which is what a KPI reduces.
# Aggregating belief rows instead would count a revision on top of the belief it revised,
# and would count each source separately when several report the same event,
# so that a total came out higher than anything anyone reported.
# Where several do report an event, the value is the one from the latest source version,
# and from the most recent belief within that.
beliefs = sensor.search_beliefs(
event_starts_after=start,
event_ends_before=end,
most_recent_beliefs_only=True,
one_deterministic_belief_per_event=True,
)
Comment on lines 2107 to 2112

@Flix6x Flix6x Sep 3, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good catch on the ambiguity, but the fix you suggest would break a documented feature — so I have
left the tie alone and documented why.

I did implement it first: a last tie-break by highest source id, matching source_priorities, which
ranks by version and then by id. Two existing tests then failed, and the second one is the answer:

test_source_transition (flexmeasures/data/models/reporting/tests/test_aggregator.py) states

In case of encountering more than one source per event, the first source defined in the sources
array is prioritized.

An AggregatorReporter given sources=[ds1, ds2] expects ds1 to win the events both report. That
works because the remaining tie keeps the order the rows arrive in. Breaking the tie by source id
made ds2 win, since it has the higher id, and the reporter silently stopped honouring the caller's
order. test_select_latest_version_and_belief_per_event_equivalence failed too, since its reference
implementation keeps the incumbent on an exact tie.

So the order is not arbitrary noise to be tidied away — it is how precedence is expressed. What was
missing is that this is nowhere written down, so _select_latest_version_and_belief_per_event now
says it:

Beliefs that tie on both keep the order they came in, which is what lets a caller express its own
precedence by the order in which it passes its sources.

For the KPI endpoint, which passes no sources of its own, an exact tie — same version, same belief
time, two sources — does leave the value up to the order the rows come back in. That is a real
limit, and a narrow one: it needs two sources to claim one event at the very same moment under the
same version. Worth its own issue if we want KPIs to be deterministic even then; it is not something
this PR can settle without taking the reporter's precedence away.

@Flix6x Flix6x Sep 4, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

But if no user preference is given, we could still make it deterministic by selecting the latest ID, right? It seems to me that this might solve the issue for KPIs without breaking the feature of users selecting a specific source list? Or am I missing some caveats?

@Flix6x Flix6x Sep 4, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You are not missing a caveat — I was. The caveat I gave earlier in this thread is wrong, and so is the premise it was answering.

There is no user-order feature to break. As shown on the other thread, reversing the sources list changes nothing; ties are settled by the source's name, through sort_index() and BeliefSource.__lt__ comparing string representations. test_source_transition passes because source1 sorts before source2. So breaking ties by id would not take a feature away — it would replace one arbitrary rule with another, and flip that test.

But the thing we were fixing is not broken either. Because that ordering comes from a deterministic sort rather than from however the rows arrive, a KPI over tied sources already answers the same way every time. Copilot's "effectively non-deterministic, depending on row ordering from the DB/pandas" is not what happens. My probe returns the same winner across repeats, and it is the alphabetically first source name.

So the choice is narrower than it looked: not "deterministic or not", but which arbitrary source should win — alphabetically first, or most recently added. Latest id is more defensible than alphabetical, since it at least tracks something real. Two things I would weigh:

  • Doing it only when no source filter is given means the two paths settle ties differently: alphabetical when a caller passes sources, newest when it does not. Doing it in both places is more coherent, and costs updating test_source_transition, whose expectation encodes the alphabetical outcome.
  • Either way it is a change of policy on ambiguous data, not a bug fix, so it does not need to ride along with this PR.

My suggestion: leave this PR as it is, since it does not touch tie behaviour, and open an issue covering all of it — the docstring that promises a precedence the code does not implement, the alphabetical rule nobody chose, and your question about whether a caller's order should outrank a more recent belief. Happy to write that up.

@Flix6x Flix6x Sep 4, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You are right, and it is not only my proposal that has it wrong — the code does this today. Your example, near enough verbatim: a scheduler at v1 and a forecaster at v9 on one sensor, one event, the scheduler holding the belief that is 12 hours fresher.

no list                 -> forecaster v9 value=22.0
[scheduler, forecaster] -> forecaster v9 value=22.0
   (scheduler v1 = 11.0, believed at the event; forecaster v9 = 22.0, believed 12h earlier)

The forecaster wins on the number alone, against a fresher belief and against a caller who listed the scheduler first.

The cause is that the two mechanisms disagree about what a version means. keep_latest_version compares versions only within a (name, type, model) family, which is the only place the number means anything. The selection that runs after it ranks every source in the frame together:

versions = [Version(source.version if source.version else "0.0.0") for source in unique_sources]
version_ranks = {version: rank for rank, version in enumerate(sorted(set(versions)))}

Two teams' independent numbering treated as one series. A plugin scheduler that starts at v10 outranks a built-in one at v3 permanently, on any sensor they share.

So the recommendation in #2476 now reads in two steps rather than one chain:

  • Within a family: latest version, then freshest belief, then highest id.
  • Between families: the caller's sources order if given, else freshest belief, else highest id.

Version never crosses the family line. Your example then resolves the way you would expect — the scheduler, because it was listed first, or because it is fresher when nothing was listed.

Issue updated.

# Count each event once, under the window it starts in.
# The search also returns events that merely overlap the window, which the chart draws,
Expand Down
149 changes: 149 additions & 0 deletions flexmeasures/api/v3_0/tests/test_assets_api.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1910,6 +1910,155 @@ def test_kpi_window_honours_the_offset_it_is_given(
assert total != shifted, "the assertion above only means something if these differ"


@pytest.mark.parametrize("requesting_user", ["test_admin_user@seita.nl"], indirect=True)
def test_kpi_counts_an_event_once_when_two_sources_report_it(
db, client, setup_api_test_data, setup_sources, requesting_user
):
"""Two sources reporting one event are two claims about it, not two contributions to it.

Summing them produced a number no source ever reported, and that no point on the chart showed.
The KPI now reduces one value per event, and these two sources are of the same version,
so the one that believed the event more recently is the one it counts.
"""
asset_type = (
db.session.query(GenericAssetType).filter_by(name="battery").one_or_none()
)
asset = GenericAsset(
name="kpi with two sources on one event",
generic_asset_type=asset_type,
account_id=requesting_user.account_id,
)
db.session.add(asset)
db.session.flush()
sensor = Sensor(
name="kpi with two sources sensor",
generic_asset=asset,
event_resolution=timedelta(days=1),
unit="EUR",
)
db.session.add(sensor)
db.session.flush()

sources = list(setup_sources.values())
reported, corrected = sources[0], sources[-1]
assert reported.id != corrected.id, "this test needs two distinct sources"

Comment on lines +1942 to +1945

@Flix6x Flix6x Sep 3, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Right — setup_sources creates sources without versions, so that test only ever showed the belief
time. Two changes in 097070f.

The docstring now says what the test does: these two sources are of the same version, so the one
that believed the event more recently is the one the KPI counts.

And a new test covers the version half, with two versions of one reporter, the newer one speaking
first:

# The newer version spoke first, and the older version spoke later.
...
assert total == pytest.approx(42.0), "the newer version's value, despite the older belief time"

It reports 141.0 without the fix — 42 plus 99 — so it covers both the summing and the ordering.

window_start = datetime(2030, 3, 15, tzinfo=utc)
db.session.bulk_insert_mappings(
TimedBelief,
[
# One event, claimed by two sources, the second more recently than the first.
dict(
event_start=window_start,
belief_horizon=timedelta(days=2),
event_value=100.0,
sensor_id=sensor.id,
source_id=reported.id,
cumulative_probability=0.5,
),
dict(
event_start=window_start,
belief_horizon=timedelta(days=1),
event_value=80.0,
sensor_id=sensor.id,
source_id=corrected.id,
cumulative_probability=0.5,
),
],
)
asset.sensors_to_show_as_kpis = [
{"title": "Daily costs", "sensor": sensor.id, "function": "sum"}
]
db.session.flush()

total = _kpi_total(
client,
asset,
window_start.isoformat(),
(window_start + timedelta(days=1)).isoformat(),
)
assert total == pytest.approx(
80.0
), "the more recent belief about the event, rather than 180.0, which neither source reported"


@pytest.mark.parametrize("requesting_user", ["test_admin_user@seita.nl"], indirect=True)
def test_kpi_prefers_the_latest_source_version_over_the_most_recent_belief(
db, client, setup_api_test_data, requesting_user
):
"""A newer version of a source wins the event, even when an older version believed it more recently.

Version comes first because it says which code produced the value,
where the belief time only says when it was said.
"""
from flexmeasures.data.models.data_sources import DataSource

asset_type = (
db.session.query(GenericAssetType).filter_by(name="battery").one_or_none()
)
asset = GenericAsset(
name="kpi with two source versions",
generic_asset_type=asset_type,
account_id=requesting_user.account_id,
)
db.session.add(asset)
db.session.flush()
sensor = Sensor(
name="kpi with two source versions sensor",
generic_asset=asset,
event_resolution=timedelta(days=1),
unit="EUR",
)
db.session.add(sensor)
# Two versions of one reporter, which is what a release upgrade leaves behind.
older_version = DataSource(
name="Reporter", type="reporter", model="Rep", version="1"
)
newer_version = DataSource(
name="Reporter", type="reporter", model="Rep", version="2"
)
db.session.add_all([older_version, newer_version])
db.session.flush()

window_start = datetime(2030, 4, 15, tzinfo=utc)
db.session.bulk_insert_mappings(
TimedBelief,
[
# The newer version spoke first, and the older version spoke later.
dict(
event_start=window_start,
belief_horizon=timedelta(days=2),
event_value=42.0,
sensor_id=sensor.id,
source_id=newer_version.id,
cumulative_probability=0.5,
),
dict(
event_start=window_start,
belief_horizon=timedelta(days=1),
event_value=99.0,
sensor_id=sensor.id,
source_id=older_version.id,
cumulative_probability=0.5,
),
],
)
asset.sensors_to_show_as_kpis = [
{"title": "Daily costs", "sensor": sensor.id, "function": "sum"}
]
db.session.flush()

total = _kpi_total(
client,
asset,
window_start.isoformat(),
(window_start + timedelta(days=1)).isoformat(),
)
assert total == pytest.approx(
42.0
), "the newer version's value, despite the older belief time"


@pytest.mark.parametrize("requesting_user", ["test_admin_user@seita.nl"], indirect=True)
def test_kpi_reports_what_the_chart_draws(
db, client, setup_api_test_data, setup_sources, requesting_user
Expand Down
4 changes: 4 additions & 0 deletions flexmeasures/data/models/time_series.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -902,6 +902,10 @@ def _select_latest_version_and_belief_per_event(
"""Keep, per event, the single belief with the latest source version,
breaking version ties by most recent belief time.

Beliefs that tie on both keep the order they came in,
which is what lets a caller express its own precedence by the order in which it passes its sources.
Comment on lines +905 to +906

@Flix6x Flix6x Sep 4, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This suggests that the user order is inferior to the belief time order. Was that the desired behaviour when the user order feature was introduced? Just a question at this point, no need to go off and make changes yet.

@Flix6x Flix6x Sep 4, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It is worse than inferior: the caller's order is not consulted at all. I probed it rather than reasoned about it.

The list order changes nothing. Running test_source_transition's scenario with the sources reversed:

DIAG sources=[ds1, ds2]: first 13 values are {1.0}
DIAG sources=[ds2, ds1]: first 13 values are {1.0}

Same answer both ways. And search_beliefs(source=[ds2, ds1]) returns the contested events with source ids in the order [1, 2], whatever order they were asked for.

What actually decides is the source's name. BeliefsDataFrame is sort_index()ed, and its index has a source level, so tied rows end up ordered by BeliefSource.__lt__, which is:

def __lt__(self, other):
    """Set a rule for ordering."""
    return self.__str__() < other.__str__()

A probe with the id order and the name order deliberately disagreeing settles it:

DIAG ids: zzz=1 aaa=2
DIAG winner: name='aaa reporter' id=2 value=22.0
DIAG -> decided by NAME

So in test_source_transition, source1 wins because it sorts before source2, not because it was listed first.

Which means the question was never faced. Every belief in that fixture has belief_horizon=timedelta(hours=24), so the contested events are exact ties on version and belief time. Belief-time precedence has never been exercised against a caller's order, because the order has never done anything. The docstring's "the first source defined in the sources array is prioritized" describes an intention that is not in the code.

Worth its own issue, I think — both the misleading docstring and the question you are asking, which is a real decision nobody has taken.

@Flix6x Flix6x Sep 4, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Researched. Short version: the array-order precedence was specified in review and never implemented, and you were the one who asked for it to be pinned down.

The PR is #819, Make source optional in the input field of the AggregatorReporter (merged 2023-08-21). Its description says only that source becomes optional, "because in many cases sensors only have one source". Everything about multiple sources grew during review, in the commits "allow passing multiple sources" and "add test + fix case of multiple sources".

The use case is a data source transition on one sensor. Victor's words:

I would avoid having mixed sources ... For instance, this could be an issue when computing the aggregated power of a group of devices including a scheduled device. We could get actual power commands and scheduled power commands mixed up.

and yours:

I have in mind the case where a new data source takes over from another, both saving data to the new sensor during some transitioning time. The default source filter (i.e. not specifying one) should still result in a complete time series.

That produced two things: sources=[...] so a caller can span the transition, and the ValueError when a sensor has several sources and none were named, so measurements and schedules cannot be mixed silently.

The intention came from your own question, on the very line we are discussing:

The first, as in the first source defined in the sources array, or as in the first source to have reported something about that event (oldest belief_time)? Please clarify in the docstring.

The docstring we have today is the answer to it — "the first source defined in the sources array is prioritized" — so array order is what was meant. Nothing was written to make it so. And you had already noticed the gap earlier in the same review:

But it is no longer true that we'd take the first source as a default. And also, this test doesn't actually check that.

That is still true. Your uncertainty back then ("I'm not 100% sure whether that would lead to taking the oldest or newest belief") also never got resolved, because the fixture gives every belief belief_horizon=timedelta(hours=24), so the two never compete.

The tests are all in test_source_transition: both sources (24 values), only ds1 (13), only ds2 (12), and no source at all (raises). The filtering half is covered properly. Only assert (result[:13] == 1).all() touches precedence, and it passes because source1 sorts before source2, not because it is first in the array.

What that means for the use case. It works today, but on an accident: at the overlapping event the winner is the alphabetically first source name. Name a scheduler's source something that sorts before the measurement source and the transition silently resolves the wrong way — the exact mix-up the ValueError was added to prevent, reappearing when sources are named explicitly.

Shall I write this up as an issue? It has three parts worth separating: the docstring promising precedence that is not implemented, the alphabetical rule nobody chose, and your original question of whether a caller's order should outrank a more recent belief.

@Flix6x Flix6x Sep 4, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

3) You are right, and my sentence was wrong. Same event and same belief horizon does mean the same belief time, so the two sources are very much in competition — that is precisely why something has to break the tie. What I meant to say is that the belief-time criterion never gets to decide between them: they tie on it, so the comparison falls through to whatever orders the sources themselves. Your 2023 question of oldest-versus-newest is what the fixture never puts to the test, not the competition itself.

2) Right on both counts, and it explains the drift. Version preference came later: keep_latest_version arrives in #1045 (2024-05-21), reworked in #1306 (2025-01-28), which is also where use_latest_version_per_event comes from. Account filtering is later still: source_account_ids in #2083 (2026-04-19), #2065 (2026-05-14) and #2209 (2026-06-02). Source type filtering is the exception — source_types predates #819, going back to 2021.

That matters here. In August 2023 there was no version step at all, and FM did not do this selection itself: one_deterministic_belief_per_event was passed straight through to timely-beliefs. FM took the selection into its own code with the version work, and _select_latest_version_and_belief_per_event, which sorts version descending then belief time descending, belongs to that later era. So the docstring was written against a pipeline that no longer exists, and a version step was later inserted above it.

1) No — the record says the opposite, though the code now does what you describe. Three things were said, and none of them is "freshest wins":

  • Victor's docstring, written as the answer to your clarification request, says the array order is prioritized.
  • Your own expectation was "we take the most recent belief per event per source, ... we then take the first source per event" — recency within a source, then a choice between sources.
  • And your guess about the cross-source case was the oldest belief, not the newest.

What runs today is none of those. Two sources, no versions, one event, the alphabetically first source holding the older belief:

DIAG sources=no list      -> winner='zzz source' value=22.0
DIAG sources=[aaa, zzz]   -> winner='zzz source' value=22.0
DIAG sources=[zzz, aaa]   -> winner='zzz source' value=22.0
DIAG (aaa=11.0 spoke 5h ahead; zzz=22.0 spoke 1h ahead, i.e. fresher)

So the precedence today is version, then belief time, then source name, and the listing is never consulted. Freshness even beats the name rule, which is why this only shows up as alphabetical when the belief times tie — as they do in that fixture.

One thing I have not verified: whether the 2023 timely-beliefs implementation took the oldest belief, as you suspected at the time. If it did, the cross-source rule flipped from oldest to newest when FM took the selection over, silently. That is checkable by running the old library, and I would want it in the issue if we write one, since it decides whether "freshest wins" was ever a decision or just a side effect.

@Flix6x Flix6x Sep 4, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Dug into it, and the answer is the opposite of "silent": the flip was deliberate and written down, just not where the docstring is.

2023 took the oldest belief, exactly as you suspected. The implementation was FlexMeasures' own, not timely-beliefs' — one_deterministic_belief_per_event does not exist in tb 1.21.0, which is what #819 was pinned to. FlexMeasures did:

bdf = (
    bdf.for_each_belief(get_median_belief)
    .groupby(level=["event_start"], group_keys=False)
    .apply(lambda x: x.head(1))
)

head(1) over an index sorted ascending, so the earliest belief time wins. Running that expression verbatim beside today's rule, on one event a scheduler believed 12 hours ahead and a meter believed at the event:

today's rule  -> 'meter'     value=22.0
2023 rule     -> 'scheduler' value=11.0

It changed in #1306 (2025-01-28), whose description says so as its first bullet:

Improve how Sensor.search_beliefs with one_deterministic_belief_per_event=True chooses between multiple sources: favour latest version first, freshest belief second

So freshest-wins was a decision, taken with a reason. What went unnoticed is that the docstring one file away still promised the array order — #1306 even fixed a typo in that sentence while leaving the claim standing.

One more thing the dig turned up, which I had not expected: the rules differ depending on whether the sources share a name. Two sources of one family are collapsed by keep_latest_version on version and then highest id, ignoring belief time entirely, while two sources of different families go on to version, then freshest belief, then name:

ids: a=1 b=2 (a holds the fresher belief)  ->  winner: id=2

Written up as #2476, with the history, the use cases you asked about — KPIs, reporters at a source transition, a bumped scheduler version, two forecasting models side by side, and the config-fragmented scheduler sources from #2464 — and a recommendation.

See `test_source_transition`, where the first source in the list wins the events both sources report.

Assumes deterministic beliefs (probabilistic depth 1) and a belief_time index level.
"""
source_codes, unique_sources = pd.factorize(bdf.index.get_level_values("source"))
Expand Down
Loading