feat(scenario aggregation): introduce new feature scenario aggregation

[dusanparipovic]

This PR enables scenario aggregation in the views, and always computes the min, max, expectation and standard deviation indicators over all scenarios.

Example: view_config.yml with scenario aggregation enabled

view:

  id: view_area

  scope:
    - location:
      taxonomy-category: balance

    - calendar: calendar_file # calendar_file.csv

  aggregation:
    time: hour
    scenario: true          # <-- collapses the scenario dimension

  catalog:
      - id: catalog

  metrics:
      - id: catalog.PRODUCTION
      - id: catalog.LOAD

That's it on the config side — same shape as before, just scenario: true instead of false/omitted. load_view_config() reads it into ViewConfig.scenario_aggregation = True (view_config.py:127-129), and ViewBuilder.__init__ constructs ScenarioAggregator(True) from it (views_builder.py:34).

Resulting build behavior

For each metric, ViewBuilder.build() still runs TermsAggregator → TimeAggregator to get a temporal_metric_view with per-scenario rows keyed on metric_id, metric_location, breakdown_properties, scenario_id, view_date. Then ScenarioAggregator.run() (scenario_aggregator.py:34-58) takes over and rewrites that parquet in place:

1. Group by metric_id, metric_location, breakdown_properties, view_date — scenario_id is dropped from the group key, so all scenarios for a given (metric, location, breakdown, timestamp) are collapsed into one group.
2. Aggregate each group's metric_value with all four SCENARIO_AGG_EXPRS:
   • exp = mean(metric_value)
   • std = std(metric_value, ddof=0) (population std — always defined, 0 for a single-scenario view, no null edge case)
   • min = min(metric_value)
   • max = max(metric_value)
3. Unpivot those four aggregate columns into long form, so instead of one wide row per group you get four output rows per group, one per statistic, with the stat name written into scenario_stat and its value into metric_value.
4. Stamp every resulting row with scenario_id = None (a synthesis row no longer corresponds to a single scenario) and scenario_aggregation = True.

So for one (metric_id, metric_location, breakdown_properties, view_date) combination that previously had, say, 10 rows (one per scenario 0–9), the built view now has exactly 4 rows:

metric_id	metric_location	breakdown_properties	view_date	scenario_id	scenario_aggregation	scenario_stat	metric_value
PRODUCTION	area1	{}	2026-01-01T00:00	null	true	exp	mean across scenarios
PRODUCTION	area1	{}	2026-01-01T00:00	null	true	std	population std across scenarios
PRODUCTION	area1	{}	2026-01-01T00:00	null	true	min	min across scenarios
PRODUCTION	area1	{}	2026-01-01T00:00	null	true	max	max across scenarios

By contrast, with scenario: false (or omitted), the row count per group is unchanged from today — one row per original scenario — but every row now also carries scenario_aggregation = false and scenario_stat = null, so the schema (metric_id | metric_location | breakdown_properties | view_date | scenario_id | scenario_aggregation | scenario_stat | metric_value) stays identical across both modes; only the row semantics differ.

[aoustry]

Is there a specific reason why scenario_aggregator modify a MetricView, where a time_aggregator returns one ? This is does not look very homogeneous.

[dusanparipovic]

As you can see on line 43, we append temporal_metric_view to the metric_views list.

We don't have any strict requirements for the MetricView object. In practice, it only stores the path to the temporary view.

If you look at the run method of the ScenarioAggregator, you'll notice that it doesn't return anything. That's because I intentionally rely on Python's reference semantics: the temporal_metric_view object is passed by reference, and the aggregator updates its content (specifically, the path stored in the MetricView object).

From my perspective, there's no need to create a new object with identical data. We can simply update the existing object instead.

[tbittar]

Add a comment in the beginning of the function to make the behavior difference explicit :

"""Rewrites temporal_metric_view's parquet file in place (same path, new content)
and returns the same MetricView instance; it does not produce a separate artifact.""

[aoustry]

Please remove Legacy, then :)

[tbittar]

Suggestion for ScenarioAggregator.run(): wrap the scan/sink block in a try/except that removes tmp_path before re-raising, e.g.

try:
    ...  # scan_parquet / sink_parquet
except Exception:
    try:
        os.remove(tmp_path)
    except FileNotFoundError:
        pass
    raise

Why: tempfile.mkstemp creates the temp parquet file immediately, and os.replace only happens at the very end. If sink_parquet raises partway (disk full, schema mismatch, etc.), that temp file is orphaned in the system temp dir — unlike TermsAggregator/TimeAggregator, this aggregator doesn't own a managed temp directory that gets cleaned up via __del__/atexit, so nothing else will ever remove it. The try/except closes that leak without touching the happy-path behavior.

[guilpier-code]

Please add something to this PR's description :

• associated ticket ?
• associated specifications ?
• A bit more details about this PR than the title gives

Otherwise, there are things that the reader may have to guess.