feat: restore runner, add Docker execution, and harden segmentation

kipruto45 · kipruto45 · commit 7587a0986356 · 2026-02-19T19:25:20.000+03:00
diff --git a/README.md b/README.md
@@ -38,6 +38,13 @@ chmod +x run_all.sh
 ./run_all.sh
 ```
 
+Run with Docker (PostgreSQL + pipeline):
+
+```bash
+docker compose up -d postgres
+docker compose --profile run run --rm pipeline
+```
+
 Manual execution:
 
 1. Create database (example):
@@ -79,6 +86,7 @@ psql "$DATABASE_URL" -f analytics/revenue_analysis.sql
 psql "$DATABASE_URL" -f tests/test_data_load.sql
 psql "$DATABASE_URL" -f tests/test_scd_logic.sql
 psql "$DATABASE_URL" -f tests/test_quality_checks.sql
+psql "$DATABASE_URL" -f tests/test_customer_segmentation.sql
 ```
 
 ## Pipeline Flow
diff --git a/analytics/customer_segmentation.sql b/analytics/customer_segmentation.sql
@@ -1,38 +1,80 @@
--- RFM-based segmentation using quartiles.
-WITH rfm_base AS (
+-- Rule-based segmentation with explicit thresholds.
+-- Optional deterministic date for tests:
+--   SET app.segment_as_of_date = '2026-02-19';
+
+CREATE OR REPLACE VIEW warehouse.v_customer_segmentation AS
+WITH params AS (
+    SELECT COALESCE(
+        NULLIF(current_setting('app.segment_as_of_date', true), '')::DATE,
+        CURRENT_DATE
+    ) AS as_of_date
+), thresholds AS (
+    SELECT
+        30::INT AS champion_max_recency_days,
+        3::INT AS champion_min_paid_orders,
+        400::NUMERIC(12,2) AS champion_min_revenue,
+        60::INT AS high_value_max_recency_days,
+        300::NUMERIC(12,2) AS high_value_min_revenue,
+        3::INT AS loyal_min_paid_orders,
+        90::INT AS loyal_max_recency_days,
+        75::INT AS at_risk_min_recency_days
+), customer_kpis AS (
     SELECT
         c.customer_id,
-        MAX(o.order_date) AS last_order_date,
-        COUNT(DISTINCT o.order_id) FILTER (WHERE o.status = 'PAID') AS frequency,
-        COALESCE(SUM(t.amount) FILTER (WHERE o.status = 'PAID' AND t.status = 'SUCCESS'), 0) AS monetary
+        c.signup_date,
+        MAX(o.order_date) FILTER (WHERE o.status = 'PAID') AS last_paid_order_date,
+        COUNT(DISTINCT o.order_id) FILTER (WHERE o.status = 'PAID') AS paid_orders,
+        COALESCE(SUM(t.amount) FILTER (WHERE o.status = 'PAID' AND t.status = 'SUCCESS'), 0) AS lifetime_revenue
     FROM oltp.customers c
-    LEFT JOIN oltp.orders o ON o.customer_id = c.customer_id
-    LEFT JOIN oltp.transactions t ON t.order_id = o.order_id
-    GROUP BY c.customer_id
-), scored AS (
+    LEFT JOIN oltp.orders o
+        ON o.customer_id = c.customer_id
+    LEFT JOIN oltp.transactions t
+        ON t.order_id = o.order_id
+    GROUP BY c.customer_id, c.signup_date
+), segmentation_base AS (
     SELECT
-        customer_id,
-        CURRENT_DATE - COALESCE(last_order_date, CURRENT_DATE) AS recency_days,
-        frequency,
-        monetary,
-        NTILE(4) OVER (ORDER BY CURRENT_DATE - COALESCE(last_order_date, CURRENT_DATE) DESC) AS recency_score,
-        NTILE(4) OVER (ORDER BY frequency ASC) AS frequency_score,
-        NTILE(4) OVER (ORDER BY monetary ASC) AS monetary_score
-    FROM rfm_base
+        ck.customer_id,
+        p.as_of_date,
+        (p.as_of_date - COALESCE(ck.last_paid_order_date, ck.signup_date, p.as_of_date))::INT AS recency_days,
+        ck.paid_orders,
+        ck.lifetime_revenue,
+        ROUND(ck.lifetime_revenue / NULLIF(ck.paid_orders, 0), 2) AS avg_paid_order_value
+    FROM customer_kpis ck
+    CROSS JOIN params p
 )
 SELECT
-    customer_id,
-    recency_days,
-    frequency,
-    monetary,
-    recency_score,
-    frequency_score,
-    monetary_score,
+    sb.customer_id,
+    sb.as_of_date,
+    sb.recency_days,
+    sb.paid_orders,
+    sb.lifetime_revenue,
+    sb.avg_paid_order_value,
     CASE
-        WHEN recency_score = 1 AND frequency_score = 4 AND monetary_score = 4 THEN 'Champions'
-        WHEN recency_score <= 2 AND monetary_score >= 3 THEN 'High Value'
-        WHEN recency_score >= 3 AND frequency_score <= 2 THEN 'At Risk'
+        WHEN sb.paid_orders = 0 THEN 'Inactive'
+        WHEN sb.recency_days <= t.champion_max_recency_days
+             AND sb.paid_orders >= t.champion_min_paid_orders
+             AND sb.lifetime_revenue >= t.champion_min_revenue
+            THEN 'Champions'
+        WHEN sb.recency_days <= t.high_value_max_recency_days
+             AND sb.lifetime_revenue >= t.high_value_min_revenue
+            THEN 'High Value'
+        WHEN sb.paid_orders >= t.loyal_min_paid_orders
+             AND sb.recency_days <= t.loyal_max_recency_days
+            THEN 'Loyal'
+        WHEN sb.recency_days >= t.at_risk_min_recency_days
+            THEN 'At Risk'
         ELSE 'Regular'
     END AS segment
-FROM scored
-ORDER BY monetary DESC;
+FROM segmentation_base sb
+CROSS JOIN thresholds t;
+
+SELECT
+    customer_id,
+    as_of_date,
+    recency_days,
+    paid_orders,
+    lifetime_revenue,
+    avg_paid_order_value,
+    segment
+FROM warehouse.v_customer_segmentation
+ORDER BY lifetime_revenue DESC, paid_orders DESC, customer_id;
diff --git a/docker-compose.yml b/docker-compose.yml
@@ -0,0 +1,39 @@
+services:
+  postgres:
+    image: postgres:16
+    container_name: sql_data_postgres
+    restart: unless-stopped
+    environment:
+      POSTGRES_DB: sql_data_engineering
+      POSTGRES_USER: postgres
+      POSTGRES_PASSWORD: postgres
+    ports:
+      - "${PG_HOST_PORT:-5433}:5432"
+    healthcheck:
+      test: ["CMD-SHELL", "pg_isready -U postgres -d sql_data_engineering"]
+      interval: 10s
+      timeout: 5s
+      retries: 10
+    volumes:
+      - postgres_data:/var/lib/postgresql/data
+
+  pipeline:
+    image: postgres:16
+    depends_on:
+      postgres:
+        condition: service_healthy
+    working_dir: /workspace
+    volumes:
+      - ./:/workspace
+    environment:
+      DB_HOST: postgres
+      DB_PORT: 5432
+      DB_NAME: sql_data_engineering
+      DB_USER: postgres
+      DB_PASSWORD: postgres
+      DATABASE_URL: postgresql://postgres:postgres@postgres:5432/sql_data_engineering
+    command: ["bash", "-lc", "chmod +x ./run_all.sh && ./run_all.sh"]
+    profiles: ["run"]
+
+volumes:
+  postgres_data:
diff --git a/run_all.sh b/run_all.sh
@@ -4,23 +4,45 @@ set -euo pipefail
 ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
 cd "$ROOT_DIR"
 
-if [[ ! -f ".env" ]]; then
-  echo "Missing .env in $ROOT_DIR"
-  exit 1
-fi
+load_env_file() {
+  local env_file="$1"
+  while IFS= read -r line; do
+    [[ -z "$line" || "$line" =~ ^[[:space:]]*# ]] && continue
+    [[ "$line" != *=* ]] && continue
+
+    local key="${line%%=*}"
+    local value="${line#*=}"
+    key="${key//[[:space:]]/}"
+
+    if [[ -z "$key" ]]; then
+      continue
+    fi
+
+    # Keep externally provided values (useful for Docker/CI overrides).
+    if [[ -z "${!key+x}" ]]; then
+      export "$key=$value"
+    fi
+  done < "$env_file"
+}
 
-set -a
-source .env
-set +a
+if [[ -f ".env" ]]; then
+  load_env_file ".env"
+else
+  echo "No .env found. Using environment variables only."
+fi
 
 if [[ -z "${DATABASE_URL:-}" ]]; then
-  echo "DATABASE_URL is not set in .env"
-  exit 1
+  if [[ -n "${DB_HOST:-}" && -n "${DB_PORT:-}" && -n "${DB_NAME:-}" && -n "${DB_USER:-}" && -n "${DB_PASSWORD:-}" ]]; then
+    export DATABASE_URL="postgresql://${DB_USER}:${DB_PASSWORD}@${DB_HOST}:${DB_PORT}/${DB_NAME}"
+  else
+    echo "DATABASE_URL is not set. Provide it directly or via DB_* variables."
+    exit 1
+  fi
 fi
 
-export PGPASSWORD="${DB_PASSWORD:-}"
+export PGPASSWORD="${DB_PASSWORD:-${PGPASSWORD:-}}"
 
-ADMIN_URL="postgresql://${DB_USER:-postgres}:${DB_PASSWORD:-}@${DB_HOST:-localhost}:${DB_PORT:-5432}/postgres"
+ADMIN_URL="postgresql://${DB_USER:-postgres}:${DB_PASSWORD:-postgres}@${DB_HOST:-localhost}:${DB_PORT:-5432}/postgres"
 TARGET_DB="${DB_NAME:-sql_data_engineering}"
 
 echo "Checking target database: $TARGET_DB"
@@ -69,5 +91,6 @@ run_sql "performance/optimization_examples.sql"
 run_sql "tests/test_data_load.sql"
 run_sql "tests/test_scd_logic.sql"
 run_sql "tests/test_quality_checks.sql"
+run_sql "tests/test_customer_segmentation.sql"
 
 echo "Pipeline completed successfully."
diff --git a/tests/test_customer_segmentation.sql b/tests/test_customer_segmentation.sql
@@ -0,0 +1,78 @@
+-- Deterministic segmentation tests.
+SET app.segment_as_of_date = '2026-02-19';
+
+-- Test 1: segmentation returns one row per customer and no NULL segment.
+DO $$
+DECLARE
+    v_customers INT;
+    v_segmented INT;
+    v_null_segments INT;
+BEGIN
+    SELECT COUNT(*) INTO v_customers FROM oltp.customers;
+    SELECT COUNT(*) INTO v_segmented FROM warehouse.v_customer_segmentation;
+    SELECT COUNT(*) INTO v_null_segments
+    FROM warehouse.v_customer_segmentation
+    WHERE segment IS NULL;
+
+    IF v_segmented <> v_customers THEN
+        RAISE EXCEPTION 'Segmentation row mismatch: expected %, got %.', v_customers, v_segmented;
+    END IF;
+
+    IF v_null_segments > 0 THEN
+        RAISE EXCEPTION 'Segmentation contains % NULL segments.', v_null_segments;
+    END IF;
+END $$;
+
+-- Test 2: segment values are within approved taxonomy.
+DO $$
+DECLARE
+    v_invalid INT;
+BEGIN
+    SELECT COUNT(*) INTO v_invalid
+    FROM warehouse.v_customer_segmentation
+    WHERE segment NOT IN ('Champions', 'High Value', 'Loyal', 'At Risk', 'Regular', 'Inactive');
+
+    IF v_invalid > 0 THEN
+        RAISE EXCEPTION 'Found % rows with invalid segment labels.', v_invalid;
+    END IF;
+END $$;
+
+-- Test 3: inactive customers must have zero paid orders and vice versa.
+DO $$
+DECLARE
+    v_mismatch INT;
+BEGIN
+    SELECT COUNT(*) INTO v_mismatch
+    FROM warehouse.v_customer_segmentation
+    WHERE (paid_orders = 0 AND segment <> 'Inactive')
+       OR (paid_orders > 0 AND segment = 'Inactive');
+
+    IF v_mismatch > 0 THEN
+        RAISE EXCEPTION 'Found % rows violating inactive segmentation rule.', v_mismatch;
+    END IF;
+END $$;
+
+-- Test 4: regression checks for known sample customers.
+DO $$
+DECLARE
+    v_customer_4 TEXT;
+    v_customer_6 TEXT;
+BEGIN
+    SELECT segment INTO v_customer_4
+    FROM warehouse.v_customer_segmentation
+    WHERE customer_id = 4;
+
+    SELECT segment INTO v_customer_6
+    FROM warehouse.v_customer_segmentation
+    WHERE customer_id = 6;
+
+    IF v_customer_4 <> 'High Value' THEN
+        RAISE EXCEPTION 'Expected customer 4 to be High Value, got %.', COALESCE(v_customer_4, 'NULL');
+    END IF;
+
+    IF v_customer_6 <> 'Inactive' THEN
+        RAISE EXCEPTION 'Expected customer 6 to be Inactive, got %.', COALESCE(v_customer_6, 'NULL');
+    END IF;
+END $$;
+
+RESET app.segment_as_of_date;
