feat: support for train transformers (#354)

Introduce train transformers.
Note: train transformers are fitted first and then the preprocess
transformers are fitted next.


---------

Signed-off-by: s0nicboOm <[email protected]>

numalogic/config/_config.py

@@ -94,8 +94,7 @@ class TrainerConf:
     retry_sec: int = 600  # 10 min
     batch_size: int = 64
     data_freq_sec: int = 60
-    # TODO: Support trainer based transform models
-    max_value_map: Optional[dict[str, float]] = None
+    transforms: Optional[list[ModelInfo]] = None
     pltrainer_conf: LightningTrainerConf = field(default_factory=LightningTrainerConf)
 
 

numalogic/transforms/_stateless.py

@@ -10,7 +10,7 @@
 # limitations under the License.
 
 from collections.abc import Sequence
-from typing import Union, Optional
+from typing import Optional, Union
 
 import numpy as np
 import numpy.typing as npt
@@ -78,27 +78,27 @@ def __init__(
         lower: Optional[Union[float, Sequence[float]]] = None,
         upper: Optional[Union[float, Sequence[float]]] = None,
     ):
-        self._validate_args(lower, upper)
-        self.lower = lower
-        self.upper = upper
+        self.lower, self.upper = self._validate_args(lower, upper)
 
     @staticmethod
     def _validate_args(
-        lower: Union[float, Sequence[float]], upper: Union[float, Sequence[float]]
-    ) -> None:
+        lower: Optional[Union[float, Sequence[float]]],
+        upper: Optional[Union[float, Sequence[float]]],
+    ) -> Optional[tuple[Optional[Union[float, npt.NDArray]], Optional[Union[float, npt.NDArray]]]]:
         if lower is None and upper is None:
             raise ValueError("At least one of lower or upper should be provided.")
-
-        if isinstance(lower, Sequence) and isinstance(upper, Sequence) and len(lower) != len(upper):
-            raise ValueError("lower and upper should have the same length.")
+        if isinstance(lower, Sequence) and isinstance(upper, Sequence):
+            if len(lower) != len(upper):
+                raise ValueError("lower and upper should have the same length.")
+            lower, upper = np.asarray(lower, dtype=np.float32), np.asarray(upper, dtype=np.float32)
+        if upper is not None and lower is not None and np.any(lower > upper):
+            raise ValueError("lower value should be less than or equal to upper value")
+        return lower, upper
 
     def transform(self, x: npt.NDArray[float], **__) -> npt.NDArray[float]:
         _df = pd.DataFrame(x, dtype=np.float32)
-        if (self.lower is not None) and (self.upper is not None):
-            return _df.clip(lower=self.lower, upper=self.upper, axis=1).to_numpy(dtype=np.float32)
-        if self.upper is not None:
-            return _df.clip(upper=self.upper, axis=1).to_numpy(dtype=np.float32)
-        return _df.clip(lower=self.lower, axis=1).to_numpy(dtype=np.float32)
+        _df = _df.clip(upper=self.upper, lower=self.lower, axis=1)
+        return _df.to_numpy(dtype=np.float32)
 
 
 class GaussianNoiseAdder(StatelessTransformer):

numalogic/udfs/trainer/_base.py

@@ -12,13 +12,13 @@
 from torch.utils.data import DataLoader
 
 from numalogic.config import PreprocessFactory, ModelFactory, ThresholdFactory, RegistryFactory
-from numalogic.config._config import NumalogicConf
+from numalogic.config._config import NumalogicConf, ModelInfo
 from numalogic.models.autoencoder import TimeseriesTrainer
 from numalogic.tools.data import StreamingDataset
 from numalogic.tools.exceptions import ConfigNotFoundError, RedisRegistryError
 from numalogic.tools.types import redis_client_t, artifact_t, KEYS, KeyedArtifact
 from numalogic.udfs import NumalogicUDF
-from numalogic.udfs._config import PipelineConf, MLPipelineConf
+from numalogic.udfs._config import PipelineConf
 from numalogic.udfs._metrics import (
     REDIS_ERROR_COUNTER,
     INSUFFICIENT_DATA_COUNTER,
@@ -76,6 +76,7 @@ def compute(
         model: artifact_t,
         input_: npt.NDArray[float],
         preproc_clf: Optional[artifact_t] = None,
+        trainer_transform: Optional[artifact_t] = None,
         threshold_clf: Optional[artifact_t] = None,
         numalogic_cfg: Optional[NumalogicConf] = None,
     ) -> dict[str, KeyedArtifact]:
@@ -86,6 +87,7 @@ def compute(
             model: Model artifact
             input_: Input data
             preproc_clf: Preprocessing artifact
+            trainer_transform: trainer specific preprocessing artifacts
             threshold_clf: Thresholding artifact
             numalogic_cfg: Numalogic configuration
 
@@ -101,6 +103,9 @@ def compute(
             raise ConfigNotFoundError("Numalogic Trainer config not found!")
         dict_artifacts = {}
         trainer_cfg = numalogic_cfg.trainer
+        if trainer_transform:
+            input_ = trainer_transform.fit_transform(input_)
+
         if preproc_clf:
             input_ = preproc_clf.fit_transform(input_)
             dict_artifacts["preproc_clf"] = KeyedArtifact(
@@ -123,6 +128,7 @@ def compute(
 
         if threshold_clf:
             threshold_clf.fit(train_reconerr)
+            _LOGGER.info("Fit data using threshold model")
             dict_artifacts["threshold_clf"] = KeyedArtifact(
                 dkeys=[numalogic_cfg.threshold.name],
                 artifact=threshold_clf,
@@ -218,9 +224,7 @@ def exec(self, keys: list[str], datum: Datum) -> Messages:
         _LOGGER.info("%s - Data fetched, shape: %s", payload.uuid, df.shape)
 
         # Construct feature array
-        x_train, nan_counter, inf_counter = self.get_feature_arr(
-            df, _conf.metrics, max_value_map=_conf.numalogic_conf.trainer.max_value_map
-        )
+        x_train, nan_counter, inf_counter = self.get_feature_arr(df, _conf.metrics)
         _add_summary(
             summary=NAN_SUMMARY,
             labels=_metric_label_values,
@@ -233,15 +237,17 @@ def exec(self, keys: list[str], datum: Datum) -> Messages:
         )
 
         # Initialize artifacts
-        preproc_clf = self._construct_preproc_clf(_conf)
+        preproc_clf = self._construct_clf(_conf.numalogic_conf.preprocess)
+        trainer_transform = self._construct_clf(_conf.numalogic_conf.trainer.transforms)
         model = self._model_factory.get_instance(_conf.numalogic_conf.model)
         thresh_clf = self._thresh_factory.get_instance(_conf.numalogic_conf.threshold)
 
         # Train artifacts
         dict_artifacts = self.compute(
-            model,
-            x_train,
+            model=model,
+            input_=x_train,
             preproc_clf=preproc_clf,
+            trainer_transform=trainer_transform,
             threshold_clf=thresh_clf,
             numalogic_cfg=_conf.numalogic_conf,
         )
@@ -274,9 +280,11 @@ def exec(self, keys: list[str], datum: Datum) -> Messages:
         )
         return Messages(Message.to_drop())
 
-    def _construct_preproc_clf(self, _conf: MLPipelineConf) -> Optional[artifact_t]:
+    def _construct_clf(self, _conf: Optional[list[ModelInfo]]) -> Optional[artifact_t]:
         preproc_clfs = []
-        for _cfg in _conf.numalogic_conf.preprocess:
+        if not _conf:
+            return None
+        for _cfg in _conf:
             _clf = self._preproc_factory.get_instance(_cfg)
             preproc_clfs.append(_clf)
         if not preproc_clfs:
@@ -348,7 +356,6 @@ def get_feature_arr(
         raw_df: pd.DataFrame,
         metrics: list[str],
         fill_value: float = 0.0,
-        max_value_map: Optional[dict[str, float]] = None,
     ) -> tuple[npt.NDArray[float], float, float]:
         """
         Get feature array from the raw dataframe.
@@ -369,16 +376,7 @@ def get_feature_arr(
             if col not in raw_df.columns:
                 raw_df[col] = fill_value
                 nan_counter += len(raw_df)
-
         feat_df = raw_df[metrics]
-        if max_value_map:
-            max_value_list = [max_value_map.get(col, np.nan) for col in metrics]
-            feat_df.clip(upper=max_value_list, inplace=True)
-            _LOGGER.info(
-                "Replaced %s with max_value_map from the map with value of %s.",
-                metrics,
-                max_value_list,
-            )
         nan_counter += raw_df.isna().sum().all()
         inf_counter = np.isinf(feat_df).sum().all()
         feat_df = feat_df.fillna(fill_value).replace([np.inf, -np.inf], fill_value)