feat: initial support for flattened vector in backtest (#361)

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

numalogic/backtest/__init__.py

@@ -1,5 +1,5 @@
 from importlib.util import find_spec
-from numalogic.backtest._prom import PromBacktester
+from numalogic.backtest._prom import PromBacktester, OutDataFrames
 
 
 def _validate_req_pkgs():
@@ -12,4 +12,4 @@ def _validate_req_pkgs():
 _validate_req_pkgs()
 
 
-__all__ = ["PromBacktester"]
+__all__ = ["PromBacktester", "OutDataFrames"]

numalogic/backtest/_prom.py

@@ -11,6 +11,7 @@
 
 import logging
 import os.path
+from dataclasses import dataclass
 from datetime import datetime, timedelta
 from pathlib import Path
 from typing import Optional, Union
@@ -40,6 +41,16 @@
 LOGGER = logging.getLogger(__name__)
 
 
+@dataclass
+class OutDataFrames:
+ input: pd.DataFrame
+ preproc_out: pd.DataFrame
+ model_out: pd.DataFrame
+ thresh_out: pd.DataFrame
+ postproc_out: pd.DataFrame
+ unified_out: pd.DataFrame
+
+
 class PromBacktester:
  def __init__(
  self,
@@ -71,6 +82,7 @@ def __init__(
  self.metrics = metrics or []
  self.conf: StreamConf = self._init_conf(metrics, numalogic_cfg, load_saved_conf)
  self.nlconf: NumalogicConf = self.conf.get_numalogic_conf()
+ self.seq_len = self.conf.window_size
 
  def _init_conf(self, metrics: list[str], nl_conf: dict, load_saved_conf: bool) -> StreamConf:
  if load_saved_conf:
@@ -144,12 +156,44 @@ def train_models(
  LOGGER.info("Models saved in %s", self._modelpath)
  return artifacts_dict
 
+ def window_inverse(self, x: NDArray[float]) -> NDArray[float]:
+ """
+ Perform inverse windowing on the given data.
+
+ If stride is 1, return the data as is.
+ If stride is 2, return the data by stacking the two windows.
+ Stride > 2 is not supported yet.
+
+ Args:
+ -------
+ x: Input data
+
+ Returns
+ -------
+ Inverse windowed data with feature recovery if stride is 2
+ """
+ x = torch.from_numpy(x)
+ stride = self.nlconf.trainer.ds_stride
+
+ if stride == 1:
+ return inverse_window(x).numpy()
+
+ # TODO support for stride > 2
+ if stride > 2:
+ raise NotImplementedError("Stride > 2 not supported!")
+
+ # Recover the features
+ x1, x2 = x[:, ::stride], x[:, 1::stride]
+ x1 = inverse_window(x1).numpy()
+ x2 = inverse_window(x2).numpy()
+ return np.hstack([x1, x2])
+
  def generate_scores(
  self,
  df: pd.DataFrame,
  model_path: Optional[str] = None,
  use_full_data: bool = False,
- ) -> pd.DataFrame:
+ ) -> OutDataFrames:
  """
  Generate scores for the given data.
 
@@ -161,7 +205,7 @@ def generate_scores(
 
  Returns
  -------
- Dataframe with timestamp and metric values
+ Dict of dataframes with timestamp and metric values for each step
 
  Raises
  ------
@@ -188,11 +232,13 @@ def generate_scores(
  # Preprocess
  x_scaled = preproc_udf.compute(model=artifacts["preproc_clf"], input_=x_test)
 
- ds = StreamingDataset(x_scaled, seq_len=self.conf.window_size)
+ n_feat = x_scaled.shape[1]
 
- x_recon = np.zeros((len(ds), self.conf.window_size, len(self.metrics)), dtype=np.float32)
- raw_scores = np.zeros((len(ds), self.conf.window_size, len(self.metrics)), dtype=np.float32)
- feature_scores = np.zeros((len(ds), len(self.metrics)), dtype=np.float32)
+ ds = StreamingDataset(x_scaled, seq_len=self.seq_len, stride=self.nlconf.trainer.ds_stride)
+
+ x_recon = np.zeros((len(ds), self.seq_len, n_feat), dtype=np.float32)
+ raw_scores = np.zeros((len(ds), self.seq_len, n_feat), dtype=np.float32)
+ feature_scores = np.zeros((len(ds), n_feat), dtype=np.float32)
  unified_scores = np.zeros((len(ds), 1), dtype=np.float32)
 
  postproc_func = PostprocessFactory().get_instance(self.nlconf.postprocess)
@@ -215,16 +261,17 @@ def generate_scores(
  feature_scores[idx], self.nlconf.score.feature_agg
  )
 
- x_recon = inverse_window(torch.from_numpy(x_recon), method="keep_first").numpy()
- raw_scores = inverse_window(torch.from_numpy(raw_scores), method="keep_first").numpy()
+ x_recon = self.window_inverse(x_recon)
+ raw_scores = self.window_inverse(raw_scores)
+
  feature_scores = np.vstack(
  [
- np.full((self.conf.window_size - 1, len(self.metrics)), fill_value=np.nan),
+ np.full((len(x_test) - len(ds), n_feat), fill_value=np.nan),
  feature_scores,
  ]
  )
  unified_scores = np.vstack(
- [np.full((self.conf.window_size - 1, 1), fill_value=np.nan), unified_scores]
+ [np.full((len(x_test) - len(ds), 1), fill_value=np.nan), unified_scores]
  )
 
  return self._construct_output(
@@ -244,7 +291,7 @@ def generate_static_scores(self, df: pd.DataFrame) -> pd.DataFrame:
  x_test = df[metrics].to_numpy(dtype=np.float32)
 
  postproc_udf = UDFFactory.get_udf_cls("postprocess")
- ds = StreamingDataset(x_test, seq_len=self.conf.window_size)
+ ds = StreamingDataset(x_test, seq_len=self.seq_len)
 
  feature_scores = np.zeros((len(ds), len(metrics)), dtype=np.float32)
  unified_scores = np.zeros((len(ds), 1), dtype=np.float32)
@@ -258,12 +305,12 @@ def generate_static_scores(self, df: pd.DataFrame) -> pd.DataFrame:
  )
  feature_scores = np.vstack(
  [
- np.full((self.conf.window_size - 1, len(metrics)), fill_value=np.nan),
+ np.full((self.seq_len - 1, len(metrics)), fill_value=np.nan),
  feature_scores,
  ]
  )
  unified_scores = np.vstack(
- [np.full((self.conf.window_size - 1, 1), fill_value=np.nan), unified_scores]
+ [np.full((self.seq_len - 1, 1), fill_value=np.nan), unified_scores]
  )
  dfs = {
  "input": df,
@@ -345,9 +392,13 @@ def _construct_output(
  thresh_out: NDArray[float],
  postproc_out: NDArray[float],
  unified_out: NDArray[float],
- ) -> pd.DataFrame:
+ ) -> OutDataFrames:
  ts_idx = input_df.index
 
+ print(
+ preproc_out.shape, nn_out.shape, thresh_out.shape, postproc_out.shape, unified_out.shape
+ )
+
  if thresh_out.shape[1] > 1:
  thresh_df = pd.DataFrame(
  thresh_out,
@@ -364,7 +415,6 @@ def _construct_output(
  if postproc_out.shape[1] > 1:
  postproc_df = pd.DataFrame(
  postproc_out,
- # columns=["unified_score"],
  columns=self.metrics,
  index=ts_idx,
  )
@@ -375,24 +425,30 @@ def _construct_output(
  index=ts_idx,
  )
 
- dfs = {
- "input": input_df,
- "preproc_out": pd.DataFrame(
+ if len(preproc_out) == len(ts_idx) and preproc_out.shape[1] == len(self.metrics):
+ preproc_df = pd.DataFrame(
  preproc_out,
  columns=self.metrics,
  index=ts_idx,
- ),
- "model_out": pd.DataFrame(
+ )
+ else:
+ preproc_df = pd.DataFrame(
+ preproc_out,
+ )
+
+ return OutDataFrames(
+ input=input_df,
+ preproc_out=preproc_df,
+ model_out=pd.DataFrame(
  nn_out,
  columns=self.metrics,
  index=ts_idx,
  ),
- "thresh_out": thresh_df,
- "postproc_out": postproc_df,
- "unified_out": pd.DataFrame(
+ thresh_out=thresh_df,
+ postproc_out=postproc_df,
+ unified_out=pd.DataFrame(
  unified_out,
  columns=["unified"],
  index=ts_idx,
  ),
- }
- return pd.concat(dfs, axis=1)
+ )

numalogic/tools/data.py

@@ -127,6 +127,9 @@ def __init__(self, data: npt.NDArray[float], seq_len: int, stride: int = 1):
  self._seq_len = seq_len
  self._data = data.astype(np.float32)
  self._stride = stride
+ _LOGGER.info(
+ "StreamingDataset initialized with seq_len: %s, stride: %s", self._seq_len, self._stride
+ )
 
  @property
  def data(self) -> npt.NDArray[float]:

tests/test_backtest.py

@@ -5,7 +5,7 @@
 from omegaconf import OmegaConf
 
 from numalogic._constants import TESTS_DIR
-from numalogic.backtest import PromBacktester
+from numalogic.backtest import PromBacktester, OutDataFrames
 from numalogic.config import (
  NumalogicConf,
  ModelInfo,
@@ -64,9 +64,11 @@ def test_train(backtester, read_data):
 
 
 def test_scores(backtester, read_data):
- out_df = backtester.generate_scores(read_data)
- assert isinstance(out_df, pd.DataFrame)
- assert out_df.shape[0] == int(backtester.test_ratio * read_data.shape[0])
+ out_dfs = backtester.generate_scores(read_data)
+ assert isinstance(out_dfs, OutDataFrames)
+ assert out_dfs.input.shape[0] == int(backtester.test_ratio * read_data.shape[0])
+ assert out_dfs.postproc_out.shape[0] == int(backtester.test_ratio * read_data.shape[0])
+ assert out_dfs.unified_out.shape[0] == int(backtester.test_ratio * read_data.shape[0])
 
 
 def test_static_scores(backtester, read_data):