Building reliable data pipelines

-- Daily revenue per customer, deduplicated and cleaned
WITH ranked_orders AS (
  SELECT
    customer_id,
    order_id,
    order_date::date           AS order_day,
    amount,
    ROW_NUMBER() OVER (
      PARTITION BY order_id
      ORDER BY updated_at DESC
    ) AS rn                       -- keep the latest version of each order
  FROM raw.orders
  WHERE status = 'paid'
    AND amount > 0
)
SELECT
  customer_id,
  order_day,
  COUNT(*)            AS orders,
  SUM(amount)         AS revenue,
  ROUND(AVG(amount),2) AS avg_order_value
FROM ranked_orders
WHERE rn = 1            -- drop duplicate order rows
GROUP BY customer_id, order_day
ORDER BY order_day DESC, revenue DESC;

import pandas as pd
from sqlalchemy import create_engine

def transform_orders(raw_path: str) -> pd.DataFrame:
    """Extract raw orders, clean them, return analysis-ready rows."""
    df = pd.read_csv(raw_path)

    # 1. Standardize & clean
    df.columns = [c.strip().lower() for c in df.columns]
    df["order_date"] = pd.to_datetime(df["order_date"], errors="coerce")
    df = df.dropna(subset=["order_id", "customer_id", "order_date"])

    # 2. Keep only valid, paid orders and drop duplicates
    df = df[(df["status"] == "paid") & (df["amount"] > 0)]
    df = df.sort_values("updated_at").drop_duplicates("order_id", keep="last")

    # 3. Derive features
    df["order_day"] = df["order_date"].dt.date
    df["revenue"] = df["amount"].round(2)
    return df[["customer_id", "order_day", "order_id", "revenue"]]

def load(df: pd.DataFrame, table: str, dsn: str) -> int:
    engine = create_engine(dsn)
    df.to_sql(table, engine, schema="marts", if_exists="replace", index=False)
    return len(df)

if __name__ == "__main__":
    clean = transform_orders("data/raw/orders.csv")
    rows = load(clean, "fct_orders", "postgresql://localhost/warehouse")
    print(f"Loaded {rows} clean rows into marts.fct_orders")