anomaly ¶

def learn_one(self, *args, **kwargs) -> None:
    """
    Update the scaler and the underlying anomaly scaler.

    Parameters
    ----------
    *args
        Depends on whether the underlying anomaly detector
        is supervised or not.

    Returns
    -------
    AnomalyScaler
        The model itself.
    """

    self.anomaly_detector.learn_one(*args, **kwargs)

@abc.abstractmethod
def score_many(self, *args, **kwargs) -> np.ndarray:
    """Return scaled anomaly scores based on raw score provided by
    the wrapped anomaly detector.

    A high score is indicative of an anomaly. A low score corresponds
    to a normal observation.

    Parameters
    ----------
    *args
        Depends on whether the underlying anomaly detector is
        supervised or not.

    Returns
    -------
    Scaled anomaly scores. Larger values indicate more anomalous examples.
    """

def score_one(self, *args, **kwargs):
    """
    Return a scaled anomaly score based on raw score provided by the
    wrapped anomaly detector. Larger values indicate more
    anomalous examples.

    Parameters
    ----------
    *args
        Depends on whether the underlying anomaly detector is
        supervised or not.

    Returns
    -------
    An scaled anomaly score. Larger values indicate more
    anomalous examples.
    """
    raw_score = self.anomaly_detector.score_one(*args, **kwargs)
    mean = self.mean.update(raw_score).get()
    score = raw_score / mean

    return score

def learn_one(self, *args, **kwargs) -> None:
    """
    Update the scaler and the underlying anomaly scaler.

    Parameters
    ----------
    *args
        Depends on whether the underlying anomaly detector
        is supervised or not.

    Returns
    -------
    AnomalyScaler
        The model itself.
    """

    self.anomaly_detector.learn_one(*args, **kwargs)

@abc.abstractmethod
def score_many(self, *args, **kwargs) -> np.ndarray:
    """Return scaled anomaly scores based on raw score provided by
    the wrapped anomaly detector.

    A high score is indicative of an anomaly. A low score corresponds
    to a normal observation.

    Parameters
    ----------
    *args
        Depends on whether the underlying anomaly detector is
        supervised or not.

    Returns
    -------
    Scaled anomaly scores. Larger values indicate more anomalous examples.
    """

def score_one(self, *args, **kwargs):
    """
    Return a scaled anomaly score based on raw score provided by the
    wrapped anomaly detector. Larger values indicate more
    anomalous examples.

    Parameters
    ----------
    *args
        Depends on whether the underlying anomaly detector is
        supervised or not.

    Returns
    -------
    An scaled anomaly score. Larger values indicate more
    anomalous examples.
    """
    raw_score = self.anomaly_detector.score_one(*args, **kwargs)
    min = self.min.update(raw_score).get()
    max = self.max.update(raw_score).get()
    score = (raw_score - min) / (max - min)

    return score

def learn_one(self, *args, **kwargs) -> None:
    """
    Update the scaler and the underlying anomaly scaler.

    Parameters
    ----------
    *args
        Depends on whether the underlying anomaly detector
        is supervised or not.

    Returns
    -------
    AnomalyScaler
        The model itself.
    """

    self.anomaly_detector.learn_one(*args, **kwargs)

@abc.abstractmethod
def score_many(self, *args, **kwargs) -> np.ndarray:
    """Return scaled anomaly scores based on raw score provided by
    the wrapped anomaly detector.

    A high score is indicative of an anomaly. A low score corresponds
    to a normal observation.

    Parameters
    ----------
    *args
        Depends on whether the underlying anomaly detector is
        supervised or not.

    Returns
    -------
    Scaled anomaly scores. Larger values indicate more anomalous examples.
    """

def score_one(self, *args, **kwargs):
    """
    Return a scaled anomaly score based on raw score provided by the
    wrapped anomaly detector. Larger values indicate more
    anomalous examples.

    Parameters
    ----------
    *args
        Depends on whether the underlying anomaly detector
        is supervised or not.

    Returns
    -------
    An scaled anomaly score. Larger values indicate more
    anomalous examples.
    """
    raw_score = self.anomaly_detector.score_one(*args, **kwargs)
    mean = self.mean.update(raw_score).get()
    if self.with_std:
        var = (
            self.sq_mean.update(raw_score**2).get() - mean**2
        )  # todo is this correct?
        score = (raw_score - mean) / var**0.5
    else:
        score = raw_score - mean

    return score

def clone(
    self,
    new_params=None,
    include_attributes: bool = False,
    copy_weights: bool = False,
):
    """Return a fresh estimator instance with (optionally) copied state.

    Parameters
    ----------
    new_params : dict | None
        Parameter overrides for the cloned instance.
    include_attributes : bool, default=False
        If True, runtime state (observed features, buffers) is also copied.
    copy_weights : bool, default=False
        If True, model weights are copied (otherwise the module is re‑initialised).
    """
    new_params = new_params or {}
    copy_weights = new_params.pop("copy_weights", copy_weights)

    params = {**self._get_all_init_params(), **new_params}

    if "module" not in new_params:
        params["module"] = self._rebuild_module()

    new_est = self.__class__(**self._filter_kwargs(self.__class__.__init__, params))

    if copy_weights and hasattr(self.module, "state_dict"):
        new_est.module.load_state_dict(self.module.state_dict())

    if include_attributes:
        new_est._restore_runtime_state(self._get_runtime_state())

    return new_est

def draw(self):  # type: ignore[override]
    """Render a (partial) computational graph of the wrapped model.

    Imports ``graphviz`` and ``torchviz`` lazily. Raises an informative
    ImportError if the optional dependencies are not installed.
    """
    try:  # pragma: no cover
        from torchviz import make_dot  # type: ignore
    except Exception as err:  # noqa: BLE001
        raise ImportError(
            "graphviz and torchviz must be installed to draw the model."
        ) from err

    first_parameter = next(self.module.parameters())
    input_shape = first_parameter.size()
    y_pred = self.module(torch.rand(input_shape))
    return make_dot(y_pred.mean(), params=dict(self.module.named_parameters()))

def learn_many(self, X: pd.DataFrame) -> None:
    """
    Performs one step of training with a batch of examples.

    Parameters
    ----------
    X
        Input batch of examples.
    """

    self._update_observed_features(X)
    X_t = self._df2tensor(X)
    self._learn(X_t)

def learn_one(self, x: dict, y: Any = None) -> None:
    """
    Performs one step of training with a single example.

    Parameters
    ----------
    x
        Input example.
    """
    self._update_observed_features(x)
    self._learn(self._dict2tensor(x))

@classmethod
def load(cls, filepath: Union[str, Path]):
    """Load a previously saved estimator.

    The method reconstructs the estimator class, its wrapped module, optimiser
    state and runtime information (feature names, buffers, etc.).
    """
    with open(filepath, "rb") as f:
        state = pickle.load(f)

    estimator_cls = cls._import_from_path(state["estimator_class"])
    init_params = state["init_params"]

    # Rebuild module if needed
    if "module" in init_params and isinstance(init_params["module"], dict):
        module_info = init_params.pop("module")
        module_cls = cls._import_from_path(module_info["class"])
        module = module_cls(
            **cls._filter_kwargs(module_cls.__init__, module_info["kwargs"])
        )
        if state.get("model_state_dict"):
            module.load_state_dict(state["model_state_dict"])
        init_params["module"] = module

    estimator = estimator_cls(
        **cls._filter_kwargs(estimator_cls.__init__, init_params)
    )

    if state.get("optimizer_state_dict") and hasattr(estimator, "optimizer"):
        try:
            estimator.optimizer.load_state_dict(
                state["optimizer_state_dict"]  # type: ignore[arg-type]
            )
        except Exception:  # noqa: E722
            pass

    estimator._restore_runtime_state(state.get("runtime_state", {}))
    return estimator

def save(self, filepath: Union[str, Path]) -> None:
    """Persist the estimator (architecture, weights, optimiser & runtime state).

    Parameters
    ----------
    filepath : str | Path
        Destination file. Parent directories are created automatically.
    """
    filepath = Path(filepath)
    filepath.parent.mkdir(parents=True, exist_ok=True)

    state = {
        "estimator_class": f"{type(self).__module__}.{type(self).__name__}",
        "init_params": self._get_all_init_params(),
        "model_state_dict": getattr(self.module, "state_dict", lambda: {})(),
        "optimizer_state_dict": getattr(self.optimizer, "state_dict", lambda: {})(),
        "runtime_state": self._get_runtime_state(),
    }

    with open(filepath, "wb") as f:
        pickle.dump(state, f)

def score_many(self, X: pd.DataFrame) -> np.ndarray:
    """
    Returns an anomaly score for the provided batch of examples in
    the form of the autoencoder's reconstruction error.

    Parameters
    ----------
    x
        Input batch of examples.

    Returns
    -------
    float
        Anomaly scores for the given batch of examples. Larger values
        indicate more anomalous examples.
    """
    self._update_observed_features(X)
    x_t = self._df2tensor(X)

    self.module.eval()
    with torch.inference_mode():
        x_pred = self.module(x_t)
    loss = torch.mean(
        self.loss_func(x_pred, x_t, reduction="none"),
        dim=list(range(1, x_t.dim())),
    )
    score = np.asarray(loss.cpu().detach().tolist())
    return score

def score_one(self, x: dict) -> float:
    """
    Returns an anomaly score for the provided example in the form of
    the autoencoder's reconstruction error.

    Parameters
    ----------
    x
        Input example.

    Returns
    -------
    float
        Anomaly score for the given example. Larger values indicate
        more anomalous examples.

    """

    self._update_observed_features(x)
    x_t = self._dict2tensor(x)
    self.module.eval()
    with torch.inference_mode():
        x_pred = self.module(x_t)
    loss = self.loss_func(x_pred, x_t).item()
    return loss

def clone(
    self,
    new_params=None,
    include_attributes: bool = False,
    copy_weights: bool = False,
):
    """Return a fresh estimator instance with (optionally) copied state.

    Parameters
    ----------
    new_params : dict | None
        Parameter overrides for the cloned instance.
    include_attributes : bool, default=False
        If True, runtime state (observed features, buffers) is also copied.
    copy_weights : bool, default=False
        If True, model weights are copied (otherwise the module is re‑initialised).
    """
    new_params = new_params or {}
    copy_weights = new_params.pop("copy_weights", copy_weights)

    params = {**self._get_all_init_params(), **new_params}

    if "module" not in new_params:
        params["module"] = self._rebuild_module()

    new_est = self.__class__(**self._filter_kwargs(self.__class__.__init__, params))

    if copy_weights and hasattr(self.module, "state_dict"):
        new_est.module.load_state_dict(self.module.state_dict())

    if include_attributes:
        new_est._restore_runtime_state(self._get_runtime_state())

    return new_est

def draw(self):  # type: ignore[override]
    """Render a (partial) computational graph of the wrapped model.

    Imports ``graphviz`` and ``torchviz`` lazily. Raises an informative
    ImportError if the optional dependencies are not installed.
    """
    try:  # pragma: no cover
        from torchviz import make_dot  # type: ignore
    except Exception as err:  # noqa: BLE001
        raise ImportError(
            "graphviz and torchviz must be installed to draw the model."
        ) from err

    first_parameter = next(self.module.parameters())
    input_shape = first_parameter.size()
    y_pred = self.module(torch.rand(input_shape))
    return make_dot(y_pred.mean(), params=dict(self.module.named_parameters()))

def learn_one(self, x: dict, y: Any = None) -> None:
    """
    Performs one step of training with a single example,
    scaling the employed learning rate based on the outlier
    probability estimate of the input example.

    Parameters
    ----------
    x
        Input example.

    Returns
    -------
    ProbabilityWeightedAutoencoder
        The autoencoder itself.
    """

    self._update_observed_features(x)
    x_t = self._dict2tensor(x)

    self.module.train()
    x_pred = self.module(x_t)
    loss = self.loss_func(x_pred, x_t)
    self._apply_loss(loss)

@classmethod
def load(cls, filepath: Union[str, Path]):
    """Load a previously saved estimator.

    The method reconstructs the estimator class, its wrapped module, optimiser
    state and runtime information (feature names, buffers, etc.).
    """
    with open(filepath, "rb") as f:
        state = pickle.load(f)

    estimator_cls = cls._import_from_path(state["estimator_class"])
    init_params = state["init_params"]

    # Rebuild module if needed
    if "module" in init_params and isinstance(init_params["module"], dict):
        module_info = init_params.pop("module")
        module_cls = cls._import_from_path(module_info["class"])
        module = module_cls(
            **cls._filter_kwargs(module_cls.__init__, module_info["kwargs"])
        )
        if state.get("model_state_dict"):
            module.load_state_dict(state["model_state_dict"])
        init_params["module"] = module

    estimator = estimator_cls(
        **cls._filter_kwargs(estimator_cls.__init__, init_params)
    )

    if state.get("optimizer_state_dict") and hasattr(estimator, "optimizer"):
        try:
            estimator.optimizer.load_state_dict(
                state["optimizer_state_dict"]  # type: ignore[arg-type]
            )
        except Exception:  # noqa: E722
            pass

    estimator._restore_runtime_state(state.get("runtime_state", {}))
    return estimator

def save(self, filepath: Union[str, Path]) -> None:
    """Persist the estimator (architecture, weights, optimiser & runtime state).

    Parameters
    ----------
    filepath : str | Path
        Destination file. Parent directories are created automatically.
    """
    filepath = Path(filepath)
    filepath.parent.mkdir(parents=True, exist_ok=True)

    state = {
        "estimator_class": f"{type(self).__module__}.{type(self).__name__}",
        "init_params": self._get_all_init_params(),
        "model_state_dict": getattr(self.module, "state_dict", lambda: {})(),
        "optimizer_state_dict": getattr(self.optimizer, "state_dict", lambda: {})(),
        "runtime_state": self._get_runtime_state(),
    }

    with open(filepath, "wb") as f:
        pickle.dump(state, f)

def score_many(self, X: pd.DataFrame) -> np.ndarray:
    """
    Returns an anomaly score for the provided batch of examples in
    the form of the autoencoder's reconstruction error.

    Parameters
    ----------
    x
        Input batch of examples.

    Returns
    -------
    float
        Anomaly scores for the given batch of examples. Larger values
        indicate more anomalous examples.
    """
    self._update_observed_features(X)
    x_t = self._df2tensor(X)

    self.module.eval()
    with torch.inference_mode():
        x_pred = self.module(x_t)
    loss = torch.mean(
        self.loss_func(x_pred, x_t, reduction="none"),
        dim=list(range(1, x_t.dim())),
    )
    score = np.asarray(loss.cpu().detach().tolist())
    return score

def score_one(self, x: dict) -> float:
    """
    Returns an anomaly score for the provided example in the form of
    the autoencoder's reconstruction error.

    Parameters
    ----------
    x
        Input example.

    Returns
    -------
    float
        Anomaly score for the given example. Larger values indicate
        more anomalous examples.

    """

    self._update_observed_features(x)
    x_t = self._dict2tensor(x)
    self.module.eval()
    with torch.inference_mode():
        x_pred = self.module(x_t)
    loss = self.loss_func(x_pred, x_t).item()
    return loss