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