validate_data.py

"""Data-quality validation with Great Expectations (GX Core 1.x).

Domain-specific guardrails that must hold before the dashboard/marts are trusted.
Run after the pipeline:  python src/validate_data.py
Writes a pass/fail report to docs/data_quality_report.md and exits non-zero on
any *critical* failure (so it can gate a refresh in CI).
"""
from __future__ import annotations
import pathlib
import sys
import pandas as pd
import great_expectations as gx

ROOT = pathlib.Path(__file__).resolve().parents[1]
FACT = ROOT / "data" / "processed" / "fact_orders.csv"
REPORT = ROOT / "docs" / "data_quality_report.md"


def main() -> None:
    if not FACT.exists():
        sys.exit("fact_orders.csv not found — run `python src/run_pipeline.py` first.")
    df = pd.read_csv(FACT)

    ctx = gx.get_context()  # ephemeral
    batch = (ctx.data_sources.add_pandas("supply_chain")
             .add_dataframe_asset("fact_orders")
             .add_batch_definition_whole_dataframe("batch")
             .get_batch(batch_parameters={"dataframe": df}))

    E = gx.expectations
    # (expectation, is_critical) — critical failures gate the pipeline
    suite_specs = [
        (E.ExpectColumnValuesToNotBeNull(column="order_id"), True),
        (E.ExpectColumnValuesToBeUnique(column="order_id"), True),
        (E.ExpectColumnValuesToBeBetween(column="revenue", min_value=0), True),
        (E.ExpectColumnValuesToBeBetween(column="freight", min_value=0), True),
        (E.ExpectColumnValuesToBeBetween(column="freight_ratio", min_value=0,
                                         max_value=10, mostly=0.99), False),
        (E.ExpectColumnValuesToBeBetween(column="review_score", min_value=1,
                                         max_value=5), True),
        (E.ExpectColumnValuesToBeBetween(column="delay_days", min_value=-120,
                                         max_value=240, mostly=0.999), False),
        (E.ExpectColumnValuesToBeInSet(column="is_late", value_set=[True, False]), True),
        (E.ExpectColumnValuesToBeInSet(column="is_delivered", value_set=[True, False]), True),
        (E.ExpectColumnValuesToNotBeNull(column="region_state", mostly=0.98), False),
    ]

    rows, n_fail, n_crit_fail = [], 0, 0
    for exp, critical in suite_specs:
        r = batch.validate(exp)
        ok = bool(r.success)
        n_fail += (not ok)
        n_crit_fail += (not ok and critical)
        kwargs = exp.configuration.kwargs if hasattr(exp, "configuration") else {}
        rows.append({
            "expectation": type(exp).__name__,
            "column": getattr(exp, "column", ""),
            "critical": "yes" if critical else "no",
            "result": "PASS" if ok else "FAIL",
            "unexpected_pct": round(r.result.get("unexpected_percent", 0.0), 3)
            if r.result else 0.0,
        })

    rep = pd.DataFrame(rows)
    lines = ["# Data Quality Report — fact_orders\n",
             f"_{len(df):,} rows · {len(suite_specs)} expectations · "
             f"{len(suite_specs)-n_fail} passed, {n_fail} failed "
             f"({n_crit_fail} critical)_\n",
             rep.to_markdown(index=False),
             "\n\nGenerated by `src/validate_data.py` (Great Expectations Core)."]
    REPORT.write_text("\n".join(lines))
    print(rep.to_string(index=False))
    print(f"\n{len(suite_specs)-n_fail}/{len(suite_specs)} passed; "
          f"{n_crit_fail} critical failures. Report -> {REPORT.relative_to(ROOT)}")
    if n_crit_fail:
        sys.exit(1)


if __name__ == "__main__":
    main()