Add custom statistics (per-channel & cross-channel, multi-output) to StatsAggregator

.gitignore

@@ -84,4 +84,5 @@ yarn-error.log*
 
 # Claude Code
 .claude/
+.isaac/
 CLAUDE.md

docs/impulse/docs/data_model/index.md

@@ -60,7 +60,7 @@ The Gold layer uses a **star schema** with fact and dimension tables. All table
 | `event_instance_fact`  | One row per event instance per container | Materialized time windows where an event condition holds. |
 | `histogram_fact`       | One row per bin per container            | 1D histogram bin values, duration-weighted.               |
 | `histogram2d_fact`     | One row per (x, y) bin per container     | 2D histogram bin values, duration-weighted.               |
-| `stats_aggregator_fact` | One row per signal per event instance    | Descriptive statistics (min, max, mean, median).          |
+| `stats_aggregator_fact` | One row per statistic label per signal per event instance | Descriptive statistics (built-in min/max/mean/median and any custom statistics). |
 | `calculated_channel_fact` | One row per sample interval per container | Materialized derived signal (a *channel*, not a summary), in the silver `channels` shape. |
 
 ### Dimension tables

docs/impulse/docs/references/api/impulse_query_engine/analyze/query/aggregations/custom_statistic.md

@@ -0,0 +1,119 @@
+---
+sidebar_label: custom_statistic
+title: impulse_query_engine.analyze.query.aggregations.custom_statistic
+---
+
+Descriptors for custom statistics (per-channel and cross-channel).
+
+
+## PerChannelStatistic
+
+```python
+class PerChannelStatistic()
+```
+
+Descriptor for a single per-channel custom statistic.
+
+**Arguments**:
+
+- `func` (`Callable`): Function with signature ``func(series: SampleSeries, t_start: float,
+t_end: float, **params) -> Sequence[float]``. Called once per input
+channel and event interval with the channel's series clipped to the
+interval. It must return a sequence of scalars whose length equals
+``aggregation_labels`` (a single label still requires a one-element
+sequence, e.g. ``[value]``). The series may be empty; return
+``float("nan")`` entries for undefined results. The function is
+cloudpickled to Spark executors, so a module-level importable function is
+recommended; never capture Spark objects.
+- `aggregation_labels` (`list of str`): Output labels this statistic produces. The values returned by ``func``
+are mapped positionally to these labels, which become the keys of the
+statistic's result maps. Labels must be non-empty, unique strings;
+changing them changes the aggregation's definition hash.
+- `params` (`dict`): Keyword arguments passed to ``func`` on every invocation
+(``func(series, t_start, t_end, **params)``). Keys must be valid Python
+identifiers matching parameter names of ``func``. Changing params
+changes the aggregation's definition hash.
+
+## CrossChannelStatistic
+
+```python
+class CrossChannelStatistic()
+```
+
+Descriptor for a single cross-channel custom statistic.
+
+**Arguments**:
+
+- `func` (`Callable`): Function with signature ``func(series: list[SampleSeries], t_start: float,
+t_end: float, **params) -> Sequence[float]``. Called once per event
+interval with the series listed in ``inputs`` (clipped to the interval,
+in declared order). It must return a sequence of scalars whose length
+equals ``aggregation_labels`` (a single label still requires a
+one-element sequence, e.g. ``[value]``). Any series may be empty; return
+``float("nan")`` entries for undefined results. The function is
+cloudpickled to Spark executors, so a module-level importable function is
+recommended; never capture Spark objects.
+- `aggregation_labels` (`list of str`): Output labels this statistic produces. The values returned by ``func``
+are mapped positionally to these labels, which become the keys of the
+statistic's result maps. Labels must be non-empty, unique strings;
+changing them changes the aggregation's definition hash.
+- `inputs` (`list of str`): Names of the input channels the function requires, resolved against the
+aggregator's ``input_names``. ``None`` (default) passes all input
+channels in input order.
+- `channel_name` (`str`): A channel name applied to all of the statistic's output rows. Consumed by
+downstream consumers (e.g. the reporting layer) only; ignored by the
+query engine. ``None`` (default) leaves it to the consumer, which
+typically falls back to each output's ``aggregation_label``.
+- `params` (`dict`): Keyword arguments passed to ``func`` on every invocation
+(``func(series, t_start, t_end, **params)``). Keys must be valid Python
+identifiers matching parameter names of ``func``. Changing params
+changes the aggregation's definition hash.
+
+#### normalize\_per\_channel\_statistics
+
+```python
+def normalize_per_channel_statistics(
+    per_channel_custom_statistics: list[PerChannelStatistic] | None
+) -> list[PerChannelStatistic]
+```
+
+Validate a per-channel statistics list.
+
+**Arguments**:
+
+- `per_channel_custom_statistics` (`list or None`): List of ``PerChannelStatistic`` descriptors.
+
+**Raises**:
+
+- `TypeError`: If the value is not a list, an item is not a ``PerChannelStatistic``
+with a callable ``func``, or params/labels are invalid.
+- `ValueError`: If a descriptor's ``aggregation_labels`` are not unique.
+
+**Returns**:
+
+`list of PerChannelStatistic`: The validated list; empty when the input is ``None``.
+
+#### normalize\_cross\_channel\_statistics
+
+```python
+def normalize_cross_channel_statistics(
+    cross_channel_custom_statistics: list[CrossChannelStatistic] | None
+) -> list[CrossChannelStatistic]
+```
+
+Validate a cross-channel statistics list.
+
+**Arguments**:
+
+- `cross_channel_custom_statistics` (`list or None`): List of ``CrossChannelStatistic`` descriptors.
+
+**Raises**:
+
+- `TypeError`: If the value is not a list, an item is not a ``CrossChannelStatistic``
+with a callable ``func``, or params/labels are invalid.
+- `ValueError`: If a descriptor's ``aggregation_labels`` are not unique.
+
+**Returns**:
+
+`list of CrossChannelStatistic`: The validated list; empty when the input is ``None``.
+

docs/impulse/docs/references/api/impulse_reporting/aggregations/stats_aggregator.md

@@ -28,7 +28,11 @@ def __init__(name: str,
              event: Event | None = None,
              desc: str = None,
              agg_type: str = "stats_aggregator",
-             values_unit: str = None)
+             values_unit: str = None,
+             cross_channel_custom_statistics: list[CrossChannelStatistic]
+             | None = None,
+             per_channel_custom_statistics: list[PerChannelStatistic]
+             | None = None)
 ```
 
 Initialize a StatsAggregator object.
@@ -44,6 +48,16 @@ are computed over the entire time series.
 - `desc` (`str`): Description of the aggregation.
 - `agg_type` (`str`): Type of aggregation, defaults to "stats_aggregator".
 - `values_unit` (`str`): Unit of the statistic values.
+- `cross_channel_custom_statistics` (`list of CrossChannelStatistic`): Custom statistics computed once per event interval across each
+descriptor's declared input channels (referencing ``channel_names``;
+all channels when none are declared). Fact rows carry the
+descriptor's ``channel_name`` (applied to all its output labels),
+defaulting to each output's ``aggregation_label``. See
+``impulse_query_engine`` ``StatsAggregator`` for the callable contract.
+- `per_channel_custom_statistics` (`list of PerChannelStatistic`): Custom statistics computed once per input channel and event interval,
+exactly like a built-in statistic. Fact rows carry the real channel
+names. Descriptor ``params`` are passed to the callable as keyword
+arguments.
 
 #### get\_id
 
@@ -176,8 +190,14 @@ Only includes computation-affecting attributes:
 - input_expressions
 - statistics to be calculated
 - event expression if there is any
-
-Excludes: name, desc, signal_name, units, page_number, report_id
+- custom statistics (name, kind, declared input indices, and function
+  bytecode, so implementation or input-wiring changes invalidate cached
+  results; only appended when custom statistics are configured so
+  aggregators without them keep their previous hash)
+
+Excludes: name, desc, signal_name, units, page_number, report_id, and the
+cross-channel descriptors' channel_name (presentation metadata, like
+channel_names).
 
 **Returns**:
 

docs/impulse/docs/references/api/impulse_reporting/channels/calculated_channel.md

@@ -20,8 +20,8 @@ the narrow result to a gold fact table, and update it incrementally.
 Structurally parallels :class:`BasicEvent` (holds an aliased expression,
 name-derived id, SHA-256 definition hash) but — like ``ContainerEvent`` — it
 drives its own solve via ``QueryBuilder.solve_calculated_channels`` rather than
-riding the centralized wide ``solved_df``.  Accordingly :meth:`get_expression`
-returns ``None`` so it is excluded from the batch solve.
+riding the centralized wide ``solved_df``.  It is dispatched separately from
+the batch solve (never passed to ``collect_solvable_expressions``).
 
 **Arguments**:
 
@@ -78,13 +78,10 @@ Return the deterministic entity id (also the fact/dimension ``channel_id``).
 #### get\_expression
 
 ```python
-def get_expression() -> TimeSeriesExpression | None
+def get_expression() -> TimeSeriesExpression
 ```
 
-Return ``None`` — calculated channels drive their own narrow solve.
-
-Returning ``None`` keeps this channel out of the centralized wide batch
-solve (``collect_solvable_expressions``), mirroring ``ContainerEvent``.
+Return the wrapped query-engine ``CalculatedChannel`` expression.
 
 
 #### get\_expression\_str

docs/impulse/docs/references/api/sidebar.json

@@ -17,6 +17,7 @@
               "items": [
                 {
                   "items": [
+                    "references/api/impulse_query_engine/analyze/query/aggregations/custom_statistic",
                     "references/api/impulse_query_engine/analyze/query/aggregations/statistic_type"
                   ],
                   "label": "impulse_query_engine.analyze.query.aggregations",

docs/impulse/docs/references/report/aggregation.md

@@ -254,6 +254,8 @@ page.add_aggregation(stats)
 | `desc`               | `str`                        | No       | Description.                                                                          |
 | `agg_type`           | `str`                        | No       | Aggregation type identifier. Defaults to `"stats_aggregator"`.                        |
 | `values_unit`        | `str`                        | No       | Unit of the statistic values.                                                         |
+| `per_channel_custom_statistics`  | `list[PerChannelStatistic]`  | No | Custom statistics computed once per channel per interval (see below).       |
+| `cross_channel_custom_statistics`| `list[CrossChannelStatistic]`| No | Custom statistics computed once per interval across a set of channels (see below). |
 
 ### Supported statistics
 
@@ -268,6 +270,77 @@ page.add_aggregation(stats)
 
 A `ValueError` is raised if unsupported statistics are provided.
 
+### Custom statistics
+
+Beyond the built-in labels, you can inject your own statistic functions. Each is declared as a descriptor
+(`PerChannelStatistic` / `CrossChannelStatistic`, see the
+[custom_statistic API reference](../api/impulse_query_engine/analyze/query/aggregations/custom_statistic.md))
+and returns a **sequence of scalars** mapped positionally to its `aggregation_labels`; a single-output
+statistic returns a one-element sequence. Custom outputs land in the fact table as ordinary
+`aggregation_label` / `statistic_value` rows — no schema change — and their labels are added to the
+dimension row's `statistics` array.
+
+There are two kinds:
+
+- **Per-channel** (`PerChannelStatistic`) — computed once per input channel per interval, exactly like a
+  built-in. Fact rows carry the **real channel name**. Signature:
+  `func(series: SampleSeries, t_start: float, t_end: float, **params) -> Sequence[float]`.
+- **Cross-channel** (`CrossChannelStatistic`) — computed once per interval across a set of channels. Fact
+  rows carry the descriptor's `channel_name` (or the output label when it is `None`). Signature:
+  `func(series: list[SampleSeries], t_start: float, t_end: float, **params) -> Sequence[float]`.
+
+```python
+import numpy as np
+from impulse_query_engine.analyze.query.aggregations.custom_statistic import (
+    PerChannelStatistic,
+    CrossChannelStatistic,
+)
+
+def percentiles(series, t_start, t_end):          # per-channel, multi-output
+    return [float(np.nanpercentile(series.values, 50)),
+            float(np.nanpercentile(series.values, 90))]
+
+def spread(series, t_start, t_end):               # cross-channel, single output
+    values = [v for s in series for v in s.values]
+    return [float(max(values) - min(values))] if values else [float("nan")]
+
+stats = StatsAggregator(
+    name="custom_stats",
+    input_expressions=[eng_rpm, veh_spd],
+    channel_names=["Engine RPM", "Vehicle Speed"],
+    statistics=["min", "max"],
+    event=eng_rpm_event,
+    per_channel_custom_statistics=[
+        PerChannelStatistic(func=percentiles, aggregation_labels=["p50", "p90"]),
+    ],
+    cross_channel_custom_statistics=[
+        CrossChannelStatistic(func=spread, aggregation_labels=["spread"],
+                              channel_name="rpm_speed_spread"),
+    ],
+)
+```
+
+**Descriptor fields**
+
+| Field                | Applies to        | Description                                                                             |
+|----------------------|-------------------|-----------------------------------------------------------------------------------------|
+| `func`               | both              | Callable returning a sequence of scalars (length must equal `aggregation_labels`).      |
+| `aggregation_labels` | both (required)   | Output labels; become `aggregation_label`s in the fact table. Non-empty, unique.        |
+| `params`             | both              | Keyword arguments passed to `func` on every call. Keys must be valid identifiers.       |
+| `inputs`             | cross-channel     | Names (from `channel_names`) of the channels to pass, in order. `None` passes all.      |
+| `channel_name`       | cross-channel     | `channel_name` for all output rows. `None` uses each output's `aggregation_label`.      |
+
+Notes:
+
+- Labels must be **globally unique** across built-ins and all custom statistics, and must not shadow a
+  built-in label.
+- `func` is cloudpickled to Spark executors: use a module-level importable function and never capture
+  Spark objects.
+- A statistic's `func`, labels, resolved inputs, `params`, and (for cross-channel) `channel_name` are part
+  of the definition hash, so changing any of them triggers reprocessing. `channel_name` is included because
+  it is the fact table's merge key — a rename must recompute so old-name rows are pruned rather than left
+  stale in incremental mode.
+
 ### Output schema
 
 **stats_aggregator_fact:**
@@ -276,10 +349,10 @@ A `ValueError` is raised if unsupported statistics are provided.
 |---------------------|----------|--------------------------------------------------|
 | `container_id`      | `int`    | Container identifier.                            |
 | `visual_id`         | `int`    | Foreign key to `stats_aggregator_dimension`.     |
-| `channel_name`      | `str`    | Signal display name.                             |
+| `channel_name`      | `str`    | Signal display name (or a cross-channel statistic's `channel_name`). |
 | `event_id`          | `int`    | Event identifier.                                |
 | `event_instance_id` | `long`   | Foreign key to `event_instance_fact`.            |
-| `aggregation_label` | `str`    | Statistic label (e.g. `"mean"`).                 |
+| `aggregation_label` | `str`    | Statistic label — a built-in (e.g. `"mean"`) or a custom `aggregation_labels` entry. |
 | `statistic_value`   | `double` | Computed statistic value.                        |
 
 **stats_aggregator_dimension:**
@@ -292,7 +365,7 @@ A `ValueError` is raised if unsupported statistics are provided.
 | `page_number`        | `int`        | Page number.                   |
 | `description`        | `str`        | Description.                   |
 | `agg_type`           | `str`        | Aggregation type identifier.   |
-| `statistics`         | `array[str]` | Requested statistic labels.    |
+| `statistics`         | `array[str]` | All output labels: built-ins plus every custom statistic's `aggregation_labels`. |
 | `channel_names`      | `array[str]` | Signal display names.          |
 | `signal_expressions` | `array[str]` | TSAL expression strings.       |
 | `values_unit`        | `str`        | Unit of statistic values.      |

docs/impulse/docs/references/report/index.md

@@ -114,10 +114,10 @@ Only the hashed attributes matter. Anything else is cosmetic and won't trigger r
 | `ContainerEvent`    | `name`                                                      |
 | `Histogram`         | `base_expr`, `bins`, `event`                                |
 | `Histogram2D`       | `x_expr`, `y_expr`, `x_bins`, `y_bins`, `event`             |
-| `StatsAggregator`   | `input_expressions`, `statistics`, `event`                  |
+| `StatsAggregator`   | `input_expressions`, `statistics`, `event`, `channel_names`, custom statistics (labels, functions, inputs, params, cross-channel `channel_name`) |
 | `CalculatedChannel` | `expr` string + `identity`                                  |
 
-Renaming an aggregation, tweaking the description, or swapping `channel_name` or units keeps the hash stable. No reprocessing.
+Renaming an aggregation, tweaking the description, or changing units keeps the hash stable. No reprocessing. `channel_names` (and a cross-channel statistic's `channel_name`) **do** affect the hash for `StatsAggregator` / `PointValueAggregator`, because they are the fact table's `channel_name` merge key — renaming forces a recompute so old-name rows are pruned rather than left stale.
 
 #### Container-update detection
 

docs/impulse/pydoc-markdown.yml

@@ -19,6 +19,7 @@ loaders:
       - impulse_query_engine.analyze.metadata.tag_expression
       - impulse_query_engine.analyze.metadata.metric_expression
       - impulse_query_engine.analyze.query.aggregations.statistic_type
+      - impulse_query_engine.analyze.query.aggregations.custom_statistic
       - impulse_query_engine.analyze.query.solvers.default_solver
       - impulse_query_engine.analyze.query.solvers.solver_config
       - impulse_query_engine.analyze.query.solvers.query_solver