feat: add local PostgreSQL and data quality reporting

Author: phlppgdfry

.env.example

@@ -0,0 +1,2 @@
+# Alleen voor lokale ontwikkeling. Plaats echte cloudcredentials nooit in Git.
+DATABASE_URL=postgresql+psycopg://analytics_user:analytics_dev_password@localhost:5433/analytics

.gitignore

@@ -5,6 +5,7 @@ __pycache__/
 
 # Local virtual environment
 .venv/
+.env
 
 # Deployment secrets: never commit database credentials.
 .streamlit/secrets.toml

README.md

@@ -119,9 +119,26 @@ De portfolio-app kan al queryresultaten uit PostgreSQL in Pandas laden via
 in [docs/CLOUD_POSTGRES.md](docs/CLOUD_POSTGRES.md). Databasecredentials horen
 uitsluitend in Streamlit secrets, nooit in Git.
 
+## PostgreSQL lokaal met Docker
+
+Voor een realistische lokale bedrijfsflow start je PostgreSQL met Docker en
+laad je de schoongemaakte data erin:
+
+```bash
+docker compose up -d
+python src/load_to_postgres.py
+DATABASE_URL="postgresql+psycopg://analytics_user:analytics_dev_password@localhost:5433/analytics" streamlit run dashboard/app.py
+```
+
+De app toont dan **Data source: PostgreSQL**. Zonder `DATABASE_URL` gebruikt
+dezelfde app automatisch de CSV-fallback, waardoor de publieke demo betrouwbaar
+en gratis blijft.
+
 ## Data quality
 
 Naast de cleaning-pipeline valideert [Pandera](https://pandera.readthedocs.io/)
 de finale dataset als data contract: verplichte kolommen, geldige landen en
 producten, unieke order-ID's, positieve prijzen/kosten en een geldige marge.
 Een foute bronrij komt dus niet stilzwijgend in het dashboard terecht.
+De pipeline schrijft bovendien `reports/data_quality_report.json` met het aantal
+bronrijen, verwijderde rijen, dubbele orders en behoudpercentage.

dashboard/app.py

@@ -1,6 +1,7 @@
 """Start met: streamlit run dashboard/app.py"""
 
 from pathlib import Path
+import os
 import sys
 
 import pandas as pd
@@ -28,7 +29,7 @@
     top_customers,
 )
 from src.clean_data import clean_sales_data  # noqa: E402
-from src.load_data import load_customer_directory_csv, load_monthly_targets_csv, load_sales_csv  # noqa: E402
+from src.load_data import load_customer_directory_csv, load_monthly_targets_csv, load_sales_csv, load_sales_from_postgres  # noqa: E402
 from visualizations.plotly_charts import create_actual_vs_target_chart, create_cohort_retention_chart, create_monthly_performance_chart  # noqa: E402
 
 
@@ -49,12 +50,26 @@
 RAW_DATA_PATH = PROJECT_ROOT / "data" / "raw" / "sales.csv"
 TARGETS_DATA_PATH = PROJECT_ROOT / "data" / "raw" / "monthly_targets.csv"
 CUSTOMERS_DATA_PATH = PROJECT_ROOT / "data" / "raw" / "customers.csv"
+POSTGRES_QUERY_PATH = PROJECT_ROOT / "sql" / "dashboard_sales.sql"
+
+
+def get_database_url() -> str | None:
+    """Lees een optionele lokale of cloud databaseverbinding zonder secrets te loggen."""
+    if database_url := os.getenv("DATABASE_URL"):
+        return database_url
+    try:
+        return st.secrets.get("DATABASE_URL")
+    except Exception:
+        return None
 
 
 @st.cache_data
-def get_sales_data(source_version: int) -> pd.DataFrame:
-    """Clean raw data; its file version safely refreshes Streamlit's cache."""
+def get_sales_data(source_version: int, database_url: str | None) -> pd.DataFrame:
+    """Gebruik PostgreSQL wanneer geconfigureerd, anders de betrouwbare CSV-fallback."""
     del source_version
+    if database_url:
+        query = POSTGRES_QUERY_PATH.read_text(encoding="utf-8")
+        return clean_sales_data(load_sales_from_postgres(database_url, query))
     return clean_sales_data(load_sales_csv(RAW_DATA_PATH))
 
 
@@ -72,12 +87,14 @@ def get_customer_directory(source_version: int) -> pd.DataFrame:
     return load_customer_directory_csv(CUSTOMERS_DATA_PATH)
 
 
-data = get_sales_data(RAW_DATA_PATH.stat().st_mtime_ns)
+database_url = get_database_url()
+data = get_sales_data(RAW_DATA_PATH.stat().st_mtime_ns, database_url)
 monthly_targets = get_monthly_targets(TARGETS_DATA_PATH.stat().st_mtime_ns)
 customer_directory = get_customer_directory(CUSTOMERS_DATA_PATH.stat().st_mtime_ns)
 st.markdown('<p class="eyebrow">Portfolio project · Sales intelligence</p>', unsafe_allow_html=True)
 st.title("Sales analytics dashboard")
 st.caption("Analyseer 24 maanden verkoopdata, vergelijk periodes en exporteer gefilterde inzichten.")
+st.caption("Data source: PostgreSQL" if database_url else "Data source: CSV fallback")
 
 with st.sidebar:
     st.header("Filters")

docker-compose.yml

@@ -0,0 +1,21 @@
+services:
+  postgres:
+    image: postgres:16-alpine
+    container_name: analytics-postgres
+    environment:
+      POSTGRES_DB: analytics
+      POSTGRES_USER: analytics_user
+      POSTGRES_PASSWORD: analytics_dev_password
+    ports:
+      - "5433:5432"
+    volumes:
+      - postgres_data:/var/lib/postgresql/data
+      - ./sql/schema.sql:/docker-entrypoint-initdb.d/01-schema.sql:ro
+    healthcheck:
+      test: ["CMD-SHELL", "pg_isready -U analytics_user -d analytics"]
+      interval: 5s
+      timeout: 5s
+      retries: 10
+
+volumes:
+  postgres_data:

reports/data_quality_report.json

@@ -0,0 +1,7 @@
+{
+  "raw_rows": 904,
+  "clean_rows": 900,
+  "removed_rows": 4,
+  "duplicate_orders": 1,
+  "retention_percent": 99.56
+}

sql/dashboard_sales.sql

@@ -0,0 +1,3 @@
+SELECT order_id, order_date, customer_id, country, product, quantity, price, unit_cost
+FROM sales_orders
+ORDER BY order_date, order_id;

sql/queries.sql

@@ -1,12 +1,15 @@
--- Gebruik deze query later met load_sales_from_postgres().
-SELECT order_id, order_date, customer_id, country, product, quantity, price,
-       quantity * price AS revenue
+-- Dashboardbron: PostgreSQL-resultaat met dezelfde vorm als de CSV-bron.
+SELECT order_id, order_date, customer_id, country, product, quantity, price, unit_cost
 FROM sales_orders
-WHERE order_date >= DATE '2026-01-01'
+WHERE order_date >= DATE '2024-01-01'
 ORDER BY order_date, order_id;
 
--- Een voorbeeld van analyse rechtstreeks in SQL.
-SELECT country, SUM(quantity * price) AS revenue
+-- Zakelijke analyse rechtstreeks in SQL.
+SELECT
+    country,
+    SUM(revenue) AS revenue,
+    SUM(gross_profit) AS gross_profit,
+    ROUND(SUM(gross_profit) / NULLIF(SUM(revenue), 0) * 100, 2) AS margin_percent
 FROM sales_orders
 GROUP BY country
-ORDER BY revenue DESC;
+ORDER BY gross_profit DESC;

sql/schema.sql

@@ -1,10 +1,26 @@
--- PostgreSQL-schema voor wanneer CSV door een echte database wordt vervangen.
+-- Lokale PostgreSQL-datalaag voor de analytics-app.
 CREATE TABLE IF NOT EXISTS sales_orders (
     order_id INTEGER PRIMARY KEY,
     order_date DATE NOT NULL,
     customer_id VARCHAR(50) NOT NULL,
     country VARCHAR(100) NOT NULL,
     product VARCHAR(100) NOT NULL,
     quantity INTEGER NOT NULL CHECK (quantity > 0),
-    price NUMERIC(10, 2) NOT NULL CHECK (price > 0)
+    price NUMERIC(10, 2) NOT NULL CHECK (price > 0),
+    unit_cost NUMERIC(10, 2) NOT NULL CHECK (unit_cost > 0 AND unit_cost < price),
+    revenue NUMERIC(12, 2) NOT NULL CHECK (revenue > 0),
+    gross_profit NUMERIC(12, 2) NOT NULL CHECK (gross_profit > 0),
+    margin_percent NUMERIC(5, 2) NOT NULL CHECK (margin_percent BETWEEN 0 AND 100)
+);
+
+CREATE TABLE IF NOT EXISTS monthly_targets (
+    month DATE PRIMARY KEY,
+    target_revenue NUMERIC(12, 2) NOT NULL CHECK (target_revenue > 0)
+);
+
+CREATE TABLE IF NOT EXISTS customers (
+    customer_id VARCHAR(50) PRIMARY KEY,
+    customer_name VARCHAR(100) NOT NULL,
+    customer_since DATE NOT NULL,
+    acquisition_channel VARCHAR(50) NOT NULL
 );

src/data_quality.py

@@ -0,0 +1,32 @@
+"""Meet en rapporteer datakwaliteit vóór en na de cleaning-pipeline."""
+
+import json
+from pathlib import Path
+
+import pandas as pd
+
+
+PROJECT_ROOT = Path(__file__).resolve().parents[1]
+DEFAULT_REPORT_PATH = PROJECT_ROOT / "reports" / "data_quality_report.json"
+
+
+def build_data_quality_report(raw_data: pd.DataFrame, cleaned_data: pd.DataFrame) -> dict[str, int | float]:
+    """Geef transparante aantallen voor verwijderde en behouden rijen."""
+    raw_rows = len(raw_data)
+    clean_rows = len(cleaned_data)
+    duplicate_orders = int(raw_data.duplicated(subset="order_id").sum())
+    return {
+        "raw_rows": raw_rows,
+        "clean_rows": clean_rows,
+        "removed_rows": raw_rows - clean_rows,
+        "duplicate_orders": duplicate_orders,
+        "retention_percent": round(clean_rows / raw_rows * 100, 2) if raw_rows else 0.0,
+    }
+
+
+def save_data_quality_report(report: dict[str, int | float], path: str | Path = DEFAULT_REPORT_PATH) -> Path:
+    """Schrijf het kwaliteitsrapport als leesbaar JSON-bestand weg."""
+    output_path = Path(path)
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+    output_path.write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8")
+    return output_path