Why Data Quality Validation Matters
AI agents are only as reliable as the data they process. This tutorial shows you how to implement comprehensive data quality validation rules for your evaluation agent, ensuring accurate indicator tracking, reliable reporting, and trustworthy insights.

What is Data Quality?
Data quality refers to the condition of a dataset based on factors like accuracy, completeness, consistency, and timeliness. For M&E professionals, poor data quality leads to incorrect conclusions, ineffective programs, and wasted resources.

# validation_rules.py
# Define all validation rules in one place for easy maintenance
VALIDATION_RULES = {
    'indicator_data': {
        # Fields that must be present in every record
        'required_fields': ['indicator_name', 'target', 'actual', 'date'],
        
        # Expected data types for each field
        'field_types': {
            'indicator_name': str,
            'target': float,
            'actual': float,
            'date': str,
            'status': str,
            'category': str
        },
        
        # Numeric ranges for values
        'value_ranges': {
            'target': {'min': 0, 'max': 1000000},
            'actual': {'min': 0, 'max': 1000000},
            'achievement': {'min': 0, 'max': 100}
        },
        
        # Allowed values for categorical fields
        'allowed_values': {
            'status': ['on_track', 'off_track', 'at_risk', 'completed', 'delayed'],
            'category': ['health', 'education', 'livelihood', 'governance', 'other']
        },
        
        # Expected date format
        'date_format': '%Y-%m-%d',
        
        # Business logic rules
        'logical_rules': [
            {
                'rule': 'actual_not_exceed_target',
                'condition': 'actual <= target', 'severity': 'warning' }, { 'rule': 'positive_values', 'condition': 'target > 0 AND actual >= 0',
                'severity': 'error'
            },
            {
                'rule': 'achievement_calculation',
                'condition': 'achievement == (actual/target)*100',
                'severity': 'warning'
            }
        ]
    }
}

import pandas as pd
from datetime import datetime
import re
class DataValidator:
    """Main validation class that checks data quality across multiple dimensions."""
    
    def __init__(self, rules=VALIDATION_RULES):
        self.rules = rules
        self.errors = []      # Critical issues that prevent processing
        self.warnings = []    # Issues that should be reviewed
        self.fixes = []       # Automatic corrections applied
    
    def validate_record(self, record):
        """Run all validation checks on a single record."""
        self.errors = []
        self.warnings = []
        self.fixes = []
        
        rules = self.rules['indicator_data']
        
        # Check 1: Required fields
        missing = [f for f in rules['required_fields'] if f not in record or record[f] is None]
        if missing:
            self.errors.append(f"Missing fields: {', '.join(missing)}")
            return False
        
        # Check 2: Data types
        for field, expected_type in rules['field_types'].items():
            if field in record and record[field] is not None:
                if not isinstance(record[field], expected_type):
                    try:
                        if expected_type == float:
                            record[field] = float(record[field])
                            self.fixes.append(f"Converted {field} to number")
                        elif expected_type == str:
                            record[field] = str(record[field])
                    except (ValueError, TypeError):
                        self.errors.append(f"{field} should be {expected_type.__name__}")
                        return False
        
        # Check 3: Value ranges
        for field, range_rules in rules['value_ranges'].items():
            if field in record and record[field] is not None:
                try:
                    value = float(record[field])
                    if value < range_rules['min'] or value > range_rules['max']:
                        self.errors.append(f"{field} = {value} outside allowed range")
                        return False
                except (ValueError, TypeError):
                    self.errors.append(f"Invalid number format in {field}")
                    return False
        
        # Check 4: Allowed values
        for field, allowed_values in rules['allowed_values'].items():
            if field in record and record[field] is not None:
                value = str(record[field]).lower().strip()
                allowed_lower = [str(v).lower() for v in allowed_values]
                if value not in allowed_lower:
                    self.warnings.append(f"{field} = '{record[field]}' not in allowed list")
                    # Auto-correct if possible
                    if value == 'on track':
                        record[field] = 'on_track'
                        self.fixes.append(f"Corrected status to on_track")
        
        # Check 5: Date format
        if 'date' in record and record['date']:
            try:
                datetime.strptime(str(record['date']), rules['date_format'])
            except ValueError:
                self.errors.append(f"Date '{record['date']}' format should be YYYY-MM-DD")
                return False
        
        # Check 6: Logical rules
        if 'target' in record and 'actual' in record:
            try:
                target = float(record['target'])
                actual = float(record['actual'])
                
                if target > 0:
                    # Calculate achievement
                    achievement = round((actual / target) * 100, 1)
                    if 'achievement' not in record or record['achievement'] != achievement:
                        record['achievement'] = achievement
                        self.fixes.append(f"Calculated achievement: {achievement}%")
                    
                    # Check if achievement exceeds target
                    if actual > target:
                        self.warnings.append(f"Actual ({actual}) exceeds target ({target})")
                else:
                    self.errors.append("Target must be greater than 0")
                    return False
            except (ValueError, TypeError):
                self.errors.append("Invalid target or actual values")
                return False
        
        return True

class DataQualityPipeline:
    """Orchestrates the data quality validation workflow."""
    
    def __init__(self, validator):
        self.validator = validator
        self.valid_records = []
        self.invalid_records = []
        self.quality_stats = {
            'total': 0,
            'valid': 0,
            'invalid': 0,
            'error_types': {},
            'warning_types': {},
            'fixes_applied': 0
        }
    
    def process_records(self, records):
        """Process all records through validation pipeline."""
        self.valid_records = []
        self.invalid_records = []
        self.quality_stats = {'total': len(records), 'valid': 0, 'invalid': 0, 
                            'error_types': {}, 'warning_types': {}, 'fixes_applied': 0}
        
        for record in records:
            is_valid = self.validator.validate_record(record)
            
            if is_valid:
                self.valid_records.append(record)
                self.quality_stats['valid'] += 1
            else:
                self.invalid_records.append({
                    'record': record,
                    'errors': self.validator.errors,
                    'warnings': self.validator.warnings
                })
                self.quality_stats['invalid'] += 1
            
            # Track statistics
            for error in self.validator.errors:
                error_type = error.split(':')[0] if ':' in error else 'unknown'
                self.quality_stats['error_types'][error_type] = \
                    self.quality_stats['error_types'].get(error_type, 0) + 1
            
            for warning in self.validator.warnings:
                warning_type = warning.split(':')[0] if ':' in warning else 'unknown'
                self.quality_stats['warning_types'][warning_type] = \
                    self.quality_stats['warning_types'].get(warning_type, 0) + 1
            
            self.quality_stats['fixes_applied'] += len(self.validator.fixes)
        
        return self.quality_stats
    
    def generate_quality_report(self):
        """Create a human-readable quality report."""
        stats = self.quality_stats
        total = stats['total']
        
        if total == 0:
            return "No data to analyze."
        
        report = f"""
        ========================================
        DATA QUALITY REPORT
        ========================================
        
        Total Records: {total}
        Valid Records: {stats['valid']} ({round(stats['valid']/total*100, 1)}%)
        Invalid Records: {stats['invalid']} ({round(stats['invalid']/total*100, 1)}%)
        
        Error Types:
        """
        
        if stats['error_types']:
            for error_type, count in stats['error_types'].items():
                report += f"\n  - {error_type}: {count}"
        else:
            report += "\n  None found"
        
        report += f"\n\nWarning Types:"
        if stats['warning_types']:
            for warning_type, count in stats['warning_types'].items():
                report += f"\n  - {warning_type}: {count}"
        else:
            report += "\n  None found"
        
        report += f"\n\nAuto-fixes Applied: {stats['fixes_applied']}"
        report += f"\n========================================\n"
        
        return report
    
    def get_valid_records(self):
        """Return only validated records."""
        return self.valid_records
    
    def get_invalid_records(self):
        """Return invalid records with their issues."""
        return self.invalid_records

# quality_agent.py
import pandas as pd
import os
class QualityValidationAgent:
    """Main agent that combines data loading, validation, and reporting."""
    
    def __init__(self, source='google_sheets', file_path='indicator_data.xlsx'):
        self.source = source
        self.file_path = file_path
        self.validator = DataValidator()
        self.pipeline = DataQualityPipeline(self.validator)
        self.results = None
    
    def load_data(self):
        """Load data from source (Google Sheets or Excel)."""
        if self.source == 'google_sheets':
            # Google Sheets loading (add your credentials)
            print("Loading from Google Sheets...")
            # Code for Google Sheets loading
            return []
        else:
            print(f"Loading from {self.file_path}...")
            if not os.path.exists(self.file_path):
                raise FileNotFoundError(f"File not found: {self.file_path}")
            df = pd.read_excel(self.file_path)
            return df.to_dict('records')
    
    def validate_and_clean(self):
        """Run validation and cleaning on loaded data."""
        # Load data
        data = self.load_data()
        
        if not data:
            print("No data loaded.")
            return None
        
        print(f"Loaded {len(data)} records.")
        
        # Process through pipeline
        stats = self.pipeline.process_records(data)
        
        # Generate report
        print(self.pipeline.generate_quality_report())
        
        # Store results
        self.results = {
            'stats': stats,
            'valid_records': self.pipeline.get_valid_records(),
            'invalid_records': self.pipeline.get_invalid_records()
        }
        
        return self.results
    
    def get_clean_data(self):
        """Return validated data ready for AI analysis."""
        if self.results and self.results.get('valid_records'):
            return self.results['valid_records']
        return None
    
    def get_quality_report(self):
        """Return quality report as dictionary."""
        if self.results:
            return self.results['stats']
        return None
    
    def save_quality_report(self, filename='quality_report.md'):
        """Save quality report to file."""
        if self.results:
            report = self.pipeline.generate_quality_report()
            with open(filename, 'w') as f:
                f.write(report)
            print(f"Quality report saved to {filename}")

# Sample health program data with common quality issues
health_data = [
    {
        'indicator_name': 'Children Vaccinated',
        'target': 5000,
        'actual': 4800,
        'date': '2024-01-15',
        'status': 'on_track',
        'category': 'health'
    },
    {
        'indicator_name': 'Health Centers Visited',
        'target': '50',  # String instead of number (should be integer)
        'actual': 45,
        'date': '2024-01-15',
        'status': 'good',  # Not in allowed values
        'category': 'health'
    },
    {
        'indicator_name': 'Community Health Workers',
        'target': 100,
        'actual': 120,  # Exceeds target (should be flagged)
        'date': '2024-01-15',
        'status': 'on_track',
        'category': 'health'
    },
    {
        'indicator_name': 'Health Outreach Events',
        'target': 200,
        'actual': 85,
        'date': '2024-01-15',
        'category': 'health'  # Missing status field
    }
]
# Run validation
validator = DataValidator()
for record in health_data:
    is_valid = validator.validate_record(record)
    if is_valid:
        print(f"✓ Valid: {record['indicator_name']}")
    else:
        print(f"✗ Invalid: {record['indicator_name']}")
        print(f"  Errors: {validator.errors}")
        print(f"  Warnings: {validator.warnings}")
        if validator.fixes:
            print(f"  Fixes: {validator.fixes}")

# Complete workflow for education program data
def process_education_data():
    agent = QualityValidationAgent(source='excel', file_path='education_data.xlsx')
    
    # Run validation
    results = agent.validate_and_clean()
    
    if results:
        # Get clean data for AI analysis
        clean_data = agent.get_clean_data()
        print(f"\nReady for analysis: {len(clean_data)} records")
        
        # Save quality report
        agent.save_quality_report('education_quality_report.md')
        
        # Now use with your AI agent
        for record in clean_data[:2]:  # Show first 2 records
            print(f"\nClean Record:")
            for key, value in record.items():
                print(f"  {key}: {value}")
    
    return results
# Run the workflow
if __name__ == "__main__":
    # First, create sample data
    sample_data = pd.DataFrame({
        'indicator_name': ['Student Enrollment', 'Teacher Training', 'School Supplies'],
        'target': [5000, 200, 1000],
        'actual': [4800, 195, 850],
        'date': ['2024-01-15', '2024-01-15', '2024-01-15'],
        'status': ['on_track', 'on_track', 'at_risk'],
        'category': ['education', 'education', 'education']
    })
    sample_data.to_excel('education_data.xlsx', index=False)
    
    # Process data
    process_education_data()

def check_duplicates(records, field='indicator_name'):
    """Check for duplicate records based on a specific field."""
    seen = set()
    duplicates = []
    for record in records:
        key = record.get(field)
        if key in seen:
            duplicates.append(key)
        seen.add(key)
    return duplicates
def check_date_consistency(records, date_field='date'):
    """Check for date consistency across records."""
    dates = []
    for record in records:
        if record.get(date_field):
            try:
                date = datetime.strptime(str(record[date_field]), '%Y-%m-%d')
                dates.append(date)
            except ValueError:
                pass
    
    if dates:
        min_date = min(dates)
        max_date = max(dates)
        return {
            'oldest': min_date.strftime('%Y-%m-%d'),
            'newest': max_date.strftime('%Y-%m-%d'),
            'span_days': (max_date - min_date).days
        }
    return None

def clean_numeric_value(value):
    """Clean string values to extract numbers."""
    if isinstance(value, (int, float)):
        return value
    if isinstance(value, str):
        # Remove currency symbols, commas, and spaces
        cleaned = re.sub(r'[^\d.]', '', value)
        try:
            return float(cleaned) if cleaned else 0
        except ValueError:
            return 0
    return 0
def auto_clean_records(records):
    """Apply automatic cleaning to common issues."""
    cleaned = []
    for record in records:
        new_record = record.copy()
        
        # Clean numeric fields
        for field in ['target', 'actual']:
            if field in new_record:
                new_record[field] = clean_numeric_value(new_record[field])
        
        # Normalize status
        if 'status' in new_record:
            status = str(new_record['status']).lower().strip()
            status_map = {
                'on track': 'on_track', 'ontrack': 'on_track', 'on-track': 'on_track',
                'at risk': 'at_risk', 'atrisk': 'at_risk',
                'off track': 'off_track', 'offtrack': 'off_track',
                'complete': 'completed', 'done': 'completed'
            }
            new_record['status'] = status_map.get(status, status)
        
        cleaned.append(new_record)
    
    return cleaned

Master Data Quality for AI Agents

The AI Agents for Evaluators Certificate teaches you to build reliable AI agents with robust data quality validation, ensuring trustworthy insights for your M&E work.

Course Features: 32 lectures · Lifetime access · Certificate included · 241 students enrolled

Enroll Now – $249 Lifetime Access

Bundle with AI in M&E course and save 30%