Data Driven Recommendations and Impact Questions

SELECT max(processed_at) FROM pipeline_runs WHERE job_name='events_ingest';

SELECT date_trunc('hour', received_at) hr, count(*) FROM raw_events
WHERE received_at >= now() - interval '48 hours'
GROUP BY hr ORDER BY hr;

SELECT date(event_time) dt,
  count(*) raw_cnt,
  sum(case when transformed_flag then 1 else 0 end) transformed_cnt
FROM events_joined
WHERE event_time >= current_date - interval '7 days'
GROUP BY dt;

SELECT dt, sum(is_converted::int) conv, count(*) attempts FROM events_clean
GROUP BY dt;

WITH windows AS (
  SELECT 7 AS days UNION ALL SELECT 14 UNION ALL SELECT 30
),
-- per-user cumulative revenue within each window after assignment
user_revenue AS (
  SELECT
    ea.user_id,
    ea.variant,
    w.days,
    SUM(o.amount) FILTER (
      WHERE o.order_date >= ea.assigned_at
        AND o.order_date < ea.assigned_at + (w.days || ' days')::interval
    )::numeric AS revenue_within_window
  FROM email_assignments ea
  CROSS JOIN windows w
  LEFT JOIN orders o
    ON o.user_id = ea.user_id
  GROUP BY ea.user_id, ea.variant, w.days
),
-- aggregate by variant and window: total revenue and assigned users
agg AS (
  SELECT
    ur.days,
    ur.variant,
    COUNT(DISTINCT ur.user_id) AS users_assigned,
    SUM(COALESCE(ur.revenue_within_window,0)) AS total_revenue
  FROM user_revenue ur
  GROUP BY ur.days, ur.variant
),
-- pivot to get treatment and control side-by-side
pivot AS (
  SELECT
    a.days,
    MAX(CASE WHEN a.variant = 'treatment' THEN a.users_assigned END) AS users_treatment,
    MAX(CASE WHEN a.variant = 'treatment' THEN a.total_revenue END) AS revenue_treatment,
    MAX(CASE WHEN a.variant = 'control' THEN a.users_assigned END) AS users_control,
    MAX(CASE WHEN a.variant = 'control' THEN a.total_revenue END) AS revenue_control
  FROM agg a
  GROUP BY a.days
)
SELECT
  p.days,
  coalesce(revenue_treatment,0) AS total_revenue_treatment,
  coalesce(revenue_control,0) AS total_revenue_control,
  coalesce(users_treatment,0) AS users_treatment,
  coalesce(users_control,0) AS users_control,
  -- revenue per 1000 users (normalize by actual assigned users in each arm)
  CASE WHEN users_treatment>0 THEN (revenue_treatment::numeric / users_treatment) * 1000 ELSE NULL END
    AS rev_per_1000_treatment,
  CASE WHEN users_control>0 THEN (revenue_control::numeric / users_control) * 1000 ELSE NULL END
    AS rev_per_1000_control,
  -- incremental revenue per 1000 users (treatment - control)
  CASE
    WHEN users_treatment>0 AND users_control>0 THEN
      (revenue_treatment::numeric / users_treatment - revenue_control::numeric / users_control) * 1000
    ELSE NULL
  END AS incremental_rev_per_1000
FROM pivot p
ORDER BY p.days;

-- Unique users per variant
SELECT variant, COUNT(DISTINCT user_id) AS users
FROM experiment_assignments
WHERE assigned_at < '[experiment_start]'
GROUP BY variant;

-- Pre-test conversion by variant (baseline period)
SELECT variant,
  COUNT(DISTINCT user_id) AS users,
  SUM(CASE WHEN converted_at BETWEEN '[pre_start]' AND '[pre_end]' THEN 1 ELSE 0 END) AS conversions,
  SUM(CASE WHEN converted_at BETWEEN '[pre_start]' AND '[pre_end]' THEN 1 ELSE 0 END)::float / COUNT(DISTINCT user_id) AS conv_rate
FROM experiment_assignments a
LEFT JOIN events e USING(user_id)
GROUP BY variant;

-- Assignment stability over time / drift
SELECT variant, DATE(assigned_at) AS day, COUNT(*) FROM experiment_assignments GROUP BY 1,2 ORDER BY 2;

SELECT user_id, COUNT(DISTINCT variant) as variants, COUNT(*) as rows
FROM experiment_assignments
GROUP BY user_id HAVING COUNT(DISTINCT variant) > 1;

SELECT variant, COUNT(*) FROM (
  SELECT user_id, CASE WHEN ABS(MOD(HASH(user_id||'[salt]'),100)) < X THEN 'treatment' ELSE 'control' END AS variant FROM users
) t GROUP BY variant;

SELECT variant, device_type, COUNT(DISTINCT user_id) FROM experiment_assignments JOIN users USING(user_id) GROUP BY 1,2;