Magic Data Validation

Natural Language Triggers

Activate this skill when the user says things like:

• "validate this data" / "check the schema"
• "verify data quality" / "run sanity checks"
• "check constraints" / "enforce the schema"
• "are there any data issues left?"
• "check for pitfalls" / "detect sentinels" / "verify schema" / "is this data ready for modeling" / "content quality check"

These produce the SAME behavior as the validation workflow.

When to Use

• After data cleaning, verify data quality
• Need to infer or enforce a schema
• Need cross-column constraint checking
• Need to detect statistical pitfalls (join explosion, Simpson's paradox, etc.)

When NOT to Use: Use magic-data-cleaning to fix issues; use magic-data-profiling for exploration.

Data Processing Expertise

Thinking

Before validating, ask:

• Fitness for what use? — Validation criteria depend on downstream use. A dataset feeding a dashboard has different quality thresholds than one training a model. Define what "valid" means for THIS use case before writing a single rule.
• What's the cost of false positives vs. false negatives? — Overly strict validation rejects good data (FP); overly lenient validation lets bad data through (FN). A pipeline that drops 30% of valid records because of one tight constraint is worse than one that lets a few edge cases through. Calibrate strictness to the use case.
• Am I validating clean or dirty data? — Validating before cleaning catches issues to fix; validating after cleaning verifies fixes worked. Both are valuable but serve different purposes. Know which you are doing.
• Does the inferred schema match the target schema? — An inferred schema reflects what IS in the data, not what SHOULD BE. A column inferred as text may be a malformatted numeric. Review and adjust inferred schemas against the intended target schema before enforcing them.
• What validation ordering? — Run checks from cheapest to most expensive: schema (structural) first, then constraints (rule-based), then content (semantic), then sanity (statistical). Each layer catches issues the previous one misses.

Rules

• Validation ordering: Always follow schema -> constraints -> content -> sanity. Schema validation is fast and catches structural issues. Content validation is slower and catches semantic issues. Running sanity checks before fixing schema violations wastes effort on data that will change.
• Fitness-for-use drives strictness: A model training dataset can tolerate 1-2% nulls in non-target columns; a financial reporting dataset cannot tolerate any. Define acceptable thresholds per column per use case, not globally.
• Inferred schema != target schema: Inferred schemas describe what exists. Target schemas describe what should exist. Always compare inferred against target and reconcile before enforcing. A column of zip codes inferred as numeric will lose leading zeros if enforced as integers.
• Constraint types: Range constraints (min/max for numerics), enum constraints (allowed values for categoricals), pattern constraints (regex for text like emails/phones), cross-column constraints (start_date constraints -> content -> sanity -- each layer depends on the previous being clean
• MUST always run sanity_check on analysis results to catch statistical pitfalls
• MUST validate after any transformation -- transformations introduce new failure modes (type coercion errors, join explosions, aggregation artifacts)
• MUST NOT infer schema from a sample and enforce in strict mode -- sampling bias produces schemas that silently reject valid production data; always infer from the full dataset, or review and loosen constraints manually before enforcing
• MUST NOT skip content validation on text columns just because schema passes -- schema checks type and length; it will not catch "N/A", "TBD", "placeholder", or single-space strings
• NEVER apply numeric validation rules to text columns -- this produces 100% failure rates that look like catastrophic data quality; check column types first; use content validation for text columns
• NEVER ignore pitfall warnings from sanity_check -- Simpson's paradox and join explosion silently invalidate downstream analysis; investigate every warning even if the data "looks fine" in aggregate
• NEVER infer schema from only the first few rows -- sampling bias produces a schema that does not represent the full dataset; use infer_schema.py on the full dataset or a representative stratified sample

Seed Patterns

Schema inference

import pandas as pd
import numpy as np

def infer_schema(df: pd.DataFrame, strict: bool = False) -> dict:
    schema = {}
    for col in df.columns:
        s = df[col].dropna()
        if pd.api.types.is_numeric_dtype(df[col]):
            dtype = "numeric"
            constraints = {"min": float(s.min()), "max": float(s.max())}
        elif s.nunique() / max(len(s), 1)  list[dict]:
    violations = []
    for col, spec in schema.items():
        if col not in df.columns:
            violations.append({"column": col, "issue": "missing_column"})
            continue
        if not spec.get("nullable", True) and df[col].isnull().any():
            null_count = int(df[col].isnull().sum())
            violations.append({"column": col, "issue": "unexpected_nulls",
                               "count": null_count})
        if spec["dtype"] == "numeric" and not pd.api.types.is_numeric_dtype(df[col]):
            violations.append({"column": col, "issue": "type_mismatch",
                               "expected": "numeric", "actual": str(df[col].dtype)})
    return violations

Constraint checking (range, enum, pattern)

import pandas as pd
import re

def check_constraints(df: pd.DataFrame, schema: dict) -> list[dict]:
    violations = []
    for col, spec in schema.items():
        if col not in df.columns:
            continue
        c = spec.get("constraints", {})
        s = df[col].dropna()
        if "min" in c and pd.api.types.is_numeric_dtype(s):
            below = s[s  c["max"]]
            if len(above):
                violations.append({"column": col, "rule": "max",
                                   "count": len(above), "sample": above.head(5).tolist()})
        if "allowed_values" in c:
            invalid = s[~s.isin(c["allowed_values"])]
            if len(invalid):
                violations.append({"column": col, "rule": "enum",
                                   "count": len(invalid), "sample": invalid.head(5).tolist()})
        if "pattern" in c:
            non_match = s[~s.astype(str).str.match(c["pattern"])]
            if len(non_match):
                violations.append({"column": col, "rule": "pattern",
                                   "count": len(non_match), "sample": non_match.head(5).tolist()})
    return violations

Content validation (sentinel and placeholder detection)

import pandas as pd

SENTINELS = {"N/A", "n/a", "NA", "TBD", "TODO", "placeholder", "test",
             "null", "none", "undefined", "missing", "-", "--", "...", " "}

def validate_content(df: pd.DataFrame, text_cols: list[str],
                     min_length: int = 3) -> list[dict]:
    issues = []
    for col in text_cols:
        s = df[col].dropna().astype(str)
        # Sentinel detection
        sentinel_mask = s.isin(SENTINELS)
        if sentinel_mask.any():
            issues.append({"column": col, "issue": "sentinel_values",
                           "count": int(sentinel_mask.sum()),
                           "examples": s[sentinel_mask].unique()[:5].tolist()})
        # Short content detection
        short_mask = s.str.len()  list[dict]:
    warnings = []
    row_count = len(df)
    for col in df.select_dtypes(include=[np.number]).columns:
        s = df[col].dropna()
        if len(s) == 0:
            continue
        # Order-of-magnitude outlier
        if s.median() != 0 and s.max() > 1000 * s.median():
            warnings.append({"pitfall": "magnitude_outlier", "column": col,
                             "detail": f"max ({s.max():.2f}) > 1000x median ({s.median():.2f})"})
    # Join explosion risk: near-unique columns
    for col in df.columns:
        if df[col].nunique() / row_count > 0.95 and row_count > 100:
            warnings.append({"pitfall": "join_explosion", "column": col,
                             "detail": f"{df[col].nunique()}/{row_count} unique values"})
    # Simpson's paradox: check if correlation reverses within groups
    num_cols = df.select_dtypes(include=[np.number]).columns.tolist()
    cat_cols = df.select_dtypes(include=["object", "category"]).columns.tolist()
    if len(num_cols) >= 2 and cat_cols:
        c1, c2 = num_cols[0], num_cols[1]
        overall = df[[c1, c2]].corr().iloc[0, 1]
        for gc in cat_cols[:2]:
            if 2  20:
                        group_corr = g[[c1, c2]].corr().iloc[0, 1]
                        if not np.isnan(group_corr) and np.sign(overall) != np.sign(group_corr):
                            warnings.append({"pitfall": "simpsons_paradox",
                                             "columns": [c1, c2], "group": gc})
                            break
    return warnings

Reference Scripts

> Scripts fall into three categories: callable tools (call directly via CLI), > scriptable tools (call directly for standard use, or read + adapt for advanced use), > and reference implementations (always read + adapt).

Scriptable tools -- call directly for standard use, read + adapt for advanced:

| Script | Tier | Standard CLI usage | When to customize | |--------|------|-------------------|-------------------| | infer_schema.py | A | python3 infer_schema.py --input data.csv --output schema.json | --strict for p5-p95 bounds and ±10% row range (production schema gates) | | content_validator.py | A | python3 content_validator.py data.csv report.json | --distribution-check for length variance anomalies; --group-by col for per-group breakdown; --depth deep for thorough inspection; --sentinel-values X,TODO for custom list | | validate_schema.py | B | python3 validate_schema.py --input data.csv --schema schema.json --output report.json | --explain for per-violation explanations. Schema from infer_schema.py or hand-authored | | sanity_check.py | A | python3 sanity_check.py --input data.csv --output sanity.json | No selective detector flags — runs all 7 pitfalls (join explosion, survivorship bias, Simpson's paradox, look-ahead, selection bias, metric gaming, ecological fallacy) |

Reference implementations -- read patterns, write custom code:

| Script | Demonstrates | Key pattern | |--------|-------------|-------------| | check_constraints.py | Cross-column constraint checking | Typed constraint dispatching (comparison, vocabulary, conditionalnull, uniquetogether); requires hand-authored constraints JSON | | validate_statistics.py | Internal consistency of statistical results | Requires cross-skill artifact (descriptive_stats.py JSON output); tolerance-gated stat comparison |

Checkpointing

Checkpointing is a judgment call — save intermediate results when the cost of re-running the preceding step justifies persisting the result.

For validation specifically:

| Signal | Checkpoint? | Reasoning | |--------|-------------|-----------| | After schema validation | Yes | Captures validation state; report is a deliverable | | After constraint checking | Yes | Constraint violations are the output — save them | | After content validation (sentinel/placeholder detection) | Yes | Results inform cleaning decisions | | After sanity checks (Simpson's paradox, join explosion) | Yes | Sanity findings are expensive to reproduce | | Quick inline validation before next step | Skip | Recovery is trivial |

Suggested names: validated_schema.json, constraint_report.json, content_issues.json, sanity_report.json

Include metadata: source file validated, timestamp, validation type, pass/fail summary.

Save pattern:

import json, datetime
from pathlib import Path

def save_checkpoint(df, path: str, metadata: dict = None) -> str:
    """Save data + provenance metadata. Use at phase boundaries."""
    p = Path(path)
    p.parent.mkdir(parents=True, exist_ok=True)
    if p.suffix == ".parquet":
        df.to_parquet(p, index=False)
    elif p.suffix == ".jsonl":
        df.to_json(p, orient="records", lines=True, force_ascii=False)
    else:
        df.to_csv(p, index=False)
    meta = {
        "rows": len(df), "cols": list(df.columns),
        "saved_at": datetime.datetime.utcnow().isoformat(),
        **(metadata or {}),
    }
    p.with_suffix(".meta.json").write_text(
        json.dumps(meta, indent=2, ensure_ascii=False)
    )
    return str(p)

Self-Healing

If validation fails unexpectedly:

1. Read error output -- check for suggestion field in JSON reports
2. Read the relevant script source to understand the failure mode
3. Inspect the data with sampling before re-running

| Error | Likely Cause | Fix | |-------|-------------|-----| | Schema mismatch | Data changed after schema inferred | Re-infer schema on current data | | Missing constraint file | Not generated yet | Run infer_schema.py --strict first | | 100% column failure rate | Numeric rules applied to text column | Check column dtype; use contentvalidator for text | | All rows flagged as short content | CJK text with min_content_length=3 | Enable --cjk-aware (default) which uses min_content_length=1 for CJK-dominant columns | | Constraint file format error | Hand-edited JSON with syntax error | Validate JSON before passing to checkconstraints |

Reference Guides

| Topic | File | Load When | |-------|------|-----------| | Common pitfalls | references/common_pitfalls.md | Pitfall detected in sanity_check | | Schema format | references/schema_format.md | Creating custom schemas | | Domain validation | references/domain_validation.md | MANDATORY -- before defining constraints or reviewing schema |

Do NOT Load references/common_pitfalls.md unless sanity_check.py flags a specific pitfall. Do NOT Load references/schema_format.md when using inferred schemas -- only needed when creating custom schemas from scratch.

Triggered by Findings

This skill is often invoked after findings presentation or as part of the lifecycle validation phase. When routed here from a findings decision:

| Findings Action | Validation Approach | |----------------|-------------------| | Schema issues -> VALIDATE | Schema inference + validation against target | | Constraint violations | Constraint checking with domain-specific rules | | Post-transform verification | Sanity check to detect pitfalls | | Content quality check | Content validation for sentinel and content length scanning |

Validation often triggers further routing -- schema issues may route to magic-data-transformation, content issues may route to magic-data-cleaning or magic-data-synthesis.

Interactive Mode [Optional]

For agents working interactively with a user. Pipeline code generation skips this section.

PAUSE Gates

• If schema inference produces type warnings (ambiguous column types, nullable conflicts, unexpected inferred types), present warnings and proposed schema to user for review before enforcing validation.

Workspace Integration

• Update workspace_state.md with validation results.
• Log validation decisions in logs/analysis_journal.md.
• Store all validation reports in logs/ alongside the data they validate. Name reports descriptively (e.g., validation_post_cleaning.json, schema_check_v2.json) so the agent can track validation history across the lifecycle.

Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

• Author: Votee-AI
• Source: Votee-AI/magic-data-agent-skills
• License: Apache-2.0
• Homepage: https://docs.votee.ai/magic-agent-skills/data-agent/getting-started

Install and usage instructions live in the source repository linked above.