┌─────────────────────────────────────────────────────────────────┐
│              DATA GOVERNANCE FRAMEWORK                          │
│                                                                 │
│  ┌───────────────────────────────────────────────────────────┐ │
│  │                    GOVERNANCE BODY                         │ │
│  │  Data Governance Council (Executives + Department Heads)  │ │
│  └───────────────────────────────────────────────────────────┘ │
│                              ↓                                   │
│  ┌───────────────────────────────────────────────────────────┐ │
│  │                    POLICIES & STANDARDS                    │ │
│  │  • Data Quality Standards                                 │ │
│  │  • Data Security Policies                                 │ │
│  │  • Data Privacy Policies                                  │ │
│  │  • Data Retention Policies                                │ │
│  └───────────────────────────────────────────────────────────┘ │
│                              ↓                                   │
│  ┌───────────────────────────────────────────────────────────┐ │
│  │                    DATA STEWARDS                          │ │
│  │  Domain experts responsible for data quality              │ │
│  └───────────────────────────────────────────────────────────┘ │
│                              ↓                                   │
│  ┌───────────────────────────────────────────────────────────┐ │
│  │                    IMPLEMENTATION                         │ │
│  │  • Data Quality Monitoring                                │ │
│  │  • Data Lineage Tracking                                  │ │
│  │  • Master Data Management                                 │ │
│  │  • Metadata Management                                    │ │
│  └───────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘

# Example: Data quality monitoring class

class DataQualityMonitor:
    """Monitor data quality for banking datasets."""
    
    def __init__(self, dataset_name):
        self.dataset_name = dataset_name
        self.issues = []
        self.metrics = {}
    
    def check_accuracy(self, df, column, validation_rule):
        """
        Check if data is accurate (validated against a rule).
        
        Args:
            df: DataFrame
            column: Column to check
            validation_rule: Function that returns True for valid values
        """
        valid_count = df[column].apply(validation_rule).sum()
        total_count = len(df)
        accuracy = valid_count / total_count
        
        self.metrics['accuracy'] = accuracy
        
        if accuracy < 0.95:
            self.issues.append({
                'dimension': 'Accuracy',
                'column': column,
                'issue': f'Accuracy is {accuracy:.2%} (threshold: 95%)',
                'invalid_count': total_count - valid_count
            })
    
    def check_completeness(self, df, column, threshold=0.95):
        """Check for missing values."""
        null_count = df[column].isnull().sum()
        total_count = len(df)
        completeness = 1 - (null_count / total_count)
        
        self.metrics[f'completeness_{column}'] = completeness
        
        if completeness < threshold:
            self.issues.append({
                'dimension': 'Completeness',
                'column': column,
                'issue': f'{null_count} missing values ({1-completeness:.2%})',
                'null_count': null_count
            })
    
    def check_consistency(self, df, column1, column2):
        """Check that columns are consistent."""
        inconsistent = df[df[column1] != df[column2]].shape[0]
        consistency = 1 - (inconsistent / len(df))
        
        self.metrics['consistency'] = consistency
        
        if inconsistent > 0:
            self.issues.append({
                'dimension': 'Consistency',
                'issue': f'Inconsistent values between {column1} and {column2}',
                'inconsistent_count': inconsistent
            })
    
    def check_validity(self, df, column, valid_values):
        """Check that values are in a valid set."""
        invalid_count = df[~df[column].isin(valid_values)].shape[0]
        validity = 1 - (invalid_count / len(df))
        
        self.metrics['validity'] = validity
        
        if invalid_count > 0:
            self.issues.append({
                'dimension': 'Validity',
                'column': column,
                'issue': f'{invalid_count} invalid values found',
                'invalid_count': invalid_count
            })
    
    def check_uniqueness(self, df, unique_keys):
        """Check for duplicate records."""
        duplicates = df.duplicated(subset=unique_keys).sum()
        uniqueness = 1 - (duplicates / len(df))
        
        self.metrics['uniqueness'] = uniqueness
        
        if duplicates > 0:
            self.issues.append({
                'dimension': 'Uniqueness',
                'issue': f'{duplicates} duplicate records found based on {unique_keys}',
                'duplicate_count': duplicates
            })
    
    def check_timeliness(self, df, date_column, max_age_days=1):
        """Check that data is timely (not too old)."""
        if pd.api.types.is_datetime64_any_dtype(df[date_column]):
            max_date = df[date_column].max()
            age_days = (pd.Timestamp.now() - max_date).days
            timeliness = max(0, 1 - (age_days / max_age_days))
            
            self.metrics['timeliness'] = timeliness
            
            if age_days > max_age_days:
                self.issues.append({
                    'dimension': 'Timeliness',
                    'issue': f'Data is {age_days} days old (threshold: {max_age_days} days)',
                    'age_days': age_days
                })
    
    def generate_report(self):
        """Generate quality report."""
        print(f"\n{'='*60}")
        print(f"DATA QUALITY REPORT: {self.dataset_name}")
        print(f"{'='*60}")
        
        print("\n📊 METRICS:")
        for metric, value in self.metrics.items():
            if isinstance(value, float):
                print(f"  {metric}: {value:.2%}")
            else:
                print(f"  {metric}: {value}")
        
        if self.issues:
            print(f"\n⚠️ ISSUES FOUND: {len(self.issues)}")
            for idx, issue in enumerate(self.issues, 1):
                print(f"  {idx}. [{issue.get('dimension', 'General')}] {issue['issue']}")
                if 'invalid_count' in issue:
                    print(f"     → {issue['invalid_count']} records affected")
                if 'null_count' in issue:
                    print(f"     → {issue['null_count']} null values")
                if 'duplicate_count' in issue:
                    print(f"     → {issue['duplicate_count']} duplicates")
        else:
            print("\n✅ No issues found! All quality checks passed.")
        
        print(f"\n{'='*60}")
        return self.issues

# Usage example
from datetime import datetime, timedelta
import pandas as pd

# Create sample data
sample_data = pd.DataFrame({
    'customer_id': [1, 2, 3, 4, 5],
    'name': ['John Smith', 'Jane Doe', 'Bob Johnson', 'Alice Williams', 'Charlie Brown'],
    'state': ['CA', 'NY', 'TX', 'CA', 'FL'],
    'credit_score': [780, 720, 680, 820, None],
    'email': ['john@email.com', 'jane@email.com', 'bob@email.com', None, 'charlie@email.com'],
    'last_transaction': [
        pd.Timestamp.now() - timedelta(days=1),
        pd.Timestamp.now() - timedelta(days=0.5),
        pd.Timestamp.now() - timedelta(days=2),
        pd.Timestamp.now() - timedelta(days=3),
        pd.Timestamp.now() - timedelta(days=4)
    ]
})

# Run quality checks
monitor = DataQualityMonitor('customers_sample')
monitor.check_completeness(sample_data, 'credit_score', threshold=0.80)
monitor.check_completeness(sample_data, 'email', threshold=0.95)
monitor.check_validity(sample_data, 'state', ['CA', 'NY', 'TX', 'FL', 'IL'])
monitor.check_uniqueness(sample_data, ['customer_id'])
monitor.check_timeliness(sample_data, 'last_transaction', max_age_days=3)

# Generate report
monitor.generate_report()

# Example: Automated data quality monitoring

class ContinuousQualityMonitor:
    """Continuous monitoring with alerting."""
    
    def __init__(self, db_connection, alert_webhook=None):
        self.conn = db_connection
        self.alert_webhook = alert_webhook
        self.monitoring_history = []
    
    def check_rule(self, rule_name, sql_query, expected_value, threshold_pct=10):
        """
        Check a data quality rule.
        
        Args:
            rule_name: Name of the rule
            sql_query: SQL query that returns a metric
            expected_value: What the metric should be
            threshold_pct: How much deviation is allowed
        """
        try:
            # Run the query
            result = pd.read_sql(sql_query, self.conn)
            actual_value = result.iloc[0, 0]
            
            # Calculate deviation
            deviation_pct = abs((actual_value - expected_value) / expected_value) * 100
            
            # Record history
            self.monitoring_history.append({
                'rule_name': rule_name,
                'timestamp': datetime.now(),
                'actual_value': actual_value,
                'expected_value': expected_value,
                'deviation_pct': deviation_pct,
                'status': 'PASS' if deviation_pct <= threshold_pct else 'FAIL'
            })
            
            # Alert if failed
            if deviation_pct > threshold_pct:
                self.send_alert(rule_name, actual_value, expected_value, deviation_pct)
            
            return deviation_pct <= threshold_pct
            
        except Exception as e:
            self.monitoring_history.append({
                'rule_name': rule_name,
                'timestamp': datetime.now(),
                'status': 'ERROR',
                'error_message': str(e)
            })
            self.send_alert(rule_name, 'ERROR', '', str(e))
            return False
    
    def send_alert(self, rule_name, actual, expected, detail):
        """Send alert via webhook or logging."""
        message = f"""
        ⚠️ DATA QUALITY ALERT
        Rule: {rule_name}
        Expected: {expected}
        Actual: {actual}
        Detail: {detail}
        Time: {datetime.now()}
        """
        print(message)
        # In production, send to Slack, email, etc.
    
    def generate_dashboard(self, days=30):
        """Generate quality dashboard for the last N days."""
        recent = [h for h in self.monitoring_history 
                 if h['timestamp'] > datetime.now() - timedelta(days=days)]
        
        dashboard = {
            'total_checks': len(recent),
            'passed_checks': len([h for h in recent if h.get('status') == 'PASS']),
            'failed_checks': len([h for h in recent if h.get('status') == 'FAIL']),
            'error_checks': len([h for h in recent if h.get('status') == 'ERROR'])
        }
        
        print(f"\n{'='*60}")
        print(f"QUALITY DASHBOARD (Last {days} days)")
        print(f"{'='*60}")
        print(f"Total checks: {dashboard['total_checks']}")
        print(f"✅ Passed: {dashboard['passed_checks']}")
        print(f"❌ Failed: {dashboard['failed_checks']}")
        print(f"⚠️ Errors: {dashboard['error_checks']}")
        
        return dashboard

# Example usage
monitor = ContinuousQualityMonitor(db_connection)

# Define quality rules
rules = [
    ("row_count", "SELECT COUNT(*) FROM transactions", expected=1000000, threshold=5),
    ("null_customer_id", "SELECT COUNT(*) FROM transactions WHERE customer_id IS NULL", expected=0, threshold=0),
    ("negative_amount", "SELECT COUNT(*) FROM transactions WHERE amount < 0", expected=0, threshold=0),
    ("recent_transactions", "SELECT COUNT(*) FROM transactions WHERE transaction_date > '2024-01-01'", expected=5000, threshold=10)
]

for rule in rules:
    monitor.check_rule(*rule)

# Generate dashboard
monitor.generate_dashboard()

Source System          ETL Process              Destination
    ↓                      ↓                       ↓
[OLTP DB] → [Extract] → [Staging] → [Transform] → [DW]
    ↓                      ↓                       ↓
    └──────────────────[Lineage Tracking]─────────┘

# Example: Simple data lineage tracking

class DataLineageTracker:
    """Track data movement and transformations."""
    
    def __init__(self):
        self.lineage = []
    
    def record_extract(self, source, table, rows, timestamp):
        """Record extraction step."""
        self.lineage.append({
            'step': 'extract',
            'source': source,
            'table': table,
            'rows': rows,
            'timestamp': timestamp
        })
    
    def record_transform(self, input_table, output_table, transform_logic):
        """Record transformation step."""
        self.lineage.append({
            'step': 'transform',
            'input_table': input_table,
            'output_table': output_table,
            'transform_logic': transform_logic,
            'timestamp': datetime.now()
        })
    
    def record_load(self, source_table, target_table, target_system, rows):
        """Record load step."""
        self.lineage.append({
            'step': 'load',
            'source_table': source_table,
            'target_table': target_table,
            'target_system': target_system,
            'rows': rows,
            'timestamp': datetime.now()
        })
    
    def get_lineage_report(self, table_name):
        """Get complete lineage for a table."""
        relevant_steps = [l for l in self.lineage 
                         if l.get('table') == table_name or l.get('input_table') == table_name]
        
        return relevant_steps
    
    def export_lineage(self, filename='lineage.json'):
        """Export lineage for audit purposes."""
        with open(filename, 'w') as f:
            json.dump(self.lineage, f, indent=2, default=str)

# Example usage
tracker = DataLineageTracker()

# Track ETL pipeline
tracker.record_extract('OLTP System', 'transactions', 1000000, datetime.now())
tracker.record_transform('transactions', 'cleaned_transactions', 
                        'Remove duplicates, standardize dates')
tracker.record_load('cleaned_transactions', 'fact_transactions', 'Data Warehouse', 950000)

# Get lineage report
report = tracker.get_lineage_report('fact_transactions')
for step in report:
    print(f"{step['step'].upper()}: {step}")

Customer Master Data Management:

Customer ID: CUST001
├── Legal Name: John Smith
├── Date of Birth: 1980-05-15
├── Tax ID: XXX-XX-XXXX
├── Addresses:
│   ├── Primary: 100 Main St, New York, NY
│   └── Mailing: PO Box 123, New York, NY
├── Contact:
│   ├── Phone: 555-0101
│   └── Email: john.smith@email.com
├── Relationships:
│   ├── Accounts:
│   │   ├── Checking (CHK1001) - Balance: $25,000
│   │   ├── Savings (SAV1001) - Balance: $75,000
│   │   └── Credit Card (CC1001) - Balance: -$5,000
│   └── Loans:
│       └── Mortgage (L001) - Balance: $250,000
└── Customer Segment: Premium

# Example: Master Data Management implementation

class CustomerMDM:
    """Master Data Management for customers."""
    
    def __init__(self):
        self.customer_master = {}
        self.duplicate_groups = {}
    
    def add_customer(self, customer_data):
        """Add or update customer in master."""
        customer_id = customer_data['customer_id']
        
        if customer_id not in self.customer_master:
            self.customer_master[customer_id] = {
                'master_id': customer_id,
                'created_at': datetime.now(),
                'updated_at': datetime.now(),
                'versions': []
            }
        
        # Versioning: keep history
        current = self.customer_master[customer_id]
        current['versions'].append({
            'timestamp': datetime.now(),
            'data': customer_data
        })
        
        # Update current data
        current['current'] = customer_data
        current['updated_at'] = datetime.now()
    
    def link_duplicates(self, primary_id, duplicate_id):
        """Link duplicate customer records."""
        if primary_id not in self.duplicate_groups:
            self.duplicate_groups[primary_id] = {'primary': primary_id, 'duplicates': []}
        
        self.duplicate_groups[primary_id]['duplicates'].append(duplicate_id)
    
    def get_master_customer(self, customer_id):
        """Get master record for a customer, resolving duplicates."""
        # Check if this ID is a duplicate
        for primary, group in self.duplicate_groups.items():
            if customer_id in group['duplicates'] or customer_id == primary:
                return self.customer_master[primary]
        
        # Or return directly
        return self.customer_master.get(customer_id)
    
    def get_golden_record(self, customer_id):
        """Get the golden record (best version) for a customer."""
        master = self.get_master_customer(customer_id)
        if not master:
            return None
        
        # Simple logic: use the most recent version
        latest = max(master['versions'], key=lambda x: x['timestamp'])
        return latest['data']
    
    def check_master_data_quality(self):
        """Check quality of master data."""
        issues = []
        
        for cust_id, master in self.customer_master.items():
            current = master.get('current', {})
            
            # Check required fields
            if not current.get('first_name') or not current.get('last_name'):
                issues.append(f"Customer {cust_id}: Missing name")
            
            if not current.get('tax_id'):
                issues.append(f"Customer {cust_id}: Missing tax ID")
            
            if not current.get('address'):
                issues.append(f"Customer {cust_id}: Missing address")
        
        return issues

# Usage example
mdm = CustomerMDM()

# Add customer data from different sources
mdm.add_customer({
    'customer_id': 'CUST001',
    'first_name': 'John',
    'last_name': 'Smith',
    'tax_id': '123-45-6789',
    'address': '100 Main St, New York, NY',
    'phone': '555-0101',
    'segment': 'Premium',
    'source': 'Core Banking'
})

# Link duplicate records
mdm.link_duplicates('CUST001', 'CUST001_DUP')

# Get golden record
golden = mdm.get_golden_record('CUST001')
print(f"Golden Record: {golden}")

# Check quality
quality_issues = mdm.check_master_data_quality()
print(f"Quality Issues: {quality_issues}")