[SYSTEMDS-3968] Add test set evaluator

src/main/python/systemds/scuro/dataloader/timeseries_loader.py

@@ -52,11 +52,13 @@ def __init__(
         normalize: bool = True,
         file_format: str = "npy",
         modality_type: Optional[ModalityType] = ModalityType.TIMESERIES,
+        channel_index: Optional[int] = None,
     ):
         super().__init__(source_path, indices, data_type, chunk_size, modality_type)
         self.signal_names = signal_names
         self.sampling_rate = sampling_rate
         self.normalize = normalize
+        self.channel_index = channel_index
         self.file_format = file_format.lower()
         self.stats = self.get_stats(source_path, sampling_rate)
         if self.file_format not in ["npy", "mat", "hdf5", "txt"]:

src/main/python/systemds/scuro/drsearch/ranking.py

@@ -21,11 +21,127 @@
 
 from typing import Callable, Iterable, Optional
 
+import numpy as np
+
+
+def _operator_signature(entry) -> frozenset:
+    """Set of operator class names in an entry's DAG, ignoring hyperparameters."""
+    try:
+        return frozenset(
+            node.operation.__name__
+            for node in entry.dag.nodes
+            if getattr(node, "operation", None) is not None
+        )
+    except Exception:
+        return frozenset()
+
+
+def _dag_size(entry) -> int:
+    try:
+        return sum(
+            1
+            for node in entry.dag.nodes
+            if getattr(node, "operation", None) is not None
+        )
+    except Exception:
+        return 0
+
+
+def rank_by_robustness(
+    entries: Iterable,
+    *,
+    performance_metric_name: str = "accuracy",
+    neighbourhood_weight: float = 0.5,
+    sharpness: int = 4,
+    one_se_parsimony: bool = True,
+    cache_scores: bool = True,
+    score_attr: str = "robustness_score",
+):
+    entries = list(entries)
+    if not entries:
+        return [], []
+
+    def perf_of(entry):
+        if entry is None:
+            return None
+        try:
+            score = float(entry.val_score[performance_metric_name])
+        except (KeyError, TypeError, ValueError):
+            return None
+        return score if np.isfinite(score) else None
+
+    indexed_entries = [
+        (index, entry, score)
+        for index, entry in enumerate(entries)
+        if (score := perf_of(entry)) is not None
+    ]
+    if not indexed_entries:
+        return [], []
+
+    original_indices, entries, performance = zip(*indexed_entries)
+    entries = list(entries)
+    perf = np.array(performance, dtype=float)
+    sizes = np.array([_dag_size(e) if e is not None else 0 for e in entries], float)
+
+    smoothed = perf
+    if neighbourhood_weight > 0.0 and len(entries) > 1:
+        signatures = [
+            _operator_signature(e) if e is not None else frozenset() for e in entries
+        ]
+        vocabulary = sorted({op for sig in signatures for op in sig})
+        if vocabulary:
+            position = {op: i for i, op in enumerate(vocabulary)}
+            membership = np.zeros((len(entries), len(vocabulary)), dtype=np.float32)
+            for row, sig in enumerate(signatures):
+                for op in sig:
+                    membership[row, position[op]] = 1.0
+            intersection = membership @ membership.T
+            counts = membership.sum(1)
+            union = counts[:, None] + counts[None, :] - intersection
+            jaccard = np.divide(
+                intersection,
+                union,
+                out=np.zeros_like(intersection),
+                where=union > 0,
+            )
+            weights = jaccard**sharpness
+            denominator = weights.sum(1)
+            neighbourhood = np.divide(
+                (weights * perf[None, :].astype(np.float32)).sum(1),
+                denominator,
+                out=perf.astype(np.float32).copy(),
+                where=denominator > 0,
+            )
+            smoothed = (
+                1.0 - neighbourhood_weight
+            ) * perf + neighbourhood_weight * neighbourhood
+
+    if cache_scores:
+        for entry, score in zip(entries, smoothed):
+            if entry is not None:
+                setattr(entry, score_attr, float(score))
+
+    if one_se_parsimony and len(entries) > 1:
+        standard_error = float(smoothed.std()) / np.sqrt(len(smoothed))
+        threshold = float(smoothed.max()) - standard_error
+        keys = [
+            (True, -sz, float(s)) if s >= threshold else (False, 0.0, float(s))
+            for s, sz in zip(smoothed, sizes)
+        ]
+    else:
+        keys = [(True, 0.0, float(s)) for s in smoothed]
+
+    local_indices = sorted(range(len(entries)), key=lambda i: keys[i], reverse=True)
+    sorted_entries = [entries[i] for i in local_indices]
+    sorted_indices = [original_indices[i] for i in local_indices]
+
+    return sorted_entries, sorted_indices
+
 
 def rank_by_tradeoff(
     entries: Iterable,
     *,
-    weights=(0.7, 0.3),
+    weights=(1.0, 0.0),
     performance_metric_name: str = "accuracy",
     runtime_accessor: Optional[Callable[[object], float]] = None,
     cache_scores: bool = True,
@@ -48,8 +164,10 @@ def runtime_accessor(entry):
             task = getattr(entry, "task_time", 0.0)
             return rep + task
 
-    performance = [float(performance_score_accessor(e)) for e in entries]
-    runtimes = [float(runtime_accessor(e)) for e in entries]
+    performance = [
+        float(performance_score_accessor(e)) if e is not None else 0.0 for e in entries
+    ]
+    runtimes = [float(runtime_accessor(e)) if e is not None else 0.0 for e in entries]
 
     perf_min, perf_max = min(performance), max(performance)
     run_min, run_max = min(runtimes), max(runtimes)
@@ -76,17 +194,25 @@ def safe_normalize(values, vmin, vmax):
 
     if cache_scores:
         for entry, score in zip(entries, scores):
+            if entry is None:
+                continue
             if hasattr(entry, score_attr):
                 setattr(entry, score_attr, score)
             else:
                 setattr(entry, score_attr, score)
 
-    sorted_entries = sorted(entries, key=lambda e: e.tradeoff_score, reverse=True)
+    sorted_entries = sorted(
+        entries,
+        key=lambda e: e.tradeoff_score if hasattr(e, "tradeoff_score") else 0.0,
+        reverse=True,
+    )
 
     sorted_indices = [
         i
         for i, _ in sorted(
-            enumerate(entries), key=lambda pair: pair[1].tradeoff_score, reverse=True
+            enumerate(entries),
+            key=lambda pair: pair[1].tradeoff_score if pair is not None else None,
+            reverse=True,
         )
     ]
 

src/main/python/systemds/scuro/drsearch/task.py

@@ -62,6 +62,18 @@ def compute_averages(self):
             self.average_scores[self.metrics] = np.mean(self.scores[self.metrics])
         return self
 
+    def fold_scores(self):
+        return {
+            metric: [float(v) for v in values] for metric, values in self.scores.items()
+        }
+
+    def score_stds(self):
+        """Per-metric standard deviation across folds (0.0 for a single fold)."""
+        return {
+            metric: (float(np.std(values, ddof=1)) if len(values) > 1 else 0.0)
+            for metric, values in self.scores.items()
+        }
+
 
 class Task:
     def __init__(
@@ -95,6 +107,7 @@ def __init__(
         self.measure_performance = measure_performance
         self.inference_time = []
         self.training_time = []
+        self.last_run_timing = {}
         self.expected_dim = 1
         self.performance_measures = performance_measures
         self.train_scores = PerformanceMeasure("train", performance_measures)
@@ -286,12 +299,109 @@ def run(self, data):
         if hasattr(model, "clean_up"):
             model.clean_up()
             del model
+
+        self.last_run_timing = {
+            "train_time_per_fold_s": list(self.training_time),
+            "test_inference_time_per_fold_s": list(self.inference_time),
+            "train_time_mean_s": (
+                float(np.mean(self.training_time)) if self.training_time else 0.0
+            ),
+            "test_inference_time_mean_s": (
+                float(np.mean(self.inference_time)) if self.inference_time else 0.0
+            ),
+            "n_test_instances": len(self.test_indices) if self.test_indices else 0,
+        }
         return [
             self.train_scores.compute_averages(),
             self.val_scores.compute_averages(),
             self.test_scores.compute_averages(),
         ]
 
+    def fit_once_and_time_inference(
+        self, data, latency_repeats: int = 200, latency_warmup: int = 20
+    ):
+        model = self.create_model()
+
+        train_X = self._gather_by_indices(data, self.train_indices)
+        train_y = self._gather_by_indices(self.labels, self.train_indices)
+        test_X = self._gather_by_indices(data, self.test_indices)
+        test_y = self._gather_by_indices(self.labels, self.test_indices)
+
+        t0 = time.perf_counter()
+        model.fit(train_X, train_y, test_X, test_y)
+        fit_time = time.perf_counter() - t0
+
+        t0 = time.perf_counter()
+        test_score = model.test(np.asarray(test_X), test_y)
+        batch_inference_time = time.perf_counter() - t0
+
+        latency = self._measure_single_sample_latency(
+            model, test_X, test_y, latency_repeats, latency_warmup
+        )
+
+        scores = PerformanceMeasure("test_single_fit", self.performance_measures)
+        scores.add_scores(test_score[0])
+        scores.compute_averages()
+
+        if hasattr(model, "clean_up"):
+            model.clean_up()
+
+        n_test = len(test_X)
+        return {
+            "single_fit_train_time_s": fit_time,
+            "test_batch_inference_time_s": batch_inference_time,
+            "test_inference_time_per_instance_ms": (
+                batch_inference_time / n_test * 1000.0 if n_test else 0.0
+            ),
+            "n_test_instances": n_test,
+            "single_fit_test_scores": scores.average_scores,
+            **latency,
+        }
+
+    @staticmethod
+    def _prediction_callable(model):
+        if hasattr(model, "predict"):
+            return model.predict, "model.predict"
+        clf = getattr(model, "clf", None)
+        if clf is not None and hasattr(clf, "predict"):
+            return clf.predict, "clf.predict"
+        return None, None
+
+    def _measure_single_sample_latency(
+        self, model, test_X, test_y, repeats: int, warmup: int
+    ):
+        empty = {
+            "inference_latency_ms_median": None,
+            "inference_latency_ms_p95": None,
+            "inference_latency_samples": 0,
+            "inference_latency_source": None,
+        }
+        if repeats <= 0 or not len(test_X):
+            return empty
+
+        predict, source = self._prediction_callable(model)
+        if predict is None:
+            return empty
+
+        timings = []
+        for i in range(warmup + repeats):
+            sample_X = np.asarray([test_X[i % len(test_X)]])
+            t0 = time.perf_counter()
+            try:
+                predict(sample_X)
+            except Exception:
+                return empty
+            elapsed = (time.perf_counter() - t0) * 1000.0
+            if i >= warmup:
+                timings.append(elapsed)
+
+        return {
+            "inference_latency_ms_median": float(np.median(timings)),
+            "inference_latency_ms_p95": float(np.percentile(timings, 95)),
+            "inference_latency_samples": len(timings),
+            "inference_latency_source": source,
+        }
+
     def _reset_params(self):
         self.inference_time = []
         self.training_time = []