refactor(quickadapter): deduplicate combined label-weight aggregate (#170) (#195)

Share the combined `select -> impute -> aggregate` pipeline between the value
path (`_compute_combined_label_weights`) and the dependency-mask path
(`compute_label_weight_imputation_dependency_mask`) via a single
`_compute_combined_label_weight_pipeline` helper returning a
`_CombinedWeightPipeline`. The mask branch now derives the per-component and
aggregate non-finite masks and the leading-stable release (#169) from the
shared result instead of re-running selection, imputation, and aggregation.

Behavior is bit-for-bit identical for `compute_label_weights` and
`compute_label_weight_imputation_dependency_mask` across all strategies,
aggregations, fill methods, and causal/non-causal modes, including raised
exception type, message, and order. The two imputers (`_impute_weights` and
`_causal_impute_weights`) stay distinct; the epsilon path keeps its causal
imputer through the shared helper.

diff --git a/quickadapter/user_data/strategies/Utils.py b/quickadapter/user_data/strategies/Utils.py
index 7f8815da1b44c88748a686bb89fdd8ffff777e70..4cf3970807e808e3fb31ada5173abda16c033be3 100644 (file)
--- a/quickadapter/user_data/strategies/Utils.py
+++ b/quickadapter/user_data/strategies/Utils.py
@@ -2651,24 +2651,84 @@ def _aggregate_imputed_metrics(
     )
 
 
-def _compute_combined_label_weights(
+@dataclass(frozen=True, slots=True)
+class _CombinedWeightPipeline:
+    """Single pass of the combined ``select -> impute -> aggregate`` pipeline.
+
+    ``selected`` is the :func:`_select_combined_metrics` output in selection
+    order ``(name, raw pre-imputation values, coefficient)``. ``combined_weights``
+    is the aggregated weights BEFORE any value-path final imputation:
+
+    - ``selected`` empty: ``combined_weights`` is ``np.asarray([], dtype=float)``
+      regardless of ``expected_length``;
+    - ``expected_length`` given and ``selected`` non-empty:
+      ``combined_weights.shape == (expected_length,)``;
+    - ``expected_length`` is ``None`` and ``selected`` non-empty:
+      ``combined_weights`` is the reduction over the selected components' shared
+      length.
+    """
+
+    selected: tuple[tuple[str, NDArray[np.floating], float], ...]
+    combined_weights: NDArray[np.floating]
+
+
+def _compute_combined_label_weight_pipeline(
     metrics: dict[str, list[float]],
     metric_coefficients: dict[str, Any],
     aggregation: CombinedAggregation,
     softmax_temperature: float,
     *,
     impute: Callable[[NDArray[np.floating]], NDArray[np.floating]] = _impute_weights,
-) -> NDArray[np.floating]:
-    selected = _select_combined_metrics(metrics, metric_coefficients)
+    expected_length: int | None = None,
+) -> _CombinedWeightPipeline:
+    """Run the combined ``select -> impute -> aggregate`` pipeline once.
+
+    When ``expected_length`` is not ``None``, raises ``ValueError`` on the first
+    selected component (in selection order) whose shape is not
+    ``(expected_length,)`` and on combined weights whose shape is not
+    ``(expected_length,)``; when ``None`` no shape validation is performed and
+    malformed inputs surface downstream (e.g. via ``np.vstack``).
+    """
+    selected = tuple(_select_combined_metrics(metrics, metric_coefficients))
+    if expected_length is not None:
+        for metric_name, values_array, _coefficient in selected:
+            if values_array.shape != (expected_length,):
+                raise ValueError(
+                    f"Invalid metric {metric_name!r} shape {values_array.shape}: "
+                    f"must be ({expected_length},)"
+                )
     if len(selected) == 0:
-        return np.asarray([], dtype=float)
+        return _CombinedWeightPipeline(selected, np.asarray([], dtype=float))
 
-    return _aggregate_imputed_metrics(
+    combined_weights = _aggregate_imputed_metrics(
         [impute(values) for _, values, _ in selected],
         [coefficient for _, _, coefficient in selected],
         aggregation,
         softmax_temperature,
     )
+    if expected_length is not None and combined_weights.shape != (expected_length,):
+        raise ValueError(
+            f"Invalid combined weights shape {combined_weights.shape}: "
+            f"must be ({expected_length},)"
+        )
+    return _CombinedWeightPipeline(selected, combined_weights)
+
+
+def _compute_combined_label_weights(
+    metrics: dict[str, list[float]],
+    metric_coefficients: dict[str, Any],
+    aggregation: CombinedAggregation,
+    softmax_temperature: float,
+    *,
+    impute: Callable[[NDArray[np.floating]], NDArray[np.floating]] = _impute_weights,
+) -> NDArray[np.floating]:
+    return _compute_combined_label_weight_pipeline(
+        metrics,
+        metric_coefficients,
+        aggregation,
+        softmax_temperature,
+        impute=impute,
+    ).combined_weights
 
 
 def _nonfinite_imputation_dependency_mask(
@@ -2774,23 +2834,21 @@ def compute_label_weight_imputation_dependency_mask(
     if strategy != WEIGHT_STRATEGIES[8]:  # "combined"
         raise ValueError(_invalid_weight_strategy_message(strategy, metrics))
 
+    pipeline = _compute_combined_label_weight_pipeline(
+        metrics,
+        label_weighting["metric_coefficients"],
+        label_weighting["aggregation"],
+        label_weighting["softmax_temperature"],
+        impute=_impute_weights,
+        expected_length=n_indices,
+    )
+
     dependency_mask = np.zeros(n_indices, dtype=bool)
-    imputed_metrics: list[NDArray[np.floating]] = []
-    coefficients_list: list[float] = []
     first_finite_indices: list[int] = []
     every_component_has_finite = True
-    for metric_name, values_array, coefficient in _select_combined_metrics(
-        metrics, label_weighting["metric_coefficients"]
-    ):
-        if values_array.shape != (n_indices,):
-            raise ValueError(
-                f"Invalid metric {metric_name!r} shape {values_array.shape}: "
-                f"must be ({n_indices},)"
-            )
+    for _metric_name, values_array, _coefficient in pipeline.selected:
         component_finite = np.isfinite(values_array)
         dependency_mask |= ~component_finite
-        imputed_metrics.append(_impute_weights(values_array))
-        coefficients_list.append(coefficient)
         if component_finite.any():
             first_finite_indices.append(int(np.argmax(component_finite)))
         else:
@@ -2799,21 +2857,11 @@ def compute_label_weight_imputation_dependency_mask(
             # leading release and defer these pivots to n conservatively.
             every_component_has_finite = False
 
-    if len(imputed_metrics) == 0:
+    if len(pipeline.selected) == 0:
         empty = np.zeros(n_indices, dtype=bool)
         return LabelWeightImputationMasks(dependency_mask, empty, -1)
 
-    combined_weights = _aggregate_imputed_metrics(
-        imputed_metrics,
-        coefficients_list,
-        label_weighting["aggregation"],
-        label_weighting["softmax_temperature"],
-    )
-    if combined_weights.shape != (n_indices,):
-        raise ValueError(
-            f"Invalid combined weights shape {combined_weights.shape}: "
-            f"must be ({n_indices},)"
-        )
+    combined_weights = pipeline.combined_weights
     dependency_mask |= _nonfinite_imputation_dependency_mask(combined_weights)
 
     leading_stable = np.zeros(n_indices, dtype=bool)