feat: first benchmarking using KPI anomaly data (#163)

Signed-off-by: Avik Basu <[email protected]>

Makefile

@@ -16,7 +16,7 @@ clean:
  @find . -type f -name "*.py[co]" -exec rm -rf {} +
 
 format: clean
- poetry run black numalogic/ examples/ tests/
+ poetry run black numalogic/ examples/ tests/ benchmarks/
 
 lint: format
  poetry run flake8 .

benchmarks/README.md

@@ -0,0 +1,10 @@
+## Benchmarks
+
+This section contains some benchmarking results of numalogic's algorithms on real as well
+synthetic data. Datasets here are publicly available from their respective repositories.
+
+Note that efforts have not really been made on hyperparameter tuning. This is just to give users an 
+idea on how each algorithm is suitable for different kinds of data, and shows how they can do
+their own benchmarking too.
+
+This is an ongoing process, and we will add more benchmarking results in the near future.

benchmarks/kpi/README.md

@@ -0,0 +1,29 @@
+## KPI Anomaly dataset
+
+KPI anomaly dataset consists of KPI (key performace index) time series data from 
+many real scenarios of Internet companies with ground truth label. 
+The dataset can be found (here)[https://github.com/NetManAIOps/KPI-Anomaly-Detection]
+
+The full dataset contains multiple KPI IDs. Different KPI time series have different structures
+and patterns. 
+For our purpose, we are running anomaly detection for some of these KPI indices.
+
+The performance table is shown below, although note that the hyperparameters have not been tuned.
+The hyperparams used are available inside the results directory under each algorithm.
+
+
+| KPI ID | KPI index | Algorithm | ROC-AUC |
+|--------------------------------------|-----------|---------------|---------|
+| 431a8542-c468-3988-a508-3afd06a218da | 14 | VanillaAE | 0.89 |
+| 431a8542-c468-3988-a508-3afd06a218da | 14 | Conv1dAE | 0.88 |
+| 431a8542-c468-3988-a508-3afd06a218da | 14 | LSTMAE | 0.86 |
+| 431a8542-c468-3988-a508-3afd06a218da | 14 | TransformerAE | 0.82 |
+
+
+Full credit to Zeyan Li et al. for constructing large-scale real world benchmark datasets for AIOps.
+
+@misc{2208.03938,
+Author = {Zeyan Li and Nengwen Zhao and Shenglin Zhang and Yongqian Sun and Pengfei Chen and Xidao Wen and Minghua Ma and Dan Pei},
+Title = {Constructing Large-Scale Real-World Benchmark Datasets for AIOps},
+Year = {2022},
+Eprint = {arXiv:2208.03938},

benchmarks/kpi/datamodule.py

@@ -0,0 +1,151 @@
+from typing import Optional, Sequence
+
+import numpy as np
+import numpy.typing as npt
+import pandas as pd
+from pytorch_lightning.utilities.types import EVAL_DATALOADERS
+from sklearn.pipeline import make_pipeline
+from torch.utils.data import DataLoader
+
+from numalogic.tools.data import TimeseriesDataModule, StreamingDataset
+
+
+class KPIDataModule(TimeseriesDataModule):
+ r"""
+ Data Module to help set up train, test and validation datasets for
+ KPI Anomaly detection. This data module splits a single dataset
+ into train, validation and test sets using a specified split ratio.
+
+ The dataset can be found in https://github.com/NetManAIOps/KPI-Anomaly-Detection
+ Details about the dataset can be found in https://arxiv.org/pdf/2208.03938.pdf
+
+ The dataset is expected to be in the format of:
+
+ |timestamp | value | label | KPI ID |
+ |-----------|--------|--------|--------|
+ |1476460800| 0.01260 | 0 |da10a69 |
+
+ Args:
+ data_dir: data directory where csv data files are stored
+ kpi_idx: index of the KPI to use
+ preproc_transforms: list of sklearn compatible preprocessing transformations
+ split_ratios: weights of train, validation and test sets respectively
+ *args, **kwargs: extra kwargs for TimeseriesDataModule
+ """
+
+ def __init__(
+ self,
+ data_dir: str,
+ kpi_idx: int,
+ preproc_transforms: Optional[list] = None,
+ split_ratios: Sequence[float] = (0.5, 0.2, 0.3),
+ *args,
+ **kwargs,
+ ):
+ super().__init__(data=None, *args, **kwargs)
+
+ if len(split_ratios) != 3 or sum(split_ratios) != 1.0:
+ raise ValueError("Sum of all the 3 ratios should be 1.0")
+
+ self.split_ratios = split_ratios
+ self.data_dir = data_dir
+ self.kpi_idx = kpi_idx
+ if preproc_transforms:
+ if len(preproc_transforms) > 1:
+ self.transforms = make_pipeline(preproc_transforms)
+ else:
+ self.transforms = preproc_transforms[0]
+ else:
+ self.transforms = None
+
+ self.train_dataset = None
+ self.val_dataset = None
+ self.test_dataset = None
+
+ self._train_labels = None
+ self._val_labels = None
+ self._test_labels = None
+
+ self.unique_kpis = None
+
+ self._kpi_df = self.get_kpi_df()
+
+ def _preprocess(self, df: pd.DataFrame) -> npt.NDArray[float]:
+ if self.transforms:
+ return self.transforms.fit_transform(df[["value"]].to_numpy())
+ return df[["value"]].to_numpy()
+
+ def setup(self, stage: str) -> None:
+ val_size = np.floor(self.split_ratios[1] * len(self._kpi_df)).astype(int)
+ test_size = np.floor(self.split_ratios[2] * len(self._kpi_df)).astype(int)
+
+ if stage == "fit":
+ train_df = self._kpi_df[: -(val_size + test_size)]
+ val_df = self._kpi_df[val_size:test_size]
+
+ self._train_labels = train_df["label"]
+ train_data = self._preprocess(train_df)
+ self.train_dataset = StreamingDataset(train_data, self.seq_len)
+
+ self._val_labels = val_df["label"]
+ val_data = self._preprocess(val_df)
+ self.val_dataset = StreamingDataset(val_data, self.seq_len)
+
+ print(f"Train size: {train_data.shape}\nVal size: {val_data.shape}")
+
+ if stage in ("test", "predict"):
+ test_df = self._kpi_df[-test_size:]
+ self._test_labels = test_df["label"]
+ test_data = self._preprocess(test_df)
+ self.test_dataset = StreamingDataset(test_data, self.seq_len)
+
+ print(f"Test size: {test_data.shape}")
+
+ @property
+ def val_data(self) -> npt.NDArray[float]:
+ return self.val_dataset.data
+
+ @property
+ def train_data(self) -> npt.NDArray[float]:
+ return self.train_dataset.data
+
+ @property
+ def test_data(self) -> npt.NDArray[float]:
+ return self.test_dataset.data
+
+ @property
+ def val_labels(self) -> npt.NDArray[int]:
+ return self._val_labels.to_numpy()
+
+ @property
+ def train_labels(self) -> npt.NDArray[int]:
+ return self._train_labels.to_numpy()
+
+ @property
+ def test_labels(self) -> npt.NDArray[int]:
+ return self._test_labels.to_numpy()
+
+ def get_kpi(self, idx: int) -> Optional[str]:
+ if self.unique_kpis is not None:
+ return self.unique_kpis[idx]
+ return None
+
+ def get_kpi_df(self) -> pd.DataFrame:
+ df = pd.read_csv(self.data_dir)
+ df["timestamp"] = pd.to_datetime(df["timestamp"], unit="s")
+ df.set_index(df["timestamp"], inplace=True)
+ df.drop(columns=["timestamp"], inplace=True)
+ self.unique_kpis = df["KPI ID"].unique()
+ grouped = df.groupby(["KPI ID", "timestamp"]).sum()
+ kpi_id = self.get_kpi(self.kpi_idx)
+ print(f"Using KPI ID: {kpi_id}")
+ return grouped.loc[kpi_id]
+
+ def val_dataloader(self) -> EVAL_DATALOADERS:
+ r"""
+ Creates and returns a DataLoader for the validation dataset if validation data is provided.
+ """
+ return DataLoader(self.val_dataset, batch_size=self.batch_size)
+
+ def predict_dataloader(self) -> EVAL_DATALOADERS:
+ return DataLoader(self.test_dataset, batch_size=self.batch_size)

benchmarks/kpi/results/kpi_idx_14/conv/hyperparams.json

@@ -0,0 +1,14 @@
+{
+ "BATCH_SIZE": 64,
+ "SPLIT_RATIOS": [0.5, 0.2, 0.3],
+ "TRAINER": {"accelerator": "cpu", "max_epochs": 30},
+ "MODEL": {
+ "name": "Conv1dAE",
+ "conf": {
+ "seq_len": 16,
+ "in_channels": 1,
+ "enc_channels": [4, 8, 16, 2],
+ "weight_decay": 1e-4
+ }
+ }
+}

benchmarks/kpi/results/kpi_idx_14/lstm/hyperparams.json

@@ -0,0 +1,16 @@
+{
+ "BATCH_SIZE": 64,
+ "SPLIT_RATIOS": [0.5, 0.2, 0.3],
+ "TRAINER": {"accelerator": "cpu", "max_epochs": 30},
+ "MODEL": {
+ "name": "LSTMAE",
+ "conf": {
+ "seq_len": 32,
+ "no_features": 1,
+ "embedding_dim": 4,
+ "encoder_layers": 2,
+ "decoder_layers": 2,
+ "weight_decay": 0.0001
+ }
+ }
+}

benchmarks/kpi/results/kpi_idx_14/transformer/hyperparams.json

@@ -0,0 +1,13 @@
+{
+ "BATCH_SIZE": 64,
+ "SPLIT_RATIOS": [0.5, 0.2, 0.3],
+ "TRAINER": {"accelerator": "cpu", "max_epochs": 30},
+ "MODEL": {
+ "name": "TransformerAE",
+ "conf": {
+ "seq_len": 16,
+ "n_features": 1,
+ "dim_feedforward": 128
+ }
+ }
+}

benchmarks/kpi/results/kpi_idx_14/vanilla/hyperparams.json

@@ -0,0 +1,18 @@
+{
+ "BATCH_SIZE": 64,
+ "SPLIT_RATIOS": [
+ 0.5,
+ 0.2,
+ 0.3
+ ],
+ "TRAINER": {
+ "accelerator": "cpu",
+ "max_epochs": 30
+ },
+ "MODEL": {
+ "name": "VanillaAE",
+ "conf": {
+ "seq_len": 10
+ }
+ }
+}

benchmarks/plots.py

@@ -0,0 +1,30 @@
+from matplotlib import pyplot as plt
+from sklearn.metrics import RocCurveDisplay
+
+
+def plot_reconerr_comparision(reconerr, input_, labels, start=0, end=None, title=None):
+ r"""
+ Plots the reconstruction error with respect to the input and output labels.
+ """
+ end = end or len(reconerr)
+ fig, ax = plt.subplots(3, 1, figsize=(12, 7))
+ ax[0].plot(reconerr[start:end], color="b", label="reconstruction error")
+ ax[0].legend(shadow=True)
+ ax[1].plot(input_[start:end], label="input data")
+ ax[1].legend(shadow=True)
+ ax[2].plot(labels[start:end], color="g", label="labels")
+ ax[2].legend(shadow=True)
+ if title:
+ ax[0].set_title(title)
+
+ return fig
+
+
+def plot_roc_curve(y_true, y_pred, model_name, title=None):
+ _ = RocCurveDisplay.from_predictions(y_true, y_pred, name=model_name)
+ plt.plot([0, 1], [0, 1], "k--", label="Baseline (AUC = 0.5)")
+ plt.xlabel("False Positive Rate")
+ plt.ylabel("True Positive Rate")
+ if title:
+ plt.title(title)
+ plt.legend()

numalogic/config/_config.py

@@ -51,6 +51,7 @@ class LightningTrainerConf:
  https://pytorch-lightning.readthedocs.io/en/stable/common/trainer.html#trainer-class-api
  """
 
+ accelerator: str = "auto"
  max_epochs: int = 100
  logger: bool = False
  check_val_every_n_epoch: int = 5

numalogic/tools/data.py

@@ -53,6 +53,13 @@ def __init__(self, data: npt.NDArray[float], seq_len: int):
  self._seq_len = seq_len
  self._data = data.astype(np.float32)
 
+ @property
+ def data(self) -> npt.NDArray[float]:
+ """
+ Returns the reference data in the input shape
+ """
+ return self._data
+
  def create_seq(self, input_: npt.NDArray[float]) -> Generator[npt.NDArray[float], None, None]:
  r"""
  A generator function that yields sequences of specified length from the input data.
@@ -132,6 +139,8 @@ def setup(self, stage: str) -> None:
  """
  if stage == "fit":
  val_size = np.floor(self.val_split_ratio * len(self.data)).astype(int)
+ _LOGGER.info("Size of validation set: %s", val_size)
+
  self.train_dataset = StreamingDataset(self.data[:-val_size, :], self.seq_len)
  self.val_dataset = StreamingDataset(self.data[-val_size:, :], self.seq_len)
 

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "numalogic"
-version = "0.3.8"
+version = "0.4.dev0"
 description = "Collection of operational Machine Learning models and tools."
 authors = ["Numalogic Developers"]
 packages = [{ include = "numalogic" }]

tests/tools/test_data.py

@@ -34,6 +34,7 @@ def test_dataset(self):
  self.assertTupleEqual((SEQ_LEN, self.n), seq.shape)
  self.assertEqual(self.data.shape[0] - SEQ_LEN + 1, len(dataset))
  assert_allclose(np.ravel(dataset[0]), np.ravel(self.data[:12, :]))
+ assert_allclose(self.data, dataset.data)
 
  def test_w_dataloader(self):
  batch_size = 4