Data Pipeline Observability: Why Your AI Model Fails in Production

# pipelines/validation/custom_validators.py

import pandas as pd

from datetime import datetime, timedelta



def validate_feature_engineering(df: pd.DataFrame) -> dict:

    """

    Custom validation for features before they reach the model.

    Returns a dict of validation results.

    """

    results = {}

    

    # Validate 1: Feature completeness

    # We need at least 95% of features populated

    feature_cols = [col for col in df.columns if col.startswith('feature_')]

    null_rate = df[feature_cols].isnull().sum().sum() / (len(df) * len(feature_cols))

    results['feature_completeness'] = {

        'passed': null_rate < 0.05,

        'null_rate': null_rate,

        'threshold': 0.05

    }

    

    # Validate 2: Feature scaling

    # After normalization, features should be roughly between -3 and 3 (3 sigma)

    for col in feature_cols:

        max_val = df[col].max()

        min_val = df[col].min()

        results[f'{col}_scaling'] = {

            'passed': max_val < 10 and min_val > -10,

            'max': max_val,

            'min': min_val

        }

    

    # Validate 3: Temporal consistency

    # Events should be recent (within last 30 days)

    if 'event_date' in df.columns:

        df['event_date'] = pd.to_datetime(df['event_date'])

        days_old = (datetime.now() - df['event_date'].max()).days

        results['temporal_freshness'] = {

            'passed': days_old < 30,

            'days_old': days_old,

            'threshold_days': 30

        }

    

    # Validate 4: Business logic

    # Revenue should always be positive

    if 'revenue' in df.columns:

        negative_revenue = (df['revenue'] < 0).sum()

        results['business_logic_revenue'] = {

            'passed': negative_revenue == 0,

            'negative_count': negative_revenue

        }

    

    return results



def validate_and_alert(df: pd.DataFrame, validation_results: dict) -> bool:

    """

    Check all validations and alert if any fail.

    Returns True if all pass, False otherwise.

    """

    all_passed = True

    

    for check_name, check_result in validation_results.items():

        if not check_result['passed']:

            all_passed = False

            print(f"ALERT: {check_name} failed")

            print(f"Details: {check_result}")

            # Send to monitoring system (Datadog, New Relic, etc.)

            # send_alert(check_name, check_result)

    

    return all_passed