Sales Data Warehouse &
Analytics Pipeline

import pandas as pd
from sqlalchemy import create_engine
from datetime import datetime

engine = create_engine("postgresql://user:pass@localhost/sales_dw")

def transform_orders(df: pd.DataFrame) -> pd.DataFrame:
    """Clean, validate, and conform raw order data."""
    df = df.dropna(subset=["order_id", "customer_id", "revenue"])
    df["revenue"] = pd.to_numeric(df["revenue"], errors="coerce")
    df["order_date"] = pd.to_datetime(df["order_date"])
    df["discount_pct"] = df["discount_pct"].fillna(0).clip(0, 100)

    # Derive date_id from dim_date surrogate key
    df["date_id"] = df["order_date"].dt.strftime("%Y%m%d").astype(int)
    return df

def load(df: pd.DataFrame, table: str):
    df.to_sql(table, engine, if_exists="append", index=False,
              method="multi", chunksize=5000)
    print(f"  Loaded {len(df):,} rows → {table}")

if __name__ == "__main__":
    raw = pd.read_csv("raw/orders_2024.csv")
    clean = transform_orders(raw)
    load(clean, "fact_orders")
    print("ETL complete:", datetime.now())

WITH yearly AS (
  SELECT
    d.year,
    SUM(f.revenue)                              AS total_revenue,
    LAG(SUM(f.revenue)) OVER (ORDER BY d.year)  AS prev_revenue
  FROM fact_orders f
  JOIN dim_date d ON f.date_id = d.date_id
  GROUP BY d.year
)
SELECT
  year,
  total_revenue,
  ROUND(
    (total_revenue - prev_revenue) / prev_revenue * 100, 2
  ) AS yoy_growth_pct
FROM yearly
ORDER BY year;

WITH first_order AS (
  SELECT customer_id, MIN(order_date) AS cohort_date
  FROM fact_orders f
  JOIN dim_date d ON f.date_id = d.date_id
  GROUP BY customer_id
),
retention AS (
  SELECT
    DATE_TRUNC('month', fo.cohort_date)  AS cohort_month,
    COUNT(DISTINCT fo.customer_id)       AS cohort_size,
    COUNT(DISTINCT CASE
      WHEN d.full_date BETWEEN fo.cohort_date + 1
                           AND fo.cohort_date + 90
      THEN f.customer_id END)            AS returned_90d
  FROM first_order fo
  JOIN fact_orders f  USING (customer_id)
  JOIN dim_date d     ON f.date_id = d.date_id
  GROUP BY cohort_month
)
SELECT
  cohort_month,
  cohort_size,
  returned_90d,
  ROUND(returned_90d::numeric / cohort_size * 100, 1) AS retention_rate
FROM retention
ORDER BY cohort_month;

SELECT
  p.category,
  p.product_name,
  SUM(f.revenue)                                     AS total_revenue,
  SUM(f.quantity * p.unit_cost)                      AS total_cost,
  ROUND(
    (SUM(f.revenue) - SUM(f.quantity * p.unit_cost))
    / SUM(f.revenue) * 100, 2
  )                                                  AS margin_pct,
  RANK() OVER (
    PARTITION BY p.category ORDER BY
    SUM(f.revenue) - SUM(f.quantity * p.unit_cost) DESC
  )                                                  AS rank_in_category
FROM fact_orders f
JOIN dim_products p ON f.product_id = p.product_id
GROUP BY p.category, p.product_id, p.product_name, p.unit_cost
ORDER BY margin_pct DESC
LIMIT 20;