Fix exclude_detection_period_from_training for large time buckets

integration_tests/tests/test_column_anomalies.py

@@ -578,3 +578,112 @@ def test_col_anom_excl_detect_train(test_id: str, dbt_project: DbtProject):
         "Expected FAIL when exclude_detection_period_from_training=True "
         "(detection data excluded from training baseline, anomaly detected)"
     )
+
+
+@pytest.mark.skip_targets(["clickhouse"])
+def test_col_excl_detect_train_monthly(test_id: str, dbt_project: DbtProject):
+    """
+    Test exclude_detection_period_from_training with monthly time buckets for column anomalies.
+
+    This tests the fix where the detection period is set to the bucket size
+    when the bucket period exceeds backfill_days. With monthly buckets (30 days)
+    and default backfill_days (2), without the fix the 2-day exclusion window
+    cannot contain any monthly bucket_end, making exclusion ineffective.
+
+    detection_period is intentionally NOT set so that backfill_days stays at
+    its default (2), which is smaller than the monthly bucket (30 days).
+    Setting detection_period would override backfill_days and mask the bug.
+
+    Scenario:
+    - 12 months of normal data with low null count (~10 nulls/day, ~300/month)
+    - 1 month of anomalous data with high null count (25 nulls/day, ~775/month)
+    - time_bucket: month (30 days >> default backfill_days of 2)
+    - Without exclusion: anomaly absorbed into training → test passes
+    - With exclusion + fix: anomaly excluded from training → test fails
+    """
+    utc_now = datetime.utcnow().date()
+    current_month_1st = utc_now.replace(day=1)
+
+    anomaly_month_start = (current_month_1st - timedelta(days=31)).replace(day=1)
+    normal_month_start = (anomaly_month_start - timedelta(days=365)).replace(day=1)
+
+    normal_data: List[Dict[str, Any]] = []
+    day = normal_month_start
+    day_idx = 0
+    while day < anomaly_month_start:
+        null_count = 7 + (day_idx % 7)
+        normal_data.extend(
+            [
+                {TIMESTAMP_COLUMN: day.strftime(DATE_FORMAT), "superhero": superhero}
+                for superhero in ["Superman", "Batman", "Wonder Woman", "Flash"] * 10
+            ]
+        )
+        normal_data.extend(
+            [
+                {TIMESTAMP_COLUMN: day.strftime(DATE_FORMAT), "superhero": None}
+                for _ in range(null_count)
+            ]
+        )
+        day += timedelta(days=1)
+        day_idx += 1
+
+    anomalous_data: List[Dict[str, Any]] = []
+    day = anomaly_month_start
+    while day < utc_now:
+        anomalous_data.extend(
+            [
+                {TIMESTAMP_COLUMN: day.strftime(DATE_FORMAT), "superhero": superhero}
+                for superhero in ["Superman", "Batman", "Wonder Woman", "Flash"] * 10
+            ]
+        )
+        anomalous_data.extend(
+            [
+                {TIMESTAMP_COLUMN: day.strftime(DATE_FORMAT), "superhero": None}
+                for _ in range(25)
+            ]
+        )
+        day += timedelta(days=1)
+
+    all_data = normal_data + anomalous_data
+
+    test_args_without_exclusion = {
+        "timestamp_column": TIMESTAMP_COLUMN,
+        "column_anomalies": ["null_count"],
+        "time_bucket": {"period": "month", "count": 1},
+        "training_period": {"period": "day", "count": 365},
+        "min_training_set_size": 5,
+        "anomaly_sensitivity": 4,
+        "anomaly_direction": "spike",
+        "exclude_detection_period_from_training": False,
+    }
+
+    test_result_without = dbt_project.test(
+        test_id + "_f",
+        DBT_TEST_NAME,
+        test_args_without_exclusion,
+        data=all_data,
+        test_column="superhero",
+        test_vars={"force_metrics_backfill": True},
+    )
+    assert test_result_without["status"] == "pass", (
+        "Expected PASS when exclude_detection_period_from_training=False "
+        "(detection data included in training baseline)"
+    )
+
+    test_args_with_exclusion = {
+        **test_args_without_exclusion,
+        "exclude_detection_period_from_training": True,
+    }
+
+    test_result_with = dbt_project.test(
+        test_id + "_t",
+        DBT_TEST_NAME,
+        test_args_with_exclusion,
+        data=all_data,
+        test_column="superhero",
+        test_vars={"force_metrics_backfill": True},
+    )
+    assert test_result_with["status"] == "fail", (
+        "Expected FAIL when exclude_detection_period_from_training=True "
+        "(large bucket fix: detection period set to bucket size)"
+    )

integration_tests/tests/test_volume_anomalies.py

@@ -619,3 +619,88 @@ def test_exclude_detection_from_training(test_id: str, dbt_project: DbtProject):
     assert (
         test_result_with_exclusion["status"] == "fail"
     ), "Test should fail when anomaly is excluded from training"
+
+
+@pytest.mark.skip_targets(["clickhouse"])
+def test_excl_detect_train_monthly(test_id: str, dbt_project: DbtProject):
+    """
+    Test exclude_detection_period_from_training with monthly time buckets.
+
+    This tests the fix where the detection period is set to the bucket size
+    when the bucket period exceeds backfill_days. With monthly buckets (30 days)
+    and default backfill_days (2), without the fix the 2-day exclusion window
+    cannot contain any monthly bucket_end, making exclusion ineffective.
+
+    detection_period is intentionally NOT set so that backfill_days stays at
+    its default (2), which is smaller than the monthly bucket (30 days).
+    Setting detection_period would override backfill_days and mask the bug.
+
+    Scenario:
+    - 12 months of normal data (~20 rows/day, ~600/month)
+    - 1 month of anomalous data (~40 rows/day, ~1240/month)
+    - time_bucket: month (30 days >> default backfill_days of 2)
+    - Without exclusion: anomaly absorbed into training → test passes
+    - With exclusion + fix: anomaly excluded from training → test fails
+    """
+    utc_now = datetime.utcnow()
+    current_month_1st = utc_now.replace(
+        day=1, hour=0, minute=0, second=0, microsecond=0
+    )
+
+    anomaly_month_start = (current_month_1st - timedelta(days=31)).replace(day=1)
+    normal_month_start = (anomaly_month_start - timedelta(days=365)).replace(day=1)
+
+    normal_data = []
+    day = normal_month_start
+    day_idx = 0
+    while day < anomaly_month_start:
+        rows_per_day = 17 + (day_idx % 7)
+        normal_data.extend(
+            [{TIMESTAMP_COLUMN: day.strftime(DATE_FORMAT)} for _ in range(rows_per_day)]
+        )
+        day += timedelta(days=1)
+        day_idx += 1
+
+    anomalous_data = []
+    day = anomaly_month_start
+    while day < utc_now:
+        anomalous_data.extend(
+            [{TIMESTAMP_COLUMN: day.strftime(DATE_FORMAT)} for _ in range(40)]
+        )
+        day += timedelta(days=1)
+
+    all_data = normal_data + anomalous_data
+
+    test_args_without_exclusion = {
+        **DBT_TEST_ARGS,
+        "training_period": {"period": "day", "count": 365},
+        "time_bucket": {"period": "month", "count": 1},
+        "sensitivity": 4,
+    }
+
+    test_result_without = dbt_project.test(
+        test_id + "_without",
+        DBT_TEST_NAME,
+        test_args_without_exclusion,
+        data=all_data,
+        test_vars={"force_metrics_backfill": True},
+    )
+    assert (
+        test_result_without["status"] == "pass"
+    ), "Test should pass when anomaly is included in training"
+
+    test_args_with_exclusion = {
+        **test_args_without_exclusion,
+        "exclude_detection_period_from_training": True,
+    }
+
+    test_result_with = dbt_project.test(
+        test_id + "_with",
+        DBT_TEST_NAME,
+        test_args_with_exclusion,
+        data=all_data,
+        test_vars={"force_metrics_backfill": True},
+    )
+    assert (
+        test_result_with["status"] == "fail"
+    ), "Test should fail when anomaly is excluded from training (large bucket fix)"

macros/edr/data_monitoring/anomaly_detection/get_anomaly_scores_query.sql

@@ -53,11 +53,24 @@
     {# Calculate detection period start for exclusion logic.
        backfill_days defines the window of recent data to test for anomalies on each run.
        It defaults to 2 days (configurable via vars.backfill_days or test-level parameter).
-       The detection period spans from (detection_end - backfill_days) to detection_end.
+       The detection period spans from (detection_end - exclusion_period_days) to detection_end.
        When exclude_detection_period_from_training is enabled, metrics in this detection period
-       are excluded from training statistics to prevent contamination from potentially anomalous data. #}
+       are excluded from training statistics to prevent contamination from potentially anomalous data.
+
+       The exclusion window must be at least one full time bucket to work correctly.
+       When backfill_days is smaller than the time bucket period (e.g., backfill_days=2
+       with weekly buckets), the detection period would be too narrow to contain any
+       bucket_end, making the exclusion ineffective. We extend it to at least one full
+       time bucket in that case. #}
     {%- if test_configuration.exclude_detection_period_from_training %}
-        {%- set detection_period_start = (detection_end - modules.datetime.timedelta(days=test_configuration.backfill_days)) %}
+        {%- set exclusion_period_days = test_configuration.backfill_days %}
+        {%- if metric_properties and metric_properties.time_bucket %}
+            {%- set bucket_in_days = elementary.convert_period(metric_properties.time_bucket, 'day').count %}
+            {%- if bucket_in_days > exclusion_period_days %}
+                {%- set exclusion_period_days = bucket_in_days %}
+            {%- endif %}
+        {%- endif %}
+        {%- set detection_period_start = (detection_end - modules.datetime.timedelta(days=exclusion_period_days)) %}
         {%- set detection_period_start_expr = elementary.edr_cast_as_timestamp(elementary.edr_datetime_to_sql(detection_period_start)) %}
     {%- endif %}