data-quality-check

Skill: Data Quality Check

Purpose

Validate data completeness, consistency, and coverage before any analysis begins, flagging issues with severity ratings so the analyst knows what blocks analysis vs. what to note as a caveat.

When to Use

Apply this skill at the start of every new analysis, when connecting to a new data source, or when results look suspicious. Run quality checks BEFORE drawing conclusions from data.

Also fires on table-scoped questions. Any question that names a specific table ("tell me about {table}", "describe {table}", "what's in {table}", "show me {table}") triggers this skill. Schema-only answers are insufficient — pair the schema description with a minimum DQ probe:

• Row count
• Null rate per column (flag anything >5%)
• Date range on the primary timestamp column
• Duplicate check on the primary key
• Surface anything from .knowledge/datasets/{active}/quirks.md for that table

If the table is large enough that probing is expensive (>100M rows or warehouse cost concerns), tell the user and ask before running the full probe — but always run at minimum row count + PK duplicate check.

Instructions

Primary method — run the named structural validators

Do not hand-roll the core checks as ad-hoc SQL. Query the rows once, then run the tested validators in helpers/validation/structural_validator.py, so the checks are identical every time and can't be skipped or mis-written. The validators operate on a DataFrame, so pull the row-level slice you're about to analyze with the repo connection first:

from helpers.data.connection_manager import ConnectionManager
from helpers.validation.structural_validator import run_structural_checks

cm = ConnectionManager(); cm.connect()
df = cm.query("select * from orders where order_date >= '2024-12-01'")   # the slice under analysis

result = run_structural_checks(df, {
    "primary_key": ["ORDER_ID"],                         # uniqueness + nulls
    "required_columns": ["TOTAL_AMOUNT", "STATUS"],      # completeness
    "completeness_threshold": 0.95,
    "date_column": "ORDER_DATE",                         # gap / range
    "value_domain": {"column": "STATUS",
                     "valid_values": ["completed", "cancelled", "returned"]},
    "min_rows": 1,
})
print(result["overall_ok"], result["checks_passed"], "/", result["checks_run"])
for name, d in result["details"].items():
    print(name, "->", "OK" if (d.get("ok") or d.get("valid")) else f"FAIL ({d.get('severity','')})")