Metric Monitoring and Segmentation Questions

SELECT utm_source, COUNT(DISTINCT user_id) AS new_users
FROM events
WHERE event_date = '2025-11-01' AND is_new_user = TRUE
GROUP BY utm_source;

SELECT ip, user_agent, COUNT(DISTINCT user_id) AS ids, COUNT(*) AS events
FROM events
WHERE event_date = '2025-11-01'
GROUP BY ip, user_agent
HAVING events > 100 OR ids > 5;

SELECT client_id, COUNT(DISTINCT user_id) AS users
FROM events
WHERE event_date = '2025-11-01'
GROUP BY client_id
HAVING users > 1;

-- Day 1 and Day 7 retention for cohort from spike date
WITH cohort AS (
  SELECT user_id FROM users WHERE signup_date = '2025-11-01'
)
SELECT 
  SUM(CASE WHEN event_date = '2025-11-02' THEN 1 ELSE 0 END) AS day1_active,
  SUM(CASE WHEN event_date BETWEEN '2025-11-02' AND '2025-11-08' THEN 1 ELSE 0 END) AS day7_active
FROM events
WHERE user_id IN (SELECT user_id FROM cohort);

# Config
MIN_SAMPLE_SIZE = 50          # minimum observations per bucket
BUCKET_SIZE_MIN = 5           # minutes
CONSECUTIVE_BUCKETS = 3       # require change to persist across 3 buckets
ALERT_THRESHOLD = 0.20        # 20% relative change triggers
CLEAR_THRESHOLD = 0.10        # must fall below 10% to clear
SILENCE_SECONDS = 3600        # backoff after alert fired

# state per segment stored in DB/cache
state = {
  'last_alert_ts': None,
  'consecutive_change_count': 0,
  'last_bucket_value': None,
  'status': 'OK'  # OK, ALERT, INSUFFICIENT_DATA
}

def evaluate_bucket(segment_id, bucket_start, bucket_end, value, sample_size):
    s = load_state(segment_id)
    now = current_time()

    # Insufficient data handling
    if sample_size < MIN_SAMPLE_SIZE:
        s['consecutive_change_count'] = 0
        s['last_bucket_value'] = value
        s['status'] = 'INSUFFICIENT_DATA'
        save_state(segment_id, s)
        return

    # If first valid bucket, initialize
    if s['last_bucket_value'] is None:
        s['last_bucket_value'] = value
        s['status'] = 'OK'
        save_state(segment_id, s)
        return

    # Compute relative change
    prev = s['last_bucket_value']
    if prev == 0:
        rel_change = float('inf') if value > 0 else 0
    else:
        rel_change = (value - prev) / prev

    # Check threshold
    if abs(rel_change) >= ALERT_THRESHOLD:
        s['consecutive_change_count'] += 1
    else:
        s['consecutive_change_count'] = 0

    s['last_bucket_value'] = value

    # Fire alert only after required consecutive buckets and backoff
    if s['consecutive_change_count'] >= CONSECUTIVE_BUCKETS:
        if s['status'] != 'ALERT' and (s['last_alert_ts'] is None or now - s['last_alert_ts'] > SILENCE_SECONDS):
            fire_alert(segment_id, rel_change, value, sample_size)
            s['last_alert_ts'] = now
            s['status'] = 'ALERT'
    elif s['status'] == 'ALERT' and abs(rel_change) <= CLEAR_THRESHOLD:
        clear_alert(segment_id)
        s['status'] = 'OK'
        s['consecutive_change_count'] = 0

    save_state(segment_id, s)

WITH stats AS (
  SELECT
    segment,
    COUNT(*)::bigint AS n,
    SUM(CASE WHEN converted THEN 1 ELSE 0 END)::bigint AS conversions,
    -- conversion rate as fraction
    AVG((converted::int))::numeric AS p
  FROM events
  WHERE segment IN ('segment_a','segment_b')
  GROUP BY segment
),
ci AS (
  SELECT
    segment,
    n,
    conversions,
    p,
    -- standard error for proportion
    sqrt(p * (1 - p) / NULLIF(n,0)) AS se,
    (p - 1.96 * sqrt(p * (1 - p) / NULLIF(n,0))) AS ci_lower,
    (p + 1.96 * sqrt(p * (1 - p) / NULLIF(n,0))) AS ci_upper
  FROM stats
),
pair AS (
  SELECT
    a.p  AS p_a,  a.n AS n_a,  a.se AS se_a,
    b.p  AS p_b,  b.n AS n_b,  b.se AS se_b
  FROM ci a
  JOIN ci b ON a.segment = 'segment_a' AND b.segment = 'segment_b'
)
SELECT
  -- per-segment results
  ci.segment,
  ci.n,
  ci.conversions,
  ROUND(ci.p::numeric, 6) AS conversion_rate,
  ROUND(ci.ci_lower::numeric, 6) AS ci_lower,
  ROUND(ci.ci_upper::numeric, 6) AS ci_upper,
  -- difference (percentage points) and its 95% CI
  ROUND((pair.p_a - pair.p_b)::numeric, 6) AS diff_pp,
  ROUND(( (pair.p_a - pair.p_b) - 1.96 * sqrt(pair.se_a^2 + pair.se_b^2) )::numeric, 6) AS diff_ci_lower,
  ROUND(( (pair.p_a - pair.p_b) + 1.96 * sqrt(pair.se_a^2 + pair.se_b^2) )::numeric, 6) AS diff_ci_upper
FROM ci
CROSS JOIN pair
ORDER BY segment;