data-quality-check

If the skill install path cannot be resolved (some sandboxed environments): read the script file(s) from this skill, write a copy into a scripts/ folder inside the working folder, and run from there. The scripts are self-contained.

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 scripts/structural_validator.py (bundled with this skill), 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 through the session's active data connection first:

import sys; sys.path.insert(0, "<this skill's scripts/ dir>")
from structural_validator import run_structural_checks

# Load the slice under analysis into a DataFrame through the active connection,
# e.g. duckdb over the mounted files, or the connected warehouse:
df = ...   # select * from orders where order_date >= '2024-12-01'

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','')})")