From 07219dd5e49cd45e26c7d45b9b048d6b1734643f Mon Sep 17 00:00:00 2001 From: rusty1s Date: Fri, 23 Oct 2026 07:48:30 +0000 Subject: [PATCH 1/6] update --- examples/timeseries/anomaly.py | 17 +- .../kumo/timeseries/anomaly/__init__.py | 6 + sdm/models/kumo/timeseries/anomaly/model.py | 32 ++- .../kumo/timeseries/anomaly/threshold.py | 248 ++++++++++++++++++ sdm/tensor/categorical.py | 17 ++ .../kumo/timeseries/anomaly/test_model.py | 43 ++- .../kumo/timeseries/anomaly/test_threshold.py | 139 ++++++++++ test/tensor/test_categorical.py | 76 ++++++ 8 files changed, 567 insertions(+), 11 deletions(-) create mode 100644 sdm/models/kumo/timeseries/anomaly/threshold.py create mode 100644 test/models/kumo/timeseries/anomaly/test_threshold.py diff --git a/examples/timeseries/anomaly.py b/examples/timeseries/anomaly.py index 047692301..23d4b81c9 100644 --- a/examples/timeseries/anomaly.py +++ b/examples/timeseries/anomaly.py @@ -9,10 +9,19 @@ time = torch.arange(200, device=device, dtype=torch.float32) x = torch.stack([torch.sin(time / 12), torch.cos(time / 20)], dim=-1) -x[140:149] += 3 # Add anomalies. +x[140:150] += 3 # Add anomalies. model = sdm.models.KumoAnomaly(device=device) +out = model(x) -# Higher reconstruction error scores indicate more likely anomalies: -score = model(x).numerical.squeeze(-1) -print(score[131:158]) +# Higher MAE scores indicate more likely anomalies: +print("Reconstruction Error (MAE):") +print(out["mae"].numerical[130:160, 0]) + +# SCS flags anomalies using adaptive segments of the preprocessed series: +print("Anomaly Labels (SCS):") +print(out["scs"].categorical.decode()[130:160, 0]) + +# MACS flags anomalies using variance-weighted score bounds: +print("Anomaly Labels (MACS):") +print(out["macs"].categorical.decode()[130:160, 0]) diff --git a/sdm/models/kumo/timeseries/anomaly/__init__.py b/sdm/models/kumo/timeseries/anomaly/__init__.py index f4fb71b05..c33f73d88 100644 --- a/sdm/models/kumo/timeseries/anomaly/__init__.py +++ b/sdm/models/kumo/timeseries/anomaly/__init__.py @@ -4,8 +4,14 @@ """KumoAnomaly.""" from sdm.models.kumo.timeseries.anomaly.model import KumoAnomaly +from sdm.models.kumo.timeseries.anomaly.threshold import ( + SCSThreshold, + MACSThreshold, +) __all__ = [ "KumoAnomaly", + "SCSThreshold", + "MACSThreshold", ] diff --git a/sdm/models/kumo/timeseries/anomaly/model.py b/sdm/models/kumo/timeseries/anomaly/model.py index f15f06a94..7eb47117d 100644 --- a/sdm/models/kumo/timeseries/anomaly/model.py +++ b/sdm/models/kumo/timeseries/anomaly/model.py @@ -14,7 +14,7 @@ import torch.nn.functional as F from torch import Tensor -from sdm import Recipe, Stype, TableTensor +from sdm import CategoricalTensor, Recipe, Stype, TableTensor from sdm._inference import inference_mode from sdm.models import Callback, Model from sdm.models._huggingface import download_checkpoint @@ -24,6 +24,10 @@ from sdm.models.kumo.timeseries.anomaly.recipe import default_recipe from sdm.models.kumo.timeseries.anomaly.sampling import DPMSolver from sdm.models.kumo.timeseries.anomaly.schedule import DiffusionSchedule +from sdm.models.kumo.timeseries.anomaly.threshold import ( + MACSThreshold, + SCSThreshold, +) from sdm.processing.execution import MemberQuery, RecipeExecution @@ -46,6 +50,10 @@ class KumoAnomaly(Model): reconstruction error across features gives a score per time step; larger scores indicate less accurately reconstructed observations. + SCS and MACS thresholding convert scores to boolean anomaly decisions. + Both use whole-series statistics, including future observations, and + their decisions are not calibrated probabilities. + Args: pretrained: Whether to load the pretrained checkpoint. device: The device. @@ -99,7 +107,7 @@ def forward( num_steps: int = 20, callbacks: Sequence[Callback] | None = None, generator: torch.Generator | None = None, - ) -> TableTensor: # [..., R, 1] + ) -> TableTensor: # [..., R, 3] r"""The forward pass. Args: @@ -116,9 +124,11 @@ def forward( pre-processing and model execution. Returns: - The anomaly scores with shape ``[..., R, 1]``, denoting the mean - absolute reconstruction error across preprocessed features at each - time step. + A table with shape ``[..., R, 3]`` containing numerical ``mae`` + reconstruction scores and categorical boolean ``scs`` and ``macs`` + anomaly decisions. MAE is averaged across preprocessed features at + each time step. Thresholding uses the complete series before + applying the recipe's output pipeline. """ callbacks = () if callbacks is None else callbacks requires_grad = self.training @@ -158,9 +168,19 @@ def forward( generator=generator, ) + scs = SCSThreshold()(out, query.x.numerical) + macs = MACSThreshold()(out) + categories = torch.tensor([False, True], device=out.device) out = TableTensor( - columns={Stype.numerical: ("mae",)}, + columns={ + Stype.numerical: ("mae",), + Stype.categorical: ("scs", "macs"), + }, numerical=out.unsqueeze(-1), + categorical=CategoricalTensor( + code=torch.stack([scs, macs], dim=-1).long(), + categories=(categories, categories), + ), ) for callback in callbacks: diff --git a/sdm/models/kumo/timeseries/anomaly/threshold.py b/sdm/models/kumo/timeseries/anomaly/threshold.py new file mode 100644 index 000000000..06fe63f73 --- /dev/null +++ b/sdm/models/kumo/timeseries/anomaly/threshold.py @@ -0,0 +1,248 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +# Adapted from NVIDIA/Kumo-TS, Kumo-Anomaly/utils/adaptive_threshold.py +# at af74798bbf7d020d3298f545fb9be383560a18ce. + +import torch +from torch import Tensor + +from sdm import TableTensor + + +def _bounds( + scores: Tensor, starts: Tensor, confidence_level: float +) -> tuple[Tensor, Tensor]: + """Compute the SDK's expanding bounds, restarting at segment starts.""" + sums = torch.cat( + [torch.zeros_like(scores[..., :1]), scores.cumsum(-1)], dim=-1 + ) + squares = torch.cat( + [torch.zeros_like(scores[..., :1]), scores.square().cumsum(-1)], dim=-1 + ) + counts = ( + torch.arange(1, scores.size(-1) + 1, device=scores.device) - starts + ) + mean = (sums[..., 1:] - sums.gather(-1, starts)) / counts + mean = torch.where(counts == 1, scores, mean) + variance = ( + squares[..., 1:] - squares.gather(-1, starts) + ) / counts - mean.square() + std = variance.clamp_min(0).sqrt() + width = torch.where(std > 0, 1.5 * std, 0.1) + if confidence_level > 0.95: + width = width * 1.2 + elif confidence_level < 0.90: + width = width * 0.8 + # The first score in each segment always has unscaled +/- 0.1 bounds. + width = torch.where(counts == 1, 0.1, width) + return mean - width, mean + width + + +def _rolling_stats(scores: Tensor, window: int) -> tuple[Tensor, Tensor]: + """Return trailing means and sample variances with min_periods=1.""" + # Center before prefix sums to reduce cancellation for offset scores. + centered = scores - scores[..., :1] + sums = torch.cat( + [torch.zeros_like(scores[..., :1]), centered.cumsum(-1)], dim=-1 + ) + squares = torch.cat( + [torch.zeros_like(scores[..., :1]), centered.square().cumsum(-1)], + dim=-1, + ) + end = torch.arange(1, scores.size(-1) + 1, device=scores.device) + start = (end - window).clamp_min(0) + count = end - start + total = sums[..., end] - sums[..., start] + mean = total / count + scores[..., :1] + variance = squares[..., end] - squares[..., start] - total.square() / count + variance = variance.clamp_min(0) / (count - 1).clamp_min(1) + return mean, torch.where(count > 1, variance, 0) + + +class SCSThreshold(torch.nn.Module): + """Segmented Confidence Sequences (SCS) with APCA segmentation. + + Segment preprocessed observations, compute expanding score bounds per + segment, and retain violations strictly above a whole-series percentile. + Defaults follow the Kumo-Anomaly SDK. Decisions use future observations; + the confidence setting is a heuristic, not a calibrated guarantee. + Segmentation copies feature means to CPU; flags stay on the score device. + + Args: + confidence_level: Bound width setting. + min_segment_size: Minimum observations on each side of an APCA split. + percentile_filter: Global score percentile used as a strict cutoff. + """ + + def __init__( + self, + confidence_level: float = 0.99, + min_segment_size: int = 50, + percentile_filter: float = 95.0, + ) -> None: + super().__init__() + self.confidence_level = confidence_level + self.min_segment_size = min_segment_size + self.percentile_filter = percentile_filter + + def _segment_starts(self, data: Tensor) -> Tensor: + """Return each row's segment start using the published APCA rule.""" + length = data.numel() + rows = torch.arange(length, device=data.device) + variation = data.std(correction=0) / (data.mean() + 1e-8) + # Preserve the signed coefficient of variation used by the SDK. + if variation < 0.1: + size = max(200, length // 15) + return rows // size * size + + starts = data.new_zeros(length, dtype=torch.long) + sums = torch.cat([data.new_zeros(1), data.cumsum(0)]) + squares = torch.cat([data.new_zeros(1), data.square().cumsum(0)]) + start = 0 + while start < length: + starts[start:] = start + if length - start < 2 * self.min_segment_size: + break + first = start + self.min_segment_size + last = length - self.min_segment_size + if last <= first: + break + step = max(1, (last - first) // 200) + splits = torch.arange(first, last, step, device=data.device) + left_sum = sums[splits] - sums[start] + right_sum = sums[-1] - sums[splits] + error = ( + squares[-1] + - squares[start] + - left_sum.square() / (splits - start) + - right_sum.square() / (length - splits) + ) + best = error.argmin() + no_split = ( + squares[-1] + - squares[start] + - (sums[-1] - sums[start]).square() / (length - start) + ) + improvement = 0.7 if variation > 0.3 else 0.5 + if error[best] >= no_split * improvement: + break + start = int(splits[best]) + return starts + + @torch.no_grad() + def forward(self, scores: Tensor, target: Tensor | TableTensor) -> Tensor: + """Return boolean anomaly flags with shape ``[..., R]``. + + Args: + scores: Finite reconstruction scores with shape ``[..., R]``. + target: Finite preprocessed observations of shape ``[..., R, D]`` + without padded features. ``R`` is the number of time steps and + ``D`` the number of features. + """ + scores = scores.double() + data = target.numerical if isinstance(target, TableTensor) else target + if data.shape[:-1] != scores.shape: + raise ValueError( + "Target batch and row dimensions must match scores" + ) + if scores.numel() == 0: + return scores.bool() + data = data.mean(-1).to(device="cpu", dtype=torch.float64) + starts = ( + torch.stack( + [ + self._segment_starts(series) + for series in data.reshape(-1, data.size(-1)) + ] + ) + .reshape(scores.shape) + .to(scores.device) + ) + lower, upper = _bounds(scores, starts, self.confidence_level) + cutoff = scores.quantile( + self.percentile_filter / 100, dim=-1, keepdim=True + ) + return ((scores < lower) | (scores > upper)) & (scores > cutoff) + + +class MACSThreshold(torch.nn.Module): + """Multi-Scale Adaptive Confidence Segments (MACS) from Kumo-Anomaly. + + Combine expanding score bounds with variance-based weights and a + regime-change check, then retain violations strictly above a whole-series + percentile. Defaults follow the SDK, including its shared expanding bounds + across scales. Decisions use future observations; the confidence setting + is a heuristic, not a calibrated guarantee. + + Args: + short_window: Window used for variance-based attention weights. + medium_window: SDK scale setting; bounds remain expanding. + long_window: Window used to identify regime changes. + confidence_level: Bound width setting. + percentile_filter: Global score percentile used as a strict cutoff. + """ + + def __init__( + self, + short_window: int = 10, + medium_window: int = 50, + long_window: int = 250, + confidence_level: float = 0.95, + percentile_filter: float = 97.0, + ) -> None: + super().__init__() + self.short_window = short_window + self.medium_window = medium_window + self.long_window = long_window + self.confidence_level = confidence_level + self.percentile_filter = percentile_filter + + @torch.no_grad() + def forward(self, scores: Tensor) -> Tensor: + """Return boolean anomaly flags with shape ``[..., R]``. + + Args: + scores: Finite reconstruction scores with shape ``[..., R]``, + where ``R`` is the number of time steps. + """ + scores = scores.double() + if scores.numel() == 0: + return scores.bool() + starts = torch.zeros_like(scores, dtype=torch.long) + lower, upper = _bounds(scores, starts, self.confidence_level) + window = max(2, min(self.short_window, scores.size(-1) // 10)) + _, variance = _rolling_stats(scores, window) + maximum = variance.amax(dim=-1, keepdim=True) + normalized = variance / torch.where(maximum > 0, maximum, 1) + group = torch.where(normalized > 0.7, 2, (normalized > 0.3).long()) + weights = scores.new_tensor( + [[0.1, 0.3, 0.6], [0.2, 0.6, 0.2], [0.6, 0.3, 0.1]] + )[group] + short_weight, medium_weight, long_weight = weights.unbind(-1) + # Keep the SDK's weighted arithmetic even though the bounds coincide. + combined_lower = ( + short_weight * lower + medium_weight * lower + long_weight * lower + ) + combined_upper = ( + short_weight * upper + medium_weight * upper + long_weight * upper + ) + mean, variance = _rolling_stats(scores, self.long_window) + std = variance.sqrt() + previous_mean = torch.cat( + [torch.zeros_like(scores[..., :1]), mean[..., :-1]], dim=-1 + ) + previous_std = torch.cat( + [torch.zeros_like(scores[..., :1]), std[..., :-1]], dim=-1 + ) + denominator = previous_std + 1e-8 + regime = ((mean - previous_mean).abs() / denominator > 2) | ( + (std - previous_std).abs() / denominator > 1.5 + ) + violations = (scores < lower) | (scores > upper) + attention = (scores < combined_lower) | (scores > combined_upper) + anomalies = attention & (~regime | violations) + cutoff = scores.quantile( + self.percentile_filter / 100, dim=-1, keepdim=True + ) + return anomalies & (scores > cutoff) diff --git a/sdm/tensor/categorical.py b/sdm/tensor/categorical.py index 2d2172762..20313d70f 100644 --- a/sdm/tensor/categorical.py +++ b/sdm/tensor/categorical.py @@ -338,6 +338,23 @@ def from_tensor(cls, tensor: Tensor) -> Self: categories=categories, ) + def decode(self) -> Tensor: + r"""Decode a single categorical column to its category values. + + Requires exactly one column and no missing values. + + Returns: + A tensor with the same shape and device, using the category + vector's dtype and tensor type. + """ + if self.size(-1) != 1: + raise ValueError( + "Decoding requires exactly one categorical column" + ) + if (self.code < 0).any(): + raise ValueError("Cannot decode missing categorical values") + return self.categories[0][self.code] + # Properties ############################################################## @property diff --git a/test/models/kumo/timeseries/anomaly/test_model.py b/test/models/kumo/timeseries/anomaly/test_model.py index 849dd2874..155456fc0 100644 --- a/test/models/kumo/timeseries/anomaly/test_model.py +++ b/test/models/kumo/timeseries/anomaly/test_model.py @@ -1,9 +1,14 @@ # SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 +import pytest import torch +from torch import Tensor +from sdm import Recipe, TableTensor from sdm.models import KumoAnomaly +from sdm.models.kumo.timeseries.anomaly import MACSThreshold, SCSThreshold +from sdm.processing import Callable from sdm.testing import withCUDA @@ -13,8 +18,44 @@ def test_forward(device: torch.device) -> None: x = torch.randn(2, 130, 3, device=device) out = model(x, num_samples=1, num_steps=1) - assert out.size() == (2, 130, 1) + assert out.size() == (2, 130, 3) assert out.columns["numerical"] == ("mae",) + assert out.columns["categorical"] == ("scs", "macs") assert out.device == device assert out.numerical.isfinite().all() assert (out.numerical >= 0).all() + assert out.categorical.code.shape == (2, 130, 2) + for categories in out.categorical.categories: + torch.testing.assert_close( + categories, torch.tensor([False, True], device=device) + ) + + +@withCUDA +def test_threshold_preprocessed_series( + device: torch.device, monkeypatch: pytest.MonkeyPatch +) -> None: + time = torch.arange(600, device=device, dtype=torch.float64) + x = torch.stack([(time / 20).sin(), (time / 17).cos()], dim=-1) * 0.1 + scores = 0.1 + 0.02 * (time / 7).sin().square() + scores[[0, 50, 199, 200, 201, 299, 398, 400, 450, 599]] += 2 + model = KumoAnomaly(pretrained=False, device=device) + + def reconstruct(x: Tensor, **kwargs: object) -> Tensor: + return scores.expand(x.shape[:-1]) + + monkeypatch.setattr(model.model, "forward", reconstruct) + recipe = Recipe( + features=Callable( + lambda table: TableTensor.from_tensor(table.numerical + 10) + ), + output=Callable(lambda table: table[..., :300, :]), + ) + out = model(x, recipe=recipe) + + torch.testing.assert_close(out.numerical.squeeze(-1), scores[:300]) + expected_scs = SCSThreshold()(scores, x + 10) + expected_macs = MACSThreshold()(scores) + expected = torch.stack([expected_scs, expected_macs], dim=-1).long() + torch.testing.assert_close(out.categorical.code, expected[:300]) + assert not torch.equal(expected_scs, SCSThreshold()(scores, x)) diff --git a/test/models/kumo/timeseries/anomaly/test_threshold.py b/test/models/kumo/timeseries/anomaly/test_threshold.py new file mode 100644 index 000000000..e329f5561 --- /dev/null +++ b/test/models/kumo/timeseries/anomaly/test_threshold.py @@ -0,0 +1,139 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +import pytest +import torch +from torch import Tensor + +from sdm import TableTensor +from sdm.models.kumo.timeseries.anomaly import MACSThreshold, SCSThreshold +from sdm.testing import withCUDA + +KINDS = ("flat", "regimes", "oscillating", "offset_regimes") +SPIKES = [ + 0, + 50, + 149, + 150, + 151, + 199, + 200, + 201, + 299, + 374, + 375, + 376, + 398, + 400, + 450, + 599, +] +# Generated with NVIDIA/Kumo-TS af74798, using the SDK strategy defaults. +SCS_INDICES = ( + [50, 149, 150, 151, 199, 299, 374, 375, 376, 398, 450, 599], + [50, 149, 150, 151, 199, 299, 374, 375, 376, 398, 450, 599], + SPIKES[1:], + [50, 149, 199, 200, 201, 299, 374, 398, 400, 450, 599], +) + + +def _series( + kind: str, device: torch.device, dtype: torch.dtype +) -> tuple[Tensor, Tensor]: + time = torch.arange(600, device=device, dtype=dtype) + target = torch.stack( + [1 + 0.1 * (time / 20).sin(), 1 + 0.1 * (time / 17).cos()], dim=-1 + ) + scores = 0.1 + 0.02 * (time / 7).sin().square() + if kind in ("regimes", "offset_regimes"): + start, end = (200, 400) if kind == "regimes" else (150, 375) + target[start:end] += 9 + target[end:] += 2 + scores[start:end] += 0.4 + scores[end:] += 0.1 + elif kind == "oscillating": + target -= 1 + scores[SPIKES] += 2 + return scores, target + + +@withCUDA +@pytest.mark.parametrize("dtype", [torch.float32, torch.float64]) +@pytest.mark.parametrize("kind", KINDS) +@pytest.mark.parametrize("as_table", [False, True]) +def test_upstream_decisions( + device: torch.device, dtype: torch.dtype, kind: str, as_table: bool +) -> None: + scores, target = _series(kind, device, dtype) + scs = SCSThreshold()( + scores, TableTensor.from_tensor(target) if as_table else target + ) + macs = MACSThreshold()(scores) + for actual, indices in ( + (scs, SCS_INDICES[KINDS.index(kind)]), + (macs, SPIKES[1:]), + ): + expected = torch.zeros(600, device=device, dtype=torch.bool) + expected[indices] = True + assert actual.device == device + torch.testing.assert_close(actual, expected) + + +@withCUDA +def test_batched_decisions(device: torch.device) -> None: + series = [_series(kind, device, torch.float64) for kind in KINDS] + scores = torch.stack([scores for scores, _ in series]).reshape(2, 2, 600) + target = torch.stack([target for _, target in series]).reshape( + 2, 2, 600, 2 + ) + scs = SCSThreshold()(scores, target) + macs = MACSThreshold()(scores) + for index, (score, data) in enumerate(series): + torch.testing.assert_close( + scs[index // 2, index % 2], SCSThreshold()(score, data) + ) + torch.testing.assert_close( + macs[index // 2, index % 2], MACSThreshold()(score) + ) + + +@withCUDA +@pytest.mark.parametrize("length", [0, 1, 2, 100, 401]) +@pytest.mark.parametrize("value", [0.0, 1.0, 1e7]) +def test_constant_scores( + device: torch.device, length: int, value: float +) -> None: + scores = torch.full((2, length), value, device=device) + target = torch.ones(2, length, 2, device=device) + for flags in (SCSThreshold()(scores, target), MACSThreshold()(scores)): + assert flags.shape == scores.shape + assert flags.dtype == torch.bool + assert not flags.any() + + +@withCUDA +@pytest.mark.parametrize( + ("confidence", "first"), [(0.85, 2), (0.9, 7), (0.95, 7), (0.99, 20)] +) +def test_confidence_width( + device: torch.device, confidence: float, first: int +) -> None: + scores = torch.arange(20, device=device, dtype=torch.float64) + expected = scores >= first + scs = SCSThreshold(confidence_level=confidence, percentile_filter=0) + macs = MACSThreshold(confidence_level=confidence, percentile_filter=0) + torch.testing.assert_close( + scs(scores, torch.ones(20, 1, device=device)), expected + ) + torch.testing.assert_close(macs(scores), expected) + + +@withCUDA +def test_percentile_ties(device: torch.device) -> None: + scores = torch.ones(100, device=device) + scores[20:40] = 10 + target = torch.ones(100, 2, device=device) + assert not SCSThreshold()(scores, target).any() + assert not MACSThreshold()(scores).any() + assert SCSThreshold(percentile_filter=70)(scores, target).any() + assert MACSThreshold(percentile_filter=70)(scores).any() diff --git a/test/tensor/test_categorical.py b/test/tensor/test_categorical.py index 8596301da..763a8e19b 100644 --- a/test/tensor/test_categorical.py +++ b/test/tensor/test_categorical.py @@ -74,6 +74,82 @@ def test_to_copy() -> None: assert out.dtype == torch.float32 +@withCUDA +@pytest.mark.parametrize("dtype", [torch.bool, torch.int64, torch.float32]) +@pytest.mark.parametrize("code_dtype", [torch.int32, torch.int64]) +def test_decode( + device: torch.device, dtype: torch.dtype, code_dtype: torch.dtype +) -> None: + categories = torch.tensor([1, 0], dtype=dtype, device=device) + code = torch.tensor( + [[[0], [1], [0]], [[1], [0], [1]]], dtype=code_dtype, device=device + ).transpose(0, 1) + tensor = CategoricalTensor(code=code, categories=(categories,)) + + out = tensor.decode() + + assert type(out) is torch.Tensor + assert out.device == device + assert out.dtype == dtype + expected = torch.tensor( + [[[1], [0]], [[0], [1]], [[1], [0]]], dtype=dtype, device=device + ) + torch.testing.assert_close(out, expected) + + +@withCUDA +def test_decode_strings(device: torch.device) -> None: + tensor = CategoricalTensor( + code=torch.tensor([[1], [0], [1]], device=device), + categories=(StringTensor.from_list(["a", "b"], device=device),), + ) + + out = tensor.decode() + + assert isinstance(out, StringTensor) + assert out.device == device + assert out.tolist() == [["b"], ["a"], ["b"]] + + +@withCUDA +def test_decode_empty(device: torch.device) -> None: + tensor = CategoricalTensor( + code=torch.empty(0, 1, dtype=torch.int64, device=device), + categories=(torch.empty(0, dtype=torch.bool, device=device),), + ) + + out = tensor.decode() + + assert out.shape == (0, 1) + assert out.dtype == torch.bool + assert out.device == device + + +@pytest.mark.parametrize("num_columns", [0, 2]) +def test_decode_requires_single_column(num_columns: int) -> None: + tensor = CategoricalTensor( + code=torch.zeros(3, num_columns, dtype=torch.int64), + categories=tuple( + torch.tensor([False, True]) for _ in range(num_columns) + ), + ) + + with pytest.raises(ValueError, match="exactly one categorical column"): + tensor.decode() + + +@withCUDA +@pytest.mark.parametrize("code", [-1, -2]) +def test_decode_missing(device: torch.device, code: int) -> None: + tensor = CategoricalTensor( + code=torch.tensor([[code]], device=device), + categories=(torch.tensor([False, True], device=device),), + ) + + with pytest.raises(ValueError, match="missing categorical values"): + tensor.decode() + + def test_from_arrow_string_values() -> None: tensor = CategoricalTensor.from_arrow( pa.array(["b", "a", None, "b"]), From ba8f7b1ef912a06a3033ba742f7e4f62067fbb68 Mon Sep 17 00:00:00 2001 From: rusty1s Date: Fri, 23 Oct 2026 09:36:18 +0000 Subject: [PATCH 2/6] update --- .../kumo/timeseries/anomaly/__init__.py | 6 -- sdm/tensor/categorical.py | 59 +++++++++++++------ 2 files changed, 41 insertions(+), 24 deletions(-) diff --git a/sdm/models/kumo/timeseries/anomaly/__init__.py b/sdm/models/kumo/timeseries/anomaly/__init__.py index c33f73d88..f4fb71b05 100644 --- a/sdm/models/kumo/timeseries/anomaly/__init__.py +++ b/sdm/models/kumo/timeseries/anomaly/__init__.py @@ -4,14 +4,8 @@ """KumoAnomaly.""" from sdm.models.kumo.timeseries.anomaly.model import KumoAnomaly -from sdm.models.kumo.timeseries.anomaly.threshold import ( - SCSThreshold, - MACSThreshold, -) __all__ = [ "KumoAnomaly", - "SCSThreshold", - "MACSThreshold", ] diff --git a/sdm/tensor/categorical.py b/sdm/tensor/categorical.py index 20313d70f..61b1f655f 100644 --- a/sdm/tensor/categorical.py +++ b/sdm/tensor/categorical.py @@ -14,7 +14,12 @@ from torch.utils import _pytree as pytree from typing_extensions import override -from sdm.tensor import StringTensor, VarLenTensor +from sdm.tensor import ( + ColumnarTensor, + NullableTensor, + StringTensor, + VarLenTensor, +) from sdm.tensor.io import ( arrow_as_tensor, combine_arrow_chunks, @@ -338,23 +343,6 @@ def from_tensor(cls, tensor: Tensor) -> Self: categories=categories, ) - def decode(self) -> Tensor: - r"""Decode a single categorical column to its category values. - - Requires exactly one column and no missing values. - - Returns: - A tensor with the same shape and device, using the category - vector's dtype and tensor type. - """ - if self.size(-1) != 1: - raise ValueError( - "Decoding requires exactly one categorical column" - ) - if (self.code < 0).any(): - raise ValueError("Cannot decode missing categorical values") - return self.categories[0][self.code] - # Properties ############################################################## @property @@ -367,6 +355,41 @@ def categories(self) -> tuple[Tensor, ...]: r"""Return category vector for each categorical column.""" return self._categories + # Methods ################################################################# + + def decode(self) -> Tensor: + r"""Decode this tensor to its category values.""" + if self.size(-1) == 0: + raise ValueError("Decoding requires at least one column") + + tensors = [] + for tensor in self.split(1, dim=-1): + code = tensor.code.squeeze(-1) + valid = code >= 0 + if not valid.all(): + tensor = tensor.categories[0][code.clamp(min=0)] + if isinstance(tensor, VarLenTensor): + data, offset = tensor.contiguous().data_offset + tensor = tensor.__class__( + data=data, + offset=offset, + valid=valid.reshape(-1), + size=tensor.size(), + ) + elif tensor.is_floating_point(): + tensor[~valid] = torch.nan + else: + tensor = NullableTensor(tensor, valid) + else: + tensor = tensor.categories[0][code] + + tensors.append(tensor) + + if len(tensors) == 1: + return tensors[0].unsqueeze(-1) + + return ColumnarTensor(tensors) + # Decorators ############################################################## @classmethod From 9fe65bfe5a619dee2ba4d186f2f394664fee8bf9 Mon Sep 17 00:00:00 2001 From: rusty1s Date: Fri, 23 Oct 2026 09:49:53 +0000 Subject: [PATCH 3/6] update --- sdm/models/kumo/timeseries/anomaly/model.py | 20 +- .../kumo/timeseries/anomaly/threshold.py | 304 ++++++++---------- .../kumo/timeseries/anomaly/test_model.py | 11 +- .../kumo/timeseries/anomaly/test_threshold.py | 44 ++- 4 files changed, 184 insertions(+), 195 deletions(-) diff --git a/sdm/models/kumo/timeseries/anomaly/model.py b/sdm/models/kumo/timeseries/anomaly/model.py index 7eb47117d..9c4f62baf 100644 --- a/sdm/models/kumo/timeseries/anomaly/model.py +++ b/sdm/models/kumo/timeseries/anomaly/model.py @@ -25,8 +25,8 @@ from sdm.models.kumo.timeseries.anomaly.sampling import DPMSolver from sdm.models.kumo.timeseries.anomaly.schedule import DiffusionSchedule from sdm.models.kumo.timeseries.anomaly.threshold import ( - MACSThreshold, - SCSThreshold, + macs_threshold, + scs_threshold, ) from sdm.processing.execution import MemberQuery, RecipeExecution @@ -50,10 +50,6 @@ class KumoAnomaly(Model): reconstruction error across features gives a score per time step; larger scores indicate less accurately reconstructed observations. - SCS and MACS thresholding convert scores to boolean anomaly decisions. - Both use whole-series statistics, including future observations, and - their decisions are not calibrated probabilities. - Args: pretrained: Whether to load the pretrained checkpoint. device: The device. @@ -126,9 +122,11 @@ def forward( Returns: A table with shape ``[..., R, 3]`` containing numerical ``mae`` reconstruction scores and categorical boolean ``scs`` and ``macs`` - anomaly decisions. MAE is averaged across preprocessed features at - each time step. Thresholding uses the complete series before - applying the recipe's output pipeline. + anomaly labels at each time step. SCS (Segmented Confidence + Sequences) segments the preprocessed series and thresholds scores + using expanding bounds within each segment. MACS (Multi-Scale + Adaptive Confidence Segments) uses variance-weighted expanding + score bounds. """ callbacks = () if callbacks is None else callbacks requires_grad = self.training @@ -168,8 +166,8 @@ def forward( generator=generator, ) - scs = SCSThreshold()(out, query.x.numerical) - macs = MACSThreshold()(out) + scs = scs_threshold(out, query.x.numerical) + macs = macs_threshold(out) categories = torch.tensor([False, True], device=out.device) out = TableTensor( columns={ diff --git a/sdm/models/kumo/timeseries/anomaly/threshold.py b/sdm/models/kumo/timeseries/anomaly/threshold.py index 06fe63f73..072e19ed3 100644 --- a/sdm/models/kumo/timeseries/anomaly/threshold.py +++ b/sdm/models/kumo/timeseries/anomaly/threshold.py @@ -60,189 +60,167 @@ def _rolling_stats(scores: Tensor, window: int) -> tuple[Tensor, Tensor]: return mean, torch.where(count > 1, variance, 0) -class SCSThreshold(torch.nn.Module): - """Segmented Confidence Sequences (SCS) with APCA segmentation. - - Segment preprocessed observations, compute expanding score bounds per - segment, and retain violations strictly above a whole-series percentile. +def _segment_starts(data: Tensor, min_segment_size: int) -> Tensor: + """Return each row's segment start using the published APCA rule.""" + length = data.numel() + rows = torch.arange(length, device=data.device) + variation = data.std(correction=0) / (data.mean() + 1e-8) + # Preserve the signed coefficient of variation used by the SDK. + if variation < 0.1: + size = max(200, length // 15) + return rows // size * size + + starts = data.new_zeros(length, dtype=torch.long) + sums = torch.cat([data.new_zeros(1), data.cumsum(0)]) + squares = torch.cat([data.new_zeros(1), data.square().cumsum(0)]) + start = 0 + while start < length: + starts[start:] = start + if length - start < 2 * min_segment_size: + break + first = start + min_segment_size + last = length - min_segment_size + if last <= first: + break + step = max(1, (last - first) // 200) + splits = torch.arange(first, last, step, device=data.device) + left_sum = sums[splits] - sums[start] + right_sum = sums[-1] - sums[splits] + error = ( + squares[-1] + - squares[start] + - left_sum.square() / (splits - start) + - right_sum.square() / (length - splits) + ) + best = error.argmin() + no_split = ( + squares[-1] + - squares[start] + - (sums[-1] - sums[start]).square() / (length - start) + ) + improvement = 0.7 if variation > 0.3 else 0.5 + if error[best] >= no_split * improvement: + break + start = int(splits[best]) + return starts + + +@torch.no_grad() +def scs_threshold( + scores: Tensor, + target: Tensor | TableTensor, + *, + confidence_level: float = 0.99, + min_segment_size: int = 50, + percentile_filter: float = 95.0, +) -> Tensor: + """Flag anomalies using Segmented Confidence Sequences (SCS). + + Segment preprocessed observations with APCA, compute expanding score + bounds per segment, and retain violations above a whole-series percentile. Defaults follow the Kumo-Anomaly SDK. Decisions use future observations; the confidence setting is a heuristic, not a calibrated guarantee. Segmentation copies feature means to CPU; flags stay on the score device. Args: + scores: Finite reconstruction scores with shape ``[..., R]``. + target: Finite preprocessed observations of shape ``[..., R, D]`` + without padded features. ``R`` is the number of time steps and + ``D`` the number of features. confidence_level: Bound width setting. min_segment_size: Minimum observations on each side of an APCA split. percentile_filter: Global score percentile used as a strict cutoff. - """ - def __init__( - self, - confidence_level: float = 0.99, - min_segment_size: int = 50, - percentile_filter: float = 95.0, - ) -> None: - super().__init__() - self.confidence_level = confidence_level - self.min_segment_size = min_segment_size - self.percentile_filter = percentile_filter - - def _segment_starts(self, data: Tensor) -> Tensor: - """Return each row's segment start using the published APCA rule.""" - length = data.numel() - rows = torch.arange(length, device=data.device) - variation = data.std(correction=0) / (data.mean() + 1e-8) - # Preserve the signed coefficient of variation used by the SDK. - if variation < 0.1: - size = max(200, length // 15) - return rows // size * size - - starts = data.new_zeros(length, dtype=torch.long) - sums = torch.cat([data.new_zeros(1), data.cumsum(0)]) - squares = torch.cat([data.new_zeros(1), data.square().cumsum(0)]) - start = 0 - while start < length: - starts[start:] = start - if length - start < 2 * self.min_segment_size: - break - first = start + self.min_segment_size - last = length - self.min_segment_size - if last <= first: - break - step = max(1, (last - first) // 200) - splits = torch.arange(first, last, step, device=data.device) - left_sum = sums[splits] - sums[start] - right_sum = sums[-1] - sums[splits] - error = ( - squares[-1] - - squares[start] - - left_sum.square() / (splits - start) - - right_sum.square() / (length - splits) - ) - best = error.argmin() - no_split = ( - squares[-1] - - squares[start] - - (sums[-1] - sums[start]).square() / (length - start) - ) - improvement = 0.7 if variation > 0.3 else 0.5 - if error[best] >= no_split * improvement: - break - start = int(splits[best]) - return starts - - @torch.no_grad() - def forward(self, scores: Tensor, target: Tensor | TableTensor) -> Tensor: - """Return boolean anomaly flags with shape ``[..., R]``. - - Args: - scores: Finite reconstruction scores with shape ``[..., R]``. - target: Finite preprocessed observations of shape ``[..., R, D]`` - without padded features. ``R`` is the number of time steps and - ``D`` the number of features. - """ - scores = scores.double() - data = target.numerical if isinstance(target, TableTensor) else target - if data.shape[:-1] != scores.shape: - raise ValueError( - "Target batch and row dimensions must match scores" - ) - if scores.numel() == 0: - return scores.bool() - data = data.mean(-1).to(device="cpu", dtype=torch.float64) - starts = ( - torch.stack( - [ - self._segment_starts(series) - for series in data.reshape(-1, data.size(-1)) - ] - ) - .reshape(scores.shape) - .to(scores.device) - ) - lower, upper = _bounds(scores, starts, self.confidence_level) - cutoff = scores.quantile( - self.percentile_filter / 100, dim=-1, keepdim=True + Returns: + Boolean anomaly flags with shape ``[..., R]``. + """ + scores = scores.double() + data = target.numerical if isinstance(target, TableTensor) else target + if data.shape[:-1] != scores.shape: + raise ValueError("Target batch and row dimensions must match scores") + if scores.numel() == 0: + return scores.bool() + data = data.mean(-1).to(device="cpu", dtype=torch.float64) + starts = ( + torch.stack( + [ + _segment_starts(series, min_segment_size) + for series in data.reshape(-1, data.size(-1)) + ] ) - return ((scores < lower) | (scores > upper)) & (scores > cutoff) - - -class MACSThreshold(torch.nn.Module): - """Multi-Scale Adaptive Confidence Segments (MACS) from Kumo-Anomaly. + .reshape(scores.shape) + .to(scores.device) + ) + lower, upper = _bounds(scores, starts, confidence_level) + cutoff = scores.quantile(percentile_filter / 100, dim=-1, keepdim=True) + return ((scores < lower) | (scores > upper)) & (scores > cutoff) + + +@torch.no_grad() +def macs_threshold( + scores: Tensor, + *, + short_window: int = 10, + medium_window: int = 50, + long_window: int = 250, + confidence_level: float = 0.95, + percentile_filter: float = 97.0, +) -> Tensor: + """Flag anomalies using Multi-Scale Adaptive Confidence Segments (MACS). Combine expanding score bounds with variance-based weights and a - regime-change check, then retain violations strictly above a whole-series + regime-change check, then retain violations above a whole-series percentile. Defaults follow the SDK, including its shared expanding bounds across scales. Decisions use future observations; the confidence setting is a heuristic, not a calibrated guarantee. Args: + scores: Finite reconstruction scores with shape ``[..., R]``, + where ``R`` is the number of time steps. short_window: Window used for variance-based attention weights. medium_window: SDK scale setting; bounds remain expanding. long_window: Window used to identify regime changes. confidence_level: Bound width setting. percentile_filter: Global score percentile used as a strict cutoff. - """ - def __init__( - self, - short_window: int = 10, - medium_window: int = 50, - long_window: int = 250, - confidence_level: float = 0.95, - percentile_filter: float = 97.0, - ) -> None: - super().__init__() - self.short_window = short_window - self.medium_window = medium_window - self.long_window = long_window - self.confidence_level = confidence_level - self.percentile_filter = percentile_filter - - @torch.no_grad() - def forward(self, scores: Tensor) -> Tensor: - """Return boolean anomaly flags with shape ``[..., R]``. - - Args: - scores: Finite reconstruction scores with shape ``[..., R]``, - where ``R`` is the number of time steps. - """ - scores = scores.double() - if scores.numel() == 0: - return scores.bool() - starts = torch.zeros_like(scores, dtype=torch.long) - lower, upper = _bounds(scores, starts, self.confidence_level) - window = max(2, min(self.short_window, scores.size(-1) // 10)) - _, variance = _rolling_stats(scores, window) - maximum = variance.amax(dim=-1, keepdim=True) - normalized = variance / torch.where(maximum > 0, maximum, 1) - group = torch.where(normalized > 0.7, 2, (normalized > 0.3).long()) - weights = scores.new_tensor( - [[0.1, 0.3, 0.6], [0.2, 0.6, 0.2], [0.6, 0.3, 0.1]] - )[group] - short_weight, medium_weight, long_weight = weights.unbind(-1) - # Keep the SDK's weighted arithmetic even though the bounds coincide. - combined_lower = ( - short_weight * lower + medium_weight * lower + long_weight * lower - ) - combined_upper = ( - short_weight * upper + medium_weight * upper + long_weight * upper - ) - mean, variance = _rolling_stats(scores, self.long_window) - std = variance.sqrt() - previous_mean = torch.cat( - [torch.zeros_like(scores[..., :1]), mean[..., :-1]], dim=-1 - ) - previous_std = torch.cat( - [torch.zeros_like(scores[..., :1]), std[..., :-1]], dim=-1 - ) - denominator = previous_std + 1e-8 - regime = ((mean - previous_mean).abs() / denominator > 2) | ( - (std - previous_std).abs() / denominator > 1.5 - ) - violations = (scores < lower) | (scores > upper) - attention = (scores < combined_lower) | (scores > combined_upper) - anomalies = attention & (~regime | violations) - cutoff = scores.quantile( - self.percentile_filter / 100, dim=-1, keepdim=True - ) - return anomalies & (scores > cutoff) + Returns: + Boolean anomaly flags with shape ``[..., R]``. + """ + scores = scores.double() + if scores.numel() == 0: + return scores.bool() + starts = torch.zeros_like(scores, dtype=torch.long) + lower, upper = _bounds(scores, starts, confidence_level) + window = max(2, min(short_window, scores.size(-1) // 10)) + _, variance = _rolling_stats(scores, window) + maximum = variance.amax(dim=-1, keepdim=True) + normalized = variance / torch.where(maximum > 0, maximum, 1) + group = torch.where(normalized > 0.7, 2, (normalized > 0.3).long()) + weights = scores.new_tensor( + [[0.1, 0.3, 0.6], [0.2, 0.6, 0.2], [0.6, 0.3, 0.1]] + )[group] + short_weight, medium_weight, long_weight = weights.unbind(-1) + # Keep the SDK's weighted arithmetic even though the bounds coincide. + combined_lower = ( + short_weight * lower + medium_weight * lower + long_weight * lower + ) + combined_upper = ( + short_weight * upper + medium_weight * upper + long_weight * upper + ) + mean, variance = _rolling_stats(scores, long_window) + std = variance.sqrt() + previous_mean = torch.cat( + [torch.zeros_like(scores[..., :1]), mean[..., :-1]], dim=-1 + ) + previous_std = torch.cat( + [torch.zeros_like(scores[..., :1]), std[..., :-1]], dim=-1 + ) + denominator = previous_std + 1e-8 + regime = ((mean - previous_mean).abs() / denominator > 2) | ( + (std - previous_std).abs() / denominator > 1.5 + ) + violations = (scores < lower) | (scores > upper) + attention = (scores < combined_lower) | (scores > combined_upper) + anomalies = attention & (~regime | violations) + cutoff = scores.quantile(percentile_filter / 100, dim=-1, keepdim=True) + return anomalies & (scores > cutoff) diff --git a/test/models/kumo/timeseries/anomaly/test_model.py b/test/models/kumo/timeseries/anomaly/test_model.py index 155456fc0..5881af7d8 100644 --- a/test/models/kumo/timeseries/anomaly/test_model.py +++ b/test/models/kumo/timeseries/anomaly/test_model.py @@ -7,7 +7,10 @@ from sdm import Recipe, TableTensor from sdm.models import KumoAnomaly -from sdm.models.kumo.timeseries.anomaly import MACSThreshold, SCSThreshold +from sdm.models.kumo.timeseries.anomaly.threshold import ( + macs_threshold, + scs_threshold, +) from sdm.processing import Callable from sdm.testing import withCUDA @@ -54,8 +57,8 @@ def reconstruct(x: Tensor, **kwargs: object) -> Tensor: out = model(x, recipe=recipe) torch.testing.assert_close(out.numerical.squeeze(-1), scores[:300]) - expected_scs = SCSThreshold()(scores, x + 10) - expected_macs = MACSThreshold()(scores) + expected_scs = scs_threshold(scores, x + 10) + expected_macs = macs_threshold(scores) expected = torch.stack([expected_scs, expected_macs], dim=-1).long() torch.testing.assert_close(out.categorical.code, expected[:300]) - assert not torch.equal(expected_scs, SCSThreshold()(scores, x)) + assert not torch.equal(expected_scs, scs_threshold(scores, x)) diff --git a/test/models/kumo/timeseries/anomaly/test_threshold.py b/test/models/kumo/timeseries/anomaly/test_threshold.py index e329f5561..73eac0428 100644 --- a/test/models/kumo/timeseries/anomaly/test_threshold.py +++ b/test/models/kumo/timeseries/anomaly/test_threshold.py @@ -6,7 +6,10 @@ from torch import Tensor from sdm import TableTensor -from sdm.models.kumo.timeseries.anomaly import MACSThreshold, SCSThreshold +from sdm.models.kumo.timeseries.anomaly.threshold import ( + macs_threshold, + scs_threshold, +) from sdm.testing import withCUDA KINDS = ("flat", "regimes", "oscillating", "offset_regimes") @@ -65,10 +68,10 @@ def test_upstream_decisions( device: torch.device, dtype: torch.dtype, kind: str, as_table: bool ) -> None: scores, target = _series(kind, device, dtype) - scs = SCSThreshold()( + scs = scs_threshold( scores, TableTensor.from_tensor(target) if as_table else target ) - macs = MACSThreshold()(scores) + macs = macs_threshold(scores) for actual, indices in ( (scs, SCS_INDICES[KINDS.index(kind)]), (macs, SPIKES[1:]), @@ -86,14 +89,14 @@ def test_batched_decisions(device: torch.device) -> None: target = torch.stack([target for _, target in series]).reshape( 2, 2, 600, 2 ) - scs = SCSThreshold()(scores, target) - macs = MACSThreshold()(scores) + scs = scs_threshold(scores, target) + macs = macs_threshold(scores) for index, (score, data) in enumerate(series): torch.testing.assert_close( - scs[index // 2, index % 2], SCSThreshold()(score, data) + scs[index // 2, index % 2], scs_threshold(score, data) ) torch.testing.assert_close( - macs[index // 2, index % 2], MACSThreshold()(score) + macs[index // 2, index % 2], macs_threshold(score) ) @@ -105,7 +108,7 @@ def test_constant_scores( ) -> None: scores = torch.full((2, length), value, device=device) target = torch.ones(2, length, 2, device=device) - for flags in (SCSThreshold()(scores, target), MACSThreshold()(scores)): + for flags in (scs_threshold(scores, target), macs_threshold(scores)): assert flags.shape == scores.shape assert flags.dtype == torch.bool assert not flags.any() @@ -120,12 +123,19 @@ def test_confidence_width( ) -> None: scores = torch.arange(20, device=device, dtype=torch.float64) expected = scores >= first - scs = SCSThreshold(confidence_level=confidence, percentile_filter=0) - macs = MACSThreshold(confidence_level=confidence, percentile_filter=0) - torch.testing.assert_close( - scs(scores, torch.ones(20, 1, device=device)), expected + scs = scs_threshold( + scores=scores, + target=torch.ones(20, 1, device=device), + confidence_level=confidence, + percentile_filter=0, + ) + macs = macs_threshold( + scores=scores, + confidence_level=confidence, + percentile_filter=0, ) - torch.testing.assert_close(macs(scores), expected) + torch.testing.assert_close(scs, expected) + torch.testing.assert_close(macs, expected) @withCUDA @@ -133,7 +143,7 @@ def test_percentile_ties(device: torch.device) -> None: scores = torch.ones(100, device=device) scores[20:40] = 10 target = torch.ones(100, 2, device=device) - assert not SCSThreshold()(scores, target).any() - assert not MACSThreshold()(scores).any() - assert SCSThreshold(percentile_filter=70)(scores, target).any() - assert MACSThreshold(percentile_filter=70)(scores).any() + assert not scs_threshold(scores, target).any() + assert not macs_threshold(scores).any() + assert scs_threshold(scores, target, percentile_filter=70).any() + assert macs_threshold(scores, percentile_filter=70).any() From 4c51fc128a6b83cbc835df9d372bab2af4e18a33 Mon Sep 17 00:00:00 2001 From: rusty1s Date: Fri, 23 Oct 2026 09:51:11 +0000 Subject: [PATCH 4/6] update --- sdm/models/kumo/timeseries/anomaly/model.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/sdm/models/kumo/timeseries/anomaly/model.py b/sdm/models/kumo/timeseries/anomaly/model.py index 9c4f62baf..e2d669e17 100644 --- a/sdm/models/kumo/timeseries/anomaly/model.py +++ b/sdm/models/kumo/timeseries/anomaly/model.py @@ -166,8 +166,8 @@ def forward( generator=generator, ) - scs = scs_threshold(out, query.x.numerical) - macs = macs_threshold(out) + scs = scs_threshold(out, query.x.numerical).long() + macs = macs_threshold(out).long() categories = torch.tensor([False, True], device=out.device) out = TableTensor( columns={ @@ -176,7 +176,7 @@ def forward( }, numerical=out.unsqueeze(-1), categorical=CategoricalTensor( - code=torch.stack([scs, macs], dim=-1).long(), + code=torch.stack([scs, macs], dim=-1), categories=(categories, categories), ), ) From b9d896b9992592fe0a6708801e54fea1b1b64b2f Mon Sep 17 00:00:00 2001 From: rusty1s Date: Fri, 23 Oct 2026 09:53:08 +0000 Subject: [PATCH 5/6] update --- .../kumo/timeseries/anomaly/test_model.py | 39 ------------------- 1 file changed, 39 deletions(-) diff --git a/test/models/kumo/timeseries/anomaly/test_model.py b/test/models/kumo/timeseries/anomaly/test_model.py index 5881af7d8..6e1c6c71e 100644 --- a/test/models/kumo/timeseries/anomaly/test_model.py +++ b/test/models/kumo/timeseries/anomaly/test_model.py @@ -1,17 +1,9 @@ # SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 -import pytest import torch -from torch import Tensor -from sdm import Recipe, TableTensor from sdm.models import KumoAnomaly -from sdm.models.kumo.timeseries.anomaly.threshold import ( - macs_threshold, - scs_threshold, -) -from sdm.processing import Callable from sdm.testing import withCUDA @@ -27,38 +19,7 @@ def test_forward(device: torch.device) -> None: assert out.device == device assert out.numerical.isfinite().all() assert (out.numerical >= 0).all() - assert out.categorical.code.shape == (2, 130, 2) for categories in out.categorical.categories: torch.testing.assert_close( categories, torch.tensor([False, True], device=device) ) - - -@withCUDA -def test_threshold_preprocessed_series( - device: torch.device, monkeypatch: pytest.MonkeyPatch -) -> None: - time = torch.arange(600, device=device, dtype=torch.float64) - x = torch.stack([(time / 20).sin(), (time / 17).cos()], dim=-1) * 0.1 - scores = 0.1 + 0.02 * (time / 7).sin().square() - scores[[0, 50, 199, 200, 201, 299, 398, 400, 450, 599]] += 2 - model = KumoAnomaly(pretrained=False, device=device) - - def reconstruct(x: Tensor, **kwargs: object) -> Tensor: - return scores.expand(x.shape[:-1]) - - monkeypatch.setattr(model.model, "forward", reconstruct) - recipe = Recipe( - features=Callable( - lambda table: TableTensor.from_tensor(table.numerical + 10) - ), - output=Callable(lambda table: table[..., :300, :]), - ) - out = model(x, recipe=recipe) - - torch.testing.assert_close(out.numerical.squeeze(-1), scores[:300]) - expected_scs = scs_threshold(scores, x + 10) - expected_macs = macs_threshold(scores) - expected = torch.stack([expected_scs, expected_macs], dim=-1).long() - torch.testing.assert_close(out.categorical.code, expected[:300]) - assert not torch.equal(expected_scs, scs_threshold(scores, x)) From 53295e70963e0042ece6d9ecf1f8f1a0a19a26ae Mon Sep 17 00:00:00 2001 From: rusty1s Date: Fri, 23 Oct 2026 09:54:30 +0000 Subject: [PATCH 6/6] update --- .../kumo/timeseries/anomaly/test_threshold.py | 143 ++---------------- 1 file changed, 9 insertions(+), 134 deletions(-) diff --git a/test/models/kumo/timeseries/anomaly/test_threshold.py b/test/models/kumo/timeseries/anomaly/test_threshold.py index 73eac0428..700dc5d30 100644 --- a/test/models/kumo/timeseries/anomaly/test_threshold.py +++ b/test/models/kumo/timeseries/anomaly/test_threshold.py @@ -1,149 +1,24 @@ # SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 -import pytest import torch -from torch import Tensor -from sdm import TableTensor from sdm.models.kumo.timeseries.anomaly.threshold import ( macs_threshold, scs_threshold, ) from sdm.testing import withCUDA -KINDS = ("flat", "regimes", "oscillating", "offset_regimes") -SPIKES = [ - 0, - 50, - 149, - 150, - 151, - 199, - 200, - 201, - 299, - 374, - 375, - 376, - 398, - 400, - 450, - 599, -] -# Generated with NVIDIA/Kumo-TS af74798, using the SDK strategy defaults. -SCS_INDICES = ( - [50, 149, 150, 151, 199, 299, 374, 375, 376, 398, 450, 599], - [50, 149, 150, 151, 199, 299, 374, 375, 376, 398, 450, 599], - SPIKES[1:], - [50, 149, 199, 200, 201, 299, 374, 398, 400, 450, 599], -) - - -def _series( - kind: str, device: torch.device, dtype: torch.dtype -) -> tuple[Tensor, Tensor]: - time = torch.arange(600, device=device, dtype=dtype) - target = torch.stack( - [1 + 0.1 * (time / 20).sin(), 1 + 0.1 * (time / 17).cos()], dim=-1 - ) - scores = 0.1 + 0.02 * (time / 7).sin().square() - if kind in ("regimes", "offset_regimes"): - start, end = (200, 400) if kind == "regimes" else (150, 375) - target[start:end] += 9 - target[end:] += 2 - scores[start:end] += 0.4 - scores[end:] += 0.1 - elif kind == "oscillating": - target -= 1 - scores[SPIKES] += 2 - return scores, target - @withCUDA -@pytest.mark.parametrize("dtype", [torch.float32, torch.float64]) -@pytest.mark.parametrize("kind", KINDS) -@pytest.mark.parametrize("as_table", [False, True]) -def test_upstream_decisions( - device: torch.device, dtype: torch.dtype, kind: str, as_table: bool -) -> None: - scores, target = _series(kind, device, dtype) - scs = scs_threshold( - scores, TableTensor.from_tensor(target) if as_table else target - ) - macs = macs_threshold(scores) - for actual, indices in ( - (scs, SCS_INDICES[KINDS.index(kind)]), - (macs, SPIKES[1:]), - ): - expected = torch.zeros(600, device=device, dtype=torch.bool) - expected[indices] = True - assert actual.device == device - torch.testing.assert_close(actual, expected) +def test_threshold(device: torch.device) -> None: + scores = torch.full((2, 400), 0.1, device=device) + scores[0, [0, 50, 200, 350]] = 2 + target = torch.ones(2, 400, 1, device=device) + expected = torch.zeros_like(scores, dtype=torch.bool) + expected[0, [50, 350]] = True + torch.testing.assert_close(scs_threshold(scores, target), expected) -@withCUDA -def test_batched_decisions(device: torch.device) -> None: - series = [_series(kind, device, torch.float64) for kind in KINDS] - scores = torch.stack([scores for scores, _ in series]).reshape(2, 2, 600) - target = torch.stack([target for _, target in series]).reshape( - 2, 2, 600, 2 - ) - scs = scs_threshold(scores, target) - macs = macs_threshold(scores) - for index, (score, data) in enumerate(series): - torch.testing.assert_close( - scs[index // 2, index % 2], scs_threshold(score, data) - ) - torch.testing.assert_close( - macs[index // 2, index % 2], macs_threshold(score) - ) - - -@withCUDA -@pytest.mark.parametrize("length", [0, 1, 2, 100, 401]) -@pytest.mark.parametrize("value", [0.0, 1.0, 1e7]) -def test_constant_scores( - device: torch.device, length: int, value: float -) -> None: - scores = torch.full((2, length), value, device=device) - target = torch.ones(2, length, 2, device=device) - for flags in (scs_threshold(scores, target), macs_threshold(scores)): - assert flags.shape == scores.shape - assert flags.dtype == torch.bool - assert not flags.any() - - -@withCUDA -@pytest.mark.parametrize( - ("confidence", "first"), [(0.85, 2), (0.9, 7), (0.95, 7), (0.99, 20)] -) -def test_confidence_width( - device: torch.device, confidence: float, first: int -) -> None: - scores = torch.arange(20, device=device, dtype=torch.float64) - expected = scores >= first - scs = scs_threshold( - scores=scores, - target=torch.ones(20, 1, device=device), - confidence_level=confidence, - percentile_filter=0, - ) - macs = macs_threshold( - scores=scores, - confidence_level=confidence, - percentile_filter=0, - ) - torch.testing.assert_close(scs, expected) - torch.testing.assert_close(macs, expected) - - -@withCUDA -def test_percentile_ties(device: torch.device) -> None: - scores = torch.ones(100, device=device) - scores[20:40] = 10 - target = torch.ones(100, 2, device=device) - assert not scs_threshold(scores, target).any() - assert not macs_threshold(scores).any() - assert scs_threshold(scores, target, percentile_filter=70).any() - assert macs_threshold(scores, percentile_filter=70).any() + expected[0, 200] = True + torch.testing.assert_close(macs_threshold(scores), expected)