data-quality-monitoring

Data Quality Monitoring

Trust your data, or fix it before anyone else sees it.

Quality Dimensions

Dimension	Definition	Measurement
Completeness	All expected data present	Row count, null ratio
Consistency	Data follows expected patterns	Schema match, value ranges
Timeliness	Data arrives on schedule	Freshness, latency
Accuracy	Data reflects reality	Comparison to source
Uniqueness	No unexpected duplicates	Distinct key ratio

Anomaly Detection

Row Count Monitoring (Z-Score)

def check_row_count(source: str, current_count: int,
                    expected: float, stddev: float,
                    threshold_pct: float = 0.25) -> dict:
    """Detect row count anomalies using rolling baseline."""
    z_score = abs(current_count - expected) / max(stddev, 1)
    deviation_pct = abs(current_count - expected) / max(expected, 1)

    return {
        "source": source,
        "current": current_count,
        "expected": round(expected),
        "deviation_pct": round(deviation_pct, 4),
        "z_score": round(z_score, 2),
        "is_anomaly": deviation_pct > threshold_pct,
        "severity": _severity(deviation_pct),
    }

def _severity(deviation: float) -> str:
    if deviation > 0.5: return "CRITICAL"
    if deviation > 0.25: return "HIGH"
    if deviation > 0.1: return "MEDIUM"
    return "LOW"

Severity Actions

Severity	Deviation	Action
CRITICAL	> 50%	BLOCK — Manual review required
HIGH	> 25%	WARN — Review before proceeding
MEDIUM	> 10%	LOG — Monitor trend
LOW	≤ 10%	PASS — Within normal range

Schema Drift Detection

def detect_schema_drift(actual_cols: dict, expected_cols: dict,
                        strict: bool = False) -> dict:
    """Compare actual vs expected schema."""
    missing = set(expected_cols) - set(actual_cols)
    added = set(actual_cols) - set(expected_cols)
    type_changes = {
        col: (expected_cols[col], actual_cols[col])
        for col in set(actual_cols) & set(expected_cols)
        if expected_cols[col] != actual_cols[col]
    }

    return {
        "has_drift": bool(missing or (strict and added) or type_changes),
        "missing_columns": list(missing),
        "added_columns": list(added),
        "type_changes": type_changes,
        "severity": "HIGH" if missing or type_changes else "LOW",
    }