Analytical Background Questions

# tests/test_transforms.py
import pandas as pd
from mypipeline.transforms import compute_retention

def test_compute_retention_basic():
    df = pd.DataFrame({"user_id":[1,1,2],"signup":[0,7,0],"active":[7,14,1]})
    out = compute_retention(df)
    assert "retention_7d" in out.columns
    assert out["retention_7d"].iloc[0] == 1

WITH prev_30 AS (
  SELECT
    r1.region,
    r1.date,
    r1.revenue,
    -- aggregate over prior 30 calendar days (exclude current day)
    AVG(r2.revenue)   AS mean_30,
    STDDEV_SAMP(r2.revenue) AS stddev_30,
    COUNT(r2.revenue) AS cnt_30
  FROM revenue r1
  LEFT JOIN revenue r2
    ON r1.region = r2.region
   AND r2.date BETWEEN DATE_SUB(r1.date, INTERVAL 30 DAY) AND DATE_SUB(r1.date, INTERVAL 1 DAY)
  GROUP BY r1.region, r1.date, r1.revenue
)
SELECT
  region,
  date,
  revenue,
  mean_30,
  stddev_30,
  cnt_30,
  CASE
    WHEN cnt_30 < 10 THEN 'insufficient_data'               -- small-sample handling
    WHEN stddev_30 IS NULL OR stddev_30 = 0 THEN
         CASE WHEN revenue > mean_30 THEN 'anomaly_high' ELSE 'normal' END
    WHEN revenue > mean_30 + 3 * stddev_30 THEN 'anomaly_high'
    ELSE 'normal'
  END AS anomaly_flag
FROM prev_30
ORDER BY region, date;