fix(monitors): constrain volume lower bound for cumulative tables

Author: aarthy-dk

testgen/commands/test_thresholds_prediction.py

@@ -110,13 +110,18 @@ def run(self) -> None:
                     )
                     test_prediction.extend([lower, upper, staleness, prediction])
                 else:
+                    functional_table_type = group["functional_table_type"].iloc[0]
+                    is_cumulative = bool(
+                        functional_table_type and str(functional_table_type).startswith("cumulative")
+                    )
                     lower, upper, prediction = compute_sarimax_threshold(
                         history,
                         sensitivity=self.test_suite.predict_sensitivity or PredictSensitivity.medium,
                         min_lookback=self.test_suite.predict_min_lookback or 1,
                         exclude_weekends=self.test_suite.predict_exclude_weekends,
                         holiday_codes=self.test_suite.holiday_codes_list,
                         schedule_tz=self.tz,
+                        is_cumulative=is_cumulative,
                     )
                     test_prediction.extend([lower, upper, None, prediction])
 
@@ -258,10 +263,13 @@ def compute_sarimax_threshold(
     exclude_weekends: bool = False,
     holiday_codes: list[str] | None = None,
     schedule_tz: str | None = None,
+    is_cumulative: bool = False,
 ) -> tuple[float | None, float | None, str | None]:
     """Compute SARIMAX-based thresholds for the next forecast point.
 
     Returns (lower, upper, forecast_json) or (None, None, None) if insufficient data.
+    For cumulative tables, the lower tolerance is floored at the last observed value
+    so that any decrease in row count is detected as an anomaly.
     """
     if len(history) < min_lookback:
         return None, None, None
@@ -291,7 +299,12 @@ def compute_sarimax_threshold(
 
         if pd.isna(lower_tolerance) or pd.isna(upper_tolerance):
             return None, None, None
-        else:
-            return float(lower_tolerance), float(upper_tolerance), forecast.to_json()
+
+        lower_tolerance = float(lower_tolerance)
+        if is_cumulative:
+            last_observed = float(history["result_signal"].iloc[-1])
+            lower_tolerance = max(lower_tolerance, last_observed)
+
+        return lower_tolerance, float(upper_tolerance), forecast.to_json()
     except NotEnoughData:
         return None, None, None

testgen/template/prediction/get_historical_test_results.sql

@@ -2,6 +2,7 @@ WITH filtered_defs AS (
   -- Filter definitions first to minimize join surface area
   SELECT id,
     test_suite_id,
+    table_groups_id,
     schema_name,
     table_name,
     column_name,
@@ -17,8 +18,13 @@ SELECT r.test_definition_id,
   CASE
     WHEN r.result_signal ~ '^-?[0-9]*\.?[0-9]+$' THEN r.result_signal::NUMERIC
     ELSE NULL
-  END AS result_signal
+  END AS result_signal,
+  dtc.functional_table_type
 FROM test_results r
 JOIN filtered_defs d ON d.id = r.test_definition_id
+LEFT JOIN data_table_chars dtc
+  ON dtc.table_groups_id = d.table_groups_id
+  AND dtc.schema_name = d.schema_name
+  AND dtc.table_name = d.table_name
 WHERE r.test_suite_id = :TEST_SUITE_ID
 ORDER BY r.test_time;

tests/unit/common/test_time_series_service.py

@@ -4,7 +4,7 @@
 import pandas as pd
 import pytest
 
-from testgen.commands.test_thresholds_prediction import compute_freshness_threshold
+from testgen.commands.test_thresholds_prediction import compute_freshness_threshold, compute_sarimax_threshold
 from testgen.common.freshness_service import (
     MIN_FRESHNESS_GAPS,
     FreshnessThreshold,
@@ -634,3 +634,77 @@ def test_without_exclusions_timezone_has_no_effect(self):
         forecast_with_tz = get_sarimax_forecast(history, num_forecast=3, exclude_weekends=False, tz="America/New_York")
 
         pd.testing.assert_frame_equal(forecast_no_tz, forecast_with_tz)
+
+
+class Test_ComputeSarimaxThreshold_CumulativeFloor:
+    """Tests for the cumulative table floor constraint in compute_sarimax_threshold."""
+
+    @staticmethod
+    def _make_monotonic_history(n_days: int = 30, start_value: int = 1000, daily_growth: int = 100) -> pd.DataFrame:
+        """Create a monotonically increasing row count history (cumulative table)."""
+        dates = pd.date_range("2026-01-01", periods=n_days, freq="1D")
+        values = [start_value + i * daily_growth for i in range(n_days)]
+        return pd.DataFrame({"result_signal": values}, index=dates)
+
+    def test_cumulative_floors_lower_at_last_observed(self):
+        history = self._make_monotonic_history(n_days=30, start_value=1000, daily_growth=100)
+        last_observed = float(history["result_signal"].iloc[-1])
+
+        lower, upper, prediction = compute_sarimax_threshold(
+            history, PredictSensitivity.medium, is_cumulative=True,
+        )
+
+        assert lower is not None
+        assert upper is not None
+        assert prediction is not None
+        assert lower >= last_observed
+
+    def test_non_cumulative_allows_lower_below_last_observed(self):
+        # With high variance, SARIMAX lower bound can drop below last observed
+        rng = np.random.default_rng(42)
+        dates = pd.date_range("2026-01-01", periods=30, freq="1D")
+        # Trending up but with large noise — lower bound should be below last value
+        values = [1000 + i * 50 + rng.normal(0, 200) for i in range(30)]
+        history = pd.DataFrame({"result_signal": values}, index=dates)
+        last_observed = float(history["result_signal"].iloc[-1])
+
+        lower, upper, prediction = compute_sarimax_threshold(
+            history, PredictSensitivity.low, is_cumulative=False,
+        )
+
+        assert lower is not None
+        # With low sensitivity (z=-3.0) and high noise, lower should be below last value
+        # This is the behavior we're protecting against with the cumulative floor
+        assert lower < last_observed
+
+    def test_cumulative_does_not_affect_upper_tolerance(self):
+        history = self._make_monotonic_history(n_days=30)
+
+        _, upper_cumulative, _ = compute_sarimax_threshold(
+            history, PredictSensitivity.medium, is_cumulative=True,
+        )
+        _, upper_normal, _ = compute_sarimax_threshold(
+            history, PredictSensitivity.medium, is_cumulative=False,
+        )
+
+        assert upper_cumulative == upper_normal
+
+    def test_cumulative_with_insufficient_data_returns_none(self):
+        history = self._make_monotonic_history(n_days=2)
+
+        lower, upper, prediction = compute_sarimax_threshold(
+            history, PredictSensitivity.medium, min_lookback=5, is_cumulative=True,
+        )
+
+        assert lower is None
+        assert upper is None
+        assert prediction is None
+
+    def test_cumulative_default_is_false(self):
+        history = self._make_monotonic_history(n_days=30)
+
+        # Without is_cumulative param, should behave as non-cumulative
+        lower_default, _, _ = compute_sarimax_threshold(history, PredictSensitivity.medium)
+        lower_explicit, _, _ = compute_sarimax_threshold(history, PredictSensitivity.medium, is_cumulative=False)
+
+        assert lower_default == lower_explicit