Dq

Skill: Data Quality Checks (Runtime)

Purpose

Validate actual data at runtime — distinct from /test, which validates transformation logic at code-change time. Even correct code can produce bad data if a source changes, data lags, or upstream anomalies occur.

See /test for the canonical Testing vs. DQ distinction.

When to stop at this skill

Done when checks exist for freshness, volume, schema drift, and distribution — each threshold justified by domain knowledge or a documented source contract, not guesswork.

Steps

Step 1 — Identify the 4 required check types

Type	Question	Threshold source
Freshness	Is the latest data within the SLA window?	contracts/source-<name>.yaml → sla.freshness
Volume	Is the row count within normal range?	Historical data: min/max from last 30 days
Schema drift	Did the source add/remove/rename fields?	contracts/source-<name>.yaml → schema
Distribution	Are values within a valid domain range?	Domain knowledge — not AI guesswork

Step 2 — Calibrate thresholds from domain knowledge

Do not let AI set thresholds — AI can suggest, but the user must verify and justify. For example:

• volume_min: Look at 30-day history, find the lowest day's record count. Subtract a 20% buffer.
• price_range: Can a stock price go to $0? Can it exceed $10,000? Domain knowledge.
• freshness: From contract SLA — if the source updates daily, freshness > 25h triggers alert.

Step 3 — Write checks and alerts

Each check must:

• Return clear pass/fail (not just a log message).
• On fail → alert immediately (log level ERROR + notification).
• Record observed vs expected values for easy debugging.

Output

Create quality/dq_checks.py:

"""
quality/dq_checks.py — Runtime data quality monitoring.

Run after each pipeline run to validate data output.
Each check: pass → log INFO; fail → log ERROR + alert.
"""

import json
import logging
from datetime import datetime, timezone, timedelta
from pathlib import Path
from typing import Callable

logger = logging.getLogger("dq-checks")


# ─── Check framework ─────────────────────────────────────────────────────────

class DQResult:
    def __init__(self, check_name: str, passed: bool, observed: any,
                 expected: str, table: str, severity: str = "error"):
        self.check_name = check_name
        self.passed = passed
        self.observed = observed
        self.expected = expected
        self.table = table
        self.severity = severity
        self.timestamp = datetime.now(timezone.utc).isoformat()

    def __repr__(self):
        status = "PASS" if self.passed else f"FAIL [{self.severity.upper()}]"
        return f"{status} | {self.table}.{self.check_name} | observed={self.observed} | expected={self.expected}"


def run_check(name: str, table: str, fn: Callable, expected_desc: str,
              severity: str = "error") -> DQResult:
    try:
        passed, observed = fn()
        result = DQResult(name, passed, observed, expected_desc, table, severity)
        if passed:
            logger.info(str(result))
        else:
            logger.error(str(result))
            send_alert(result)
        return result
    except Exception as e:
        logger.error(f"Check {name} raised exception: {e}")
        return DQResult(name, False, str(e), expected_desc, table, severity)


def send_alert(result: DQResult):
    """Send alert — customize with Slack webhook, email, etc."""
    # TODO: Replace with real notification (Slack, PagerDuty, etc.)
    logger.critical(f"DQ ALERT: {result}")


# ─── Freshness checks ─────────────────────────────────────────────────────────

def check_freshness(conn, table: str, timestamp_col: str, max_age_hours: int) -> DQResult:
    """Is the data within the SLA window? table/timestamp_col are trusted identifiers."""
    def _check():
        row = conn.sql(f"""
            SELECT MAX({timestamp_col}) AS latest
            FROM {table}
        """).fetchone()
        latest = row[0]
        if latest is None:
            return False, "NULL (no data)"
        # Make tz-aware (assume UTC if naive) — avoids TypeError on naive timestamps
        if hasattr(latest, "tzinfo") and not latest.tzinfo:
            latest = latest.replace(tzinfo=timezone.utc)
        elif not hasattr(latest, "tzinfo"):
            latest = datetime.fromisoformat(str(latest)).replace(tzinfo=timezone.utc)
        age_hours = (datetime.now(timezone.utc) - latest).total_seconds() / 3600
        return age_hours <= max_age_hours, f"{age_hours:.1f}h ago"

    return run_check(
        name=f"freshness_{timestamp_col}",
        table=table,
        fn=_check,
        expected_desc=f"<= {max_age_hours}h",
        severity="error"
    )


# ─── Volume checks ────────────────────────────────────────────────────────────

def check_volume(conn, table: str, partition_col: str, partition_value: str,
                 min_rows: int, max_rows: int) -> DQResult:
    """Is the row count within normal range? Uses a bound parameter for partition_value."""
    def _check():
        # partition_col/table are trusted identifiers; partition_value is bound (no injection)
        row = conn.execute(
            f"SELECT COUNT(*) FROM {table} WHERE {partition_col} = ?",
            [partition_value],
        ).fetchone()
        count = row[0]
        return min_rows <= count <= max_rows, count

    return run_check(
        name=f"volume_{partition_value}",
        table=table,
        fn=_check,
        expected_desc=f"between {min_rows} and {max_rows}",
    )


# ─── Schema drift check ───────────────────────────────────────────────────────

def check_schema_drift(conn, table: str, expected_columns: list[str]) -> DQResult:
    """Has the source added/removed/renamed a column?"""
    def _check():
        actual_cols = {row[0] for row in conn.sql(f"DESCRIBE {table}").fetchall()}
        expected = set(expected_columns)
        added = actual_cols - expected
        removed = expected - actual_cols
        if added or removed:
            return False, f"added={list(added)}, removed={list(removed)}"
        return True, "no drift"

    return run_check(
        name="schema_drift",
        table=table,
        fn=_check,
        expected_desc=f"exactly {expected_columns}",
        severity="warning"
    )


# ─── Distribution checks ──────────────────────────────────────────────────────

def check_column_range(conn, table: str, column: str,
                       min_val: float, max_val: float) -> DQResult:
    """Are values within valid range?"""
    def _check():
        row = conn.sql(f"""
            SELECT
                MIN({column}) AS min_val,
                MAX({column}) AS max_val,
                COUNT(*) FILTER (WHERE {column} < {min_val} OR {column} > {max_val}) AS out_of_range
            FROM {table}
        """).fetchone()
        out_of_range = row[2]
        return out_of_range == 0, f"min={row[0]:.4f}, max={row[1]:.4f}, out_of_range={out_of_range}"

    return run_check(
        name=f"range_{column}",
        table=table,
        fn=_check,
        expected_desc=f"all values in [{min_val}, {max_val}]",
    )


# ─── Run all checks ───────────────────────────────────────────────────────────

def run_all_checks(conn, run_date: str = None) -> dict:
    if not run_date:
        run_date = datetime.now(timezone.utc).strftime("%Y-%m-%d")

    results = []

    # Freshness checks — thresholds from contracts/
    results.append(check_freshness(
        conn, table="fct_candles", timestamp_col="_loaded_at",
        max_age_hours=25  # Daily source → alert if >25h (25h = weekend buffer)
    ))

    # Volume checks — thresholds from historical data
    results.append(check_volume(
        conn, table="fct_candles",
        partition_col="trade_date", partition_value=run_date,
        min_rows=8000,   # NYSE+NASDAQ typically >9000 stocks, lowest day ~8000
        max_rows=12000,  # Upper cap to detect data duplication
    ))

    # Schema drift — from contract schema
    results.append(check_schema_drift(
        conn, table="fct_candles",
        expected_columns=["candle_id", "company_id", "time_id", "open", "high", "low", "close", "volume"]
    ))

    # Distribution — domain knowledge
    results.append(check_column_range(
        conn, table="fct_candles",
        column="close", min_val=0.01, max_val=100000  # Stock price: $0.01 - $100k
    ))

    # Summary
    passed = sum(1 for r in results if r.passed)
    failed = sum(1 for r in results if not r.passed)
    logger.info(f"DQ Summary: {passed} passed, {failed} failed")

    summary = {
        "run_date": run_date,
        "passed": passed,
        "failed": failed,
        "results": [str(r) for r in results],
    }

    # Persist a human-readable report (referenced by the DAG and DONE WHEN)
    report_lines = [
        f"# Data Quality Report — {run_date}",
        f"Passed: {passed} | Failed: {failed}",
        "",
    ]
    for r in results:
        report_lines.append(f"- {r!r}")
    Path("docs").mkdir(exist_ok=True)
    Path("docs/dq_report.md").write_text("\n".join(report_lines))

    return summary


if __name__ == "__main__":
    import duckdb
    conn = duckdb.connect("data/warehouse.duckdb")
    results = run_all_checks(conn)
    print(json.dumps(results, indent=2))

DONE WHEN

• Freshness check exists for every table with threshold from contract SLA
• Volume check exists with min/max from historical data (not arbitrary numbers)
• Schema drift check compares against contract schema
• Distribution check for important numeric columns with range from domain knowledge
• Every failed check → log ERROR and trigger alert
• docs/dq_report.md has check results with threshold rationale

Next Step

Previous: /test. After done → run /contract-check to validate actual pipeline data against source contracts.

References

• Template: skills/dq/assets/dq_checks_template.md
• Previous skill: skills/test/SKILL.md
• Next skill: skills/contract-check/SKILL.md
• Phase deep-dive: phases/phase-6-data-quality.md
• Quality patterns: implementation/quality/data-quality-patterns.md