Merge pull request #114 from andrewm4894/clean-up-change-alert-snooze-logic

Clean up change alert snooze logic

Author: andrewm4894

Makefile

@@ -16,17 +16,17 @@ SHELL=/bin/bash
 dashboard:
 	streamlit run ./dashboard.py --server.port 8501
 
-# start streamlit dashboard as a daemon
+# start streamlit dashboard as a daemon with no log file
 dashboardd:
-	nohup streamlit run ./dashboard.py --server.port 8501 &
+	nohup streamlit run ./dashboard.py --server.port 8501 > /dev/null 2>&1 &
 
 # start dagster locally
 local:
 	dagster dev -f anomstack/main.py
 
-# start dagster locally as a daemon
+# start dagster locally as a daemon with no log file
 locald:
-	nohup dagster dev -f anomstack/main.py &
+	nohup dagster dev -f anomstack/main.py > /dev/null 2>&1 &
 
 # kill any running dagster process
 kill-locald:

anomstack/df/utils.py

@@ -0,0 +1,18 @@
+from io import StringIO
+from dagster import get_dagster_logger
+import pandas as pd
+
+
+def log_df_info(df: pd.DataFrame, logger=None):
+    """
+    Logs the info of a DataFrame to the logger.
+
+    Parameters:
+    df (pd.DataFrame): The DataFrame whose info needs to be logged.
+    """
+    if not logger:
+        logger = get_dagster_logger()
+    buffer = StringIO()
+    df.info(buf=buffer)
+    info_str = buffer.getvalue()
+    logger.info("df.info():\n%s", info_str)

anomstack/external/duckdb/duckdb.py

@@ -28,10 +28,7 @@ def read_sql_duckdb(sql: str) -> pd.DataFrame:
     os.makedirs(os.path.dirname(duckdb_path), exist_ok=True)
 
     conn = connect(duckdb_path)
-
-    logger.debug(f"sql:\n{sql}")
     df = query(connection=conn, query=sql).df()
-    logger.debug(f"df:\n{df}")
 
     return df
 

anomstack/external/gcp/bigquery.py

@@ -25,18 +25,11 @@ def read_sql_bigquery(sql: str) -> pd.DataFrame:
     Returns:
         pd.DataFrame: The result of the query as a DataFrame.
     """
-
-    logger = get_dagster_logger()
-
-    logger.debug(f"sql:\n{sql}")
-
     credentials = get_google_credentials()
-
     df = pd.read_gbq(
         query=sql,
         credentials=credentials,
     )
-    logger.debug(f"df:\n{df}")
 
     return df
 

anomstack/external/snowflake/snowflake.py

@@ -3,7 +3,6 @@
 """
 
 import pandas as pd
-from dagster import get_dagster_logger
 from snowflake import connector
 
 from anomstack.external.snowflake.credentials import get_snowflake_credentials
@@ -21,11 +20,6 @@ def read_sql_snowflake(sql: str, cols_lowercase: bool = True) -> pd.DataFrame:
     Returns:
         pd.DataFrame: The result of the SQL query as a pandas DataFrame.
     """
-
-    logger = get_dagster_logger()
-
-    logger.debug(f"sql:\n{sql}")
-
     credentials = get_snowflake_credentials()
 
     conn = connector.connect(
@@ -41,8 +35,6 @@ def read_sql_snowflake(sql: str, cols_lowercase: bool = True) -> pd.DataFrame:
     if cols_lowercase:
         df.columns = df.columns.str.lower()
 
-    logger.debug(f"df:\n{df}")
-
     return df
 
 

anomstack/external/sqlite/sqlite.py

@@ -29,9 +29,7 @@ def read_sql_sqlite(sql: str) -> pd.DataFrame:
 
     conn = sqlite3.connect(sqlite_path)
 
-    logger.debug(f"sql:\n{sql}")
     df = pd.read_sql_query(sql, conn)
-    logger.debug(f"df:\n{df}")
 
     conn.close()
     return df

anomstack/jobs/change.py

@@ -24,7 +24,9 @@
 from anomstack.sql.read import read_sql
 from anomstack.validate.validate import validate_df
 
-ANOMSTACK_MAX_RUNTIME_SECONDS_TAG = os.getenv("ANOMSTACK_MAX_RUNTIME_SECONDS_TAG", 3600)
+ANOMSTACK_MAX_RUNTIME_SECONDS_TAG = os.getenv(
+    "ANOMSTACK_MAX_RUNTIME_SECONDS_TAG", 3600
+)
 
 
 def build_change_job(spec: dict) -> JobDefinition:
@@ -62,7 +64,6 @@ def noop():
     metric_tags = spec.get("metric_tags", {})
     change_threshold = spec.get("change_threshold", 3.5)
     change_detect_last_n = spec.get("change_detect_last_n", 1)
-    change_snooze_n = spec.get("change_snooze_n", 3)
 
     @job(
         name=f"{metric_batch}_change",

anomstack/ml/change.py

@@ -12,8 +12,7 @@
 def detect_change(
     df_metric: pd.DataFrame,
     threshold: float = 3.5,
-    detect_last_n: int = 1,
-    snooze_n: int = 3
+    detect_last_n: int = 1
 ) -> pd.DataFrame:
     """
     Detects change in a metric based on the Median Absolute Deviation (MAD)
@@ -25,8 +24,6 @@ def detect_change(
             Defaults to 3.5.
         detect_last_n (int, optional): Number of last observations to use for
             detection. Defaults to 1.
-        snooze_n (int, optional): Number of observations to snooze after a
-            change is detected. Defaults to 3.
 
     Returns:
         pd.DataFrame: DataFrame with the detected change information.
@@ -44,24 +41,19 @@ def detect_change(
     y_detect_scores = detector.decision_function(X_detect)
     logger.debug(f"y_detect_scores: {y_detect_scores}")
     df_metric["metric_score"] = list(X_train_scores) + list(y_detect_scores)
-    df_metric["metric_alert"] = np.where((df_metric["metric_score"] > threshold),1,0)
+    df_metric["metric_alert"] = np.where(
+        df_metric["metric_score"] > threshold, 1, 0
+    )
     logger.debug(f"df_metric:\n{df_metric}")
     change_detected_count = df_metric["metric_alert"].tail(detect_last_n).sum()
-    recent_change_detected_count = df_metric["metric_alert"].tail(snooze_n).sum()
     if change_detected_count > 0:
-        # TODO: clean up the logic here as this could stay
-        # snoozed for a long time until we see sufficient 0's
-        # in last snooze_n so its more like a dyanmic suppression
-        if recent_change_detected_count <= 1:
-            logger.info(f"change detected for {metric_name} at {X_detect_timestamps}")
-            return df_metric
-        else:
-            logger.info(
-                f"change detected for {metric_name} at {X_detect_timestamps}, "
-                f"but snoozing as {recent_change_detected_count} recent changes already detected"
-            )
-            return pd.DataFrame()
+        logger.info(
+            f"change detected for {metric_name} at {X_detect_timestamps}"
+        )
+        return df_metric
     else:
-        logger.info(f"no change detected for {metric_name} at {X_detect_timestamps}")
+        logger.info(
+            f"no change detected for {metric_name} at {X_detect_timestamps}"
+        )
 
         return pd.DataFrame()

anomstack/sql/read.py

@@ -9,11 +9,14 @@
 import sqlglot
 from dagster import get_dagster_logger
 
+from anomstack.df.utils import log_df_info
 from anomstack.external.duckdb.duckdb import read_sql_duckdb
 from anomstack.external.gcp.bigquery import read_sql_bigquery
 from anomstack.external.snowflake.snowflake import read_sql_snowflake
 from anomstack.external.sqlite.sqlite import read_sql_sqlite
 
+pd.options.display.max_columns = 10
+
 
 def db_translate(sql: str, db: str) -> str:
     """
@@ -58,7 +61,7 @@ def read_sql(sql: str, db: str) -> pd.DataFrame:
 
     sql = db_translate(sql, db)
 
-    logger.debug(f"sql:\n{sql}")
+    logger.debug(f"-- read_sql() is about to read this qry:\n{sql}")
 
     if db == "bigquery":
         df = read_sql_bigquery(sql)
@@ -71,6 +74,8 @@ def read_sql(sql: str, db: str) -> pd.DataFrame:
     else:
         raise ValueError(f"Unknown db: {db}")
 
-    logger.debug(f"df:\n{df}")
+    log_df_info(df, logger)
+    logger.debug(f"df.head():\n{df.head()}")
+    logger.debug(f"df.tail():\n{df.tail()}")
 
     return df

metrics/defaults/sql/change.sql

@@ -1,53 +1,123 @@
 /*
 Template for generating the input data for the change detection job.
+
+Written for SQLite but will be translated to target dialect based on `db` param via sqlglot.
 */
 
-WITH
-
-metric_value_data AS (
-  SELECT DISTINCT
-    metric_timestamp,
-    metric_batch,
-    metric_name,
-    AVG(metric_value) AS metric_value
-  FROM {{ table_key }}
-  WHERE metric_batch = '{{ metric_batch }}'
-    AND metric_type = 'metric'
-    AND DATE(metric_timestamp) >= DATE('now', '-{{ change_metric_timestamp_max_days_ago }} day')
-  GROUP BY metric_timestamp, metric_batch, metric_name
+with
+
+metric_value_data as
+(
+select distinct
+  metric_timestamp,
+  metric_batch,
+  metric_name,
+  avg(metric_value) as metric_value
+from
+  {{ table_key }}
+where
+  metric_batch = '{{ metric_batch }}'
+  and
+  metric_type = 'metric'
+  and
+  -- Filter to the last {{ change_metric_timestamp_max_days_ago }} days
+  date(metric_timestamp) >= date('now', '-{{ change_metric_timestamp_max_days_ago }} day')
+group by
+  metric_timestamp, metric_batch, metric_name
+),
+
+metric_change_alert_data as
+(
+select distinct
+  metric_timestamp,
+  metric_batch,
+  metric_name,
+  max(metric_value) as metric_change_alert
+from
+  {{ table_key }}
+where
+  metric_batch = '{{ metric_batch }}'
+  and
+  metric_type = 'change'
+  and
+  -- Filter to the last {{ change_metric_timestamp_max_days_ago }} days
+  date(metric_timestamp) >= date('now', '-{{ change_metric_timestamp_max_days_ago }} day')
+group by
+  metric_timestamp, metric_batch, metric_name
+),
+
+metric_value_recency_ranked as
+(
+select distinct
+  metric_value_data.metric_timestamp,
+  metric_value_data.metric_batch,
+  metric_value_data.metric_name,
+  metric_value_data.metric_value,
+  -- If change alert not found found for the metric, default to 0
+  coalesce(metric_change_alert_data.metric_change_alert, 0) as metric_change_alert,
+  -- Rank the metric values by recency, with 1 being the most recent
+  row_number() over (partition by metric_value_data.metric_name order by metric_value_data.metric_timestamp desc) as metric_value_recency_rank
+from
+  metric_value_data
+left outer join
+  metric_change_alert_data
+on
+  metric_value_data.metric_batch = metric_change_alert_data.metric_batch
+  and
+  metric_value_data.metric_name = metric_change_alert_data.metric_name
+  and
+  metric_value_data.metric_timestamp = metric_change_alert_data.metric_timestamp
 ),
 
-metric_value_recency_ranked AS (
-  SELECT DISTINCT
-    metric_timestamp,
-    metric_batch,
-    metric_name,
-    metric_value,
-    ROW_NUMBER() OVER (PARTITION BY metric_name ORDER BY metric_timestamp DESC) AS metric_value_recency_rank
-  FROM metric_value_data
+-- Snooze any metrics with change alerts in the last {{ change_snooze_n }} values
+snoozed_metric_names as
+(
+select distinct
+  metric_name
+from
+  metric_value_recency_ranked
+where
+  -- Exclude metrics with change alerts in the last {{ change_snooze_n }} values
+  metric_change_alert = 1
+  and
+  metric_value_recency_rank <= {{ change_snooze_n }}
 ),
 
-data_smoothed AS (
-  SELECT
-    metric_timestamp,
-    metric_batch,
-    metric_name,
-    metric_value,
-    metric_value_recency_rank,
-    -- Smooth the metric value over the last {{ change_smooth_n }} values
-    (SELECT AVG(mv.metric_value)
-     FROM metric_value_recency_ranked mv
-     WHERE mv.metric_name = mr.metric_name
-     AND mv.metric_value_recency_rank BETWEEN mr.metric_value_recency_rank - {{ change_smooth_n }} AND mr.metric_value_recency_rank) AS metric_value_smooth
-  FROM metric_value_recency_ranked mr
-  WHERE metric_value_recency_rank <= {{ change_max_n }}
+data_smoothed as
+(
+select
+  metric_timestamp,
+  metric_batch,
+  metric_name,
+  metric_value,
+  metric_change_alert,
+  metric_value_recency_rank,
+  -- Smooth the metric value over the last {{ change_smooth_n }} values
+  (
+    select
+      avg(mv.metric_value)
+    from
+      metric_value_recency_ranked mv
+    where
+      mv.metric_name = mr.metric_name
+      and
+      mv.metric_value_recency_rank between mr.metric_value_recency_rank - {{ change_smooth_n }} and mr.metric_value_recency_rank
+  ) as metric_value_smooth
+from
+  metric_value_recency_ranked mr
+where
+  metric_value_recency_rank <= {{ change_max_n }}
+  and
+  -- Exclude snoozed metrics
+  metric_name not in (select metric_name from snoozed_metric_names)
 )
 
-SELECT
+select
   metric_timestamp,
   metric_batch,
   metric_name,
   metric_value,
   metric_value_smooth
-FROM data_smoothed
+from
+  data_smoothed
 ;