Source Systems → [Extract] → Staging Area → [Transform] → Data Warehouse → [Load] → Data Marts
     ↓                    ↓                ↓              ↓
OLTP Systems        Raw Data         Cleaned Data    Business-Ready Data

# Example: Extracting data from a banking OLTP system
import pyodbc
import pandas as pd

def extract_transactions(connection_string, last_extract_date):
    """Extract transactions since the last extraction."""
    conn = pyodbc.connect(connection_string)
    
    # Query to get only new/modified records
    query = """
    SELECT transaction_id, account_id, transaction_date, 
           amount, transaction_type, status
    FROM transactions 
    WHERE last_modified > ? 
    ORDER BY transaction_id
    """
    
    # Execute with parameter
    df = pd.read_sql(query, conn, params=[last_extract_date])
    conn.close()
    
    return df

# Usage
last_run = '2024-01-14 23:59:59'
new_transactions = extract_transactions(db_connection, last_run)

# Example: Transformation functions
def clean_customer_data(df):
    """Apply transformations to customer data."""
    # Remove duplicates
    df = df.drop_duplicates(subset=['customer_id'])
    
    # Handle missing values
    df['credit_score'] = df['credit_score'].fillna(600)  # Default for missing
    
    # Standardize state codes
    state_map = {'CA': 'California', 'NY': 'New York', 'TX': 'Texas'}
    df['state_full'] = df['state'].map(state_map)
    
    # Calculate derived fields
    df['age'] = (pd.to_datetime('today') - pd.to_datetime(df['date_of_birth'])).dt.days // 365
    
    # Create customer segment logic
    df['segment'] = pd.cut(df['annual_income'], 
                          bins=[0, 50000, 100000, 1000000],
                          labels=['Basic', 'Standard', 'Premium'])
    
    return df

# Usage
raw_customers = pd.read_csv('customers_raw.csv')
cleaned_customers = clean_customer_data(raw_customers)

# Example: Loading data to warehouse
import sqlalchemy

def load_to_warehouse(df, table_name, engine, load_type='incremental'):
    """Load DataFrame to data warehouse."""
    if load_type == 'full':
        # Replace entire table
        df.to_sql(table_name, engine, if_exists='replace', index=False)
    
    elif load_type == 'incremental':
        # Append new records
        df.to_sql(table_name, engine, if_exists='append', index=False)
    
    elif load_type == 'upsert':
        # For simplicity in this example, we'll do a delete+insert
        # In production, use MERGE or database-specific UPSERT
        with engine.connect() as conn:
            # Delete existing records
            conn.execute(f"DELETE FROM {table_name} WHERE customer_id IN :ids", 
                        {'ids': tuple(df['customer_id'].tolist())})
            # Insert new records
            df.to_sql(table_name, conn, if_exists='append', index=False)

# Usage
engine = sqlalchemy.create_engine('postgresql://user:pass@host/db')
load_to_warehouse(cleaned_customers, 'dim_customers', engine, 'upsert')

Time          Activity
10:00 PM      Branch closes, customer access restricted
10:30 PM      Transaction cutoff (all transactions up to now)
11:00 PM      Batch processing (interest, fees, transfers)
12:00 AM      Date rollover
12:30 AM      Data extraction from OLTP systems
1:00 AM       Extract begins to staging area
2:00 AM      Transformation begins
3:30 AM      Data loading to warehouse
4:30 AM      Aggregation building
5:30 AM      Report generation
6:00 AM      Systems online for new day

# Example: ETL vs ELT in Python

# ETL Approach (Transform Before Load)
def etl_pipeline():
    raw_data = extract_from_source()
    transformed = clean_and_transform(raw_data)  # Transform first
    load_to_warehouse(transformed)               # Load after

# ELT Approach (Load Before Transform)
def elt_pipeline():
    raw_data = extract_from_source()
    load_to_data_lake(raw_data)                  # Load raw first
    # Transform later when needed
    transformed = transform_from_lake(raw_data)

# Hybrid Approach (Most Common in Banking)
def hybrid_pipeline():
    raw_data = extract_from_source()
    # Basic validation (critical for compliance)
    validated = basic_validation(raw_data)
    load_to_staging(validated)                   # Load with minimal transform
    # Full transformation in the warehouse
    warehouse.transform_data()

# Example: Data quality checks in ETL pipeline

class DataQualityChecker:
    def __init__(self, df, table_name):
        self.df = df
        self.table_name = table_name
        self.issues = []
    
    def check_for_nulls(self, columns, threshold=0.05):
        """Check if any column has > threshold % nulls."""
        for col in columns:
            null_pct = self.df[col].isnull().mean()
            if null_pct > threshold:
                self.issues.append({
                    'table': self.table_name,
                    'column': col,
                    'issue': f'Null percentage {null_pct:.2%} exceeds {threshold:.0%} threshold'
                })
    
    def check_for_duplicates(self, key_columns):
        """Check for duplicate rows based on key columns."""
        duplicates = self.df.duplicated(subset=key_columns).sum()
        if duplicates > 0:
            self.issues.append({
                'table': self.table_name,
                'issue': f'Found {duplicates} duplicate records based on {key_columns}'
            })
    
    def check_data_type(self, column, expected_type):
        """Verify data type matches expectation."""
        actual_type = self.df[column].dtype
        if str(actual_type) != expected_type:
            self.issues.append({
                'table': self.table_name,
                'column': column,
                'issue': f'Expected {expected_type}, got {actual_type}'
            })
    
    def report_issues(self):
        """Generate quality report."""
        if self.issues:
            print(f"Data Quality Issues Found in {self.table_name}:")
            for issue in self.issues:
                print(f"  - {issue.get('column', '')}: {issue['issue']}")
        else:
            print(f"✓ No data quality issues found in {self.table_name}")
        return len(self.issues) == 0

# Usage
checker = DataQualityChecker(cleaned_customers, 'dim_customers')
checker.check_for_nulls(['email', 'phone'], threshold=0.10)
checker.check_for_duplicates(['customer_id'])
checker.report_issues()

# Import required libraries
import pandas as pd
import numpy as np
import sqlite3
from datetime import datetime, timedelta
import os

# Create sample data (if not exists)
def create_sample_transactions():
    """Generate realistic banking transaction data."""
    np.random.seed(42)
    
    # Create customers
    customers = pd.DataFrame({
        'customer_id': range(1, 101),
        'name': [f'Customer_{i}' for i in range(1, 101)],
        'segment': np.random.choice(['Premium', 'Standard', 'Basic'], 100, p=[0.2, 0.5, 0.3]),
        'state': np.random.choice(['CA', 'NY', 'TX', 'FL', 'IL'], 100),
        'credit_score': np.random.normal(700, 50, 100).astype(int).clip(500, 850),
        'annual_income': np.random.normal(80000, 30000, 100).clip(20000, 200000)
    })
    customers.to_csv('customers.csv', index=False)
    
    # Create accounts
    accounts = []
    for cust_id in range(1, 101):
        # Each customer has 1-3 accounts
        num_accounts = np.random.choice([1, 2, 3], p=[0.4, 0.4, 0.2])
        for i in range(num_accounts):
            account_type = np.random.choice(['Checking', 'Savings', 'Credit Card'])
            accounts.append({
                'account_id': len(accounts) + 1,
                'customer_id': cust_id,
                'account_type': account_type,
                'balance': np.random.uniform(100, 50000),
                'is_active': 1
            })
    accounts_df = pd.DataFrame(accounts)
    accounts_df.to_csv('accounts.csv', index=False)
    
    # Create transactions (last 90 days)
    transactions = []
    for account_id in accounts_df['account_id']:
        # 10-50 transactions per account
        num_txns = np.random.randint(10, 50)
        for i in range(num_txns):
            days_ago = np.random.randint(0, 90)
            trans_date = datetime.now() - timedelta(days=days_ago)
            trans_type = np.random.choice(['Deposit', 'Withdrawal', 'Purchase', 'Payment'], 
                                         p=[0.2, 0.3, 0.3, 0.2])
            amount = np.random.uniform(10, 500)
            
            transactions.append({
                'transaction_id': len(transactions) + 1,
                'account_id': account_id,
                'transaction_date': trans_date.strftime('%Y-%m-%d'),
                'transaction_type': trans_type,
                'amount': round(amount, 2),
                'status': np.random.choice(['Completed', 'Pending', 'Failed'], p=[0.9, 0.05, 0.05])
            })
    transactions_df = pd.DataFrame(transactions)
    transactions_df.to_csv('transactions.csv', index=False)
    
    return customers, accounts_df, transactions_df

# Create sample data if files don't exist
if not all(os.path.exists(f) for f in ['customers.csv', 'accounts.csv', 'transactions.csv']):
    customers, accounts, transactions = create_sample_transactions()
    print("Sample data created!")

class BankingETL:
    """Complete ETL pipeline for banking data."""
    
    def __init__(self, db_path='banking.db'):
        self.db_path = db_path
        self.source_data = {}
        self.transformed_data = {}
        self.quality_report = []
    
    # ============= EXTRACT =============
    def extract(self):
        """Extract data from source files."""
        print("🔍 EXTRACT: Reading source files...")
        
        self.source_data['customers'] = pd.read_csv('customers.csv')
        self.source_data['accounts'] = pd.read_csv('accounts.csv')
        self.source_data['transactions'] = pd.read_csv('transactions.csv')
        
        # Record extraction metrics
        for name, df in self.source_data.items():
            print(f"  - {name}: {len(df)} records")
            # Check for errors
            if df.isnull().any().any():
                print(f"    ⚠️ Missing values found in {name}")
    
    # ============= TRANSFORM =============
    def transform(self):
        """Apply transformations to extracted data."""
        print("\n⚙️ TRANSFORM: Processing data...")
        
        # ------------- Transform Customers -------------
        customers = self.source_data['customers'].copy()
        
        # Clean credit scores
        customers['credit_score'] = customers['credit_score'].clip(300, 850)
        
        # Create derived fields
        customers['income_segment'] = pd.cut(
            customers['annual_income'],
            bins=[0, 40000, 70000, 100000, 1000000],
            labels=['Low', 'Medium-Low', 'Medium-High', 'High']
        )
        
        # Add customer tier based on credit score and income
        def assign_tier(row):
            if row['credit_score'] > 750 and row['annual_income'] > 100000:
                return 'Platinum'
            elif row['credit_score'] > 700 and row['annual_income'] > 60000:
                return 'Gold'
            elif row['credit_score'] > 650:
                return 'Silver'
            else:
                return 'Bronze'
        
        customers['tier'] = customers.apply(assign_tier, axis=1)
        self.transformed_data['customers'] = customers
        
        # ------------- Transform Accounts -------------
        accounts = self.source_data['accounts'].copy()
        
        # Add account age (in months)
        # (simulated since we don't have open dates)
        accounts['account_age_months'] = np.random.randint(1, 120, len(accounts))
        
        # Add account tier based on balance
        def account_tier(balance):
            if balance > 50000:
                return 'High-Value'
            elif balance > 20000:
                return 'Medium-Value'
            else:
                return 'Low-Value'
        
        accounts['value_tier'] = accounts['balance'].apply(account_tier)
        self.transformed_data['accounts'] = accounts
        
        # ------------- Transform Transactions -------------
        transactions = self.source_data['transactions'].copy()
        
        # Parse dates
        transactions['transaction_date'] = pd.to_datetime(transactions['transaction_date'])
        
        # Add month and day of week
        transactions['month'] = transactions['transaction_date'].dt.month
        transactions['day_of_week'] = transactions['transaction_date'].dt.dayofweek
        
        # Flag weekend transactions
        transactions['is_weekend'] = transactions['day_of_week'] >= 5
        
        # Add amount category
        def amount_category(amount):
            if amount < 50:
                return 'Small'
            elif amount < 200:
                return 'Medium'
            elif amount < 500:
                return 'Large'
            else:
                return 'Very Large'
        
        transactions['amount_category'] = transactions['amount'].apply(amount_category)
        
        # Add transaction hour (simulated for demo)
        transactions['hour'] = np.random.randint(0, 24, len(transactions))
        
        self.transformed_data['transactions'] = transactions
        
        # Report transformation results
        for name, df in self.transformed_data.items():
            print(f"  - {name}: transformed ({len(df)} records)")
    
    # ============= DATA QUALITY =============
    def check_quality(self):
        """Perform data quality checks on transformed data."""
        print("\n✅ QUALITY CHECKS: Validating data...")
        
        # Check customers
        cust = self.transformed_data['customers']
        null_count = cust[['credit_score', 'annual_income']].isnull().sum().sum()
        if null_count > 0:
            self.quality_report.append(f"⚠️ {null_count} null values in customer data")
        
        # Check for valid credit scores
        invalid_scores = cust[(cust['credit_score'] < 300) | (cust['credit_score'] > 850)].shape[0]
        if invalid_scores > 0:
            self.quality_report.append(f"⚠️ {invalid_scores} invalid credit scores found")
        
        # Check accounts
        acct = self.transformed_data['accounts']
        negative_balances = acct[acct['balance'] < 0].shape[0]
        if negative_balances > 0:
            self.quality_report.append(f"⚠️ {negative_balances} accounts with negative balances")
        
        # Check transactions
        txns = self.transformed_data['transactions']
        failed_txns = txns[txns['status'] == 'Failed'].shape[0]
        if failed_txns > 0:
            self.quality_report.append(f"⚠️ {failed_txns} failed transactions found")
        
        # Report quality results
        if self.quality_report:
            print("QUALITY ISSUES FOUND:")
            for issue in self.quality_report:
                print(f"  {issue}")
        else:
            print("✓ All quality checks passed!")
    
    # ============= LOAD =============
    def load(self):
        """Load transformed data to database."""
        print("\n💾 LOAD: Loading to database...")
        
        # Connect to SQLite database
        conn = sqlite3.connect(self.db_path)
        
        try:
            # Load each transformed DataFrame
            for table_name, df in self.transformed_data.items():
                # For simplicity, replace the table
                df.to_sql(table_name, conn, if_exists='replace', index=False)
                print(f"  - Loaded {len(df)} records to {table_name}")
            
            # Create indexes for performance
            cursor = conn.cursor()
            cursor.execute("CREATE INDEX idx_customer_id ON customers(customer_id)")
            cursor.execute("CREATE INDEX idx_account_id ON accounts(account_id)")
            cursor.execute("CREATE INDEX idx_account_id_txn ON transactions(account_id)")
            cursor.execute("CREATE INDEX idx_txn_date ON transactions(transaction_date)")
            cursor.execute("CREATE INDEX idx_txn_status ON transactions(status)")
            print("  - Created performance indexes")
            
            conn.commit()
            print("✓ Data loaded successfully!")
            
        except Exception as e:
            print(f"❌ Error loading data: {e}")
            conn.rollback()
        finally:
            conn.close()
    
    # ============= VALIDATE =============
    def validate_load(self):
        """Verify data was loaded correctly."""
        print("\n🔍 VALIDATE: Verifying data...")
        
        conn = sqlite3.connect(self.db_path)
        
        try:
            # Check row counts
            for table in ['customers', 'accounts', 'transactions']:
                cursor = conn.execute(f"SELECT COUNT(*) FROM {table}")
                count = cursor.fetchone()[0]
                print(f"  - {table}: {count} rows")
            
            # Check for data integrity
            cursor = conn.execute("""
                SELECT COUNT(*) 
                FROM transactions t
                JOIN accounts a ON t.account_id = a.account_id
                WHERE a.is_active = 1
            """)
            active_txns = cursor.fetchone()[0]
            print(f"  - Active account transactions: {active_txns}")
            
        except Exception as e:
            print(f"❌ Validation error: {e}")
        finally:
            conn.close()
    
    # ============= RUN PIPELINE =============
    def run(self):
        """Run the complete ETL pipeline."""
        print("=" * 60)
        print("🏦 BANKING ETL PIPELINE")
        print("=" * 60)
        
        try:
            self.extract()
            self.transform()
            self.check_quality()
            self.load()
            self.validate_load()
            
            print("\n" + "=" * 60)
            print("✅ ETL Pipeline Complete!")
            print("=" * 60)
            
            # Return success indicator
            return len(self.quality_report) == 0
            
        except Exception as e:
            print(f"\n❌ Pipeline failed: {e}")
            return False

# ============= RUN THE PIPELINE =============
if __name__ == "__main__":
    etl = BankingETL()
    success = etl.run()
    
    # Query the database to verify
    if success:
        print("\n📊 SAMPLE QUERY RESULTS:")
        conn = sqlite3.connect('banking.db')
        result = conn.execute("""
            SELECT 
                c.segment,
                COUNT(DISTINCT c.customer_id) as customer_count,
                AVG(c.credit_score) as avg_credit_score,
                COUNT(t.transaction_id) as transaction_count,
                SUM(t.amount) as total_amount
            FROM customers c
            JOIN accounts a ON c.customer_id = a.customer_id
            JOIN transactions t ON a.account_id = t.account_id
            WHERE t.status = 'Completed'
            GROUP BY c.segment
            ORDER BY total_amount DESC
        """).fetchall()
        
        for row in result:
            print(f"  {row[0]}: {row[1]} customers, Avg Score: {row[2]:.0f}, Txns: {row[3]}, Amount: ${row[4]:,.2f}")
        conn.close()