LESSON 5: FEATURE ENGINEERING FOR FINANCIAL DATA

SECTION 1: LEARNING OBJECTIVES

By the end of this lesson, you will be able to:

Understand the importance of feature engineering in financial data analytics and its impact on model performance.
Create domain-specific financial features including ratios, aggregates, and risk indicators.
Engineer time-based features for transaction data (recency, frequency, monetary value).
Build interaction features that capture relationships between financial variables.
Apply binning and discretization techniques for continuous financial variables.
Implement feature extraction from text data for financial documents.
Validate engineered features using statistical methods and business logic.
Document feature engineering decisions for regulatory compliance (SR 11-7).

SECTION 2: INTRODUCTION TO FEATURE ENGINEERING

2.1 What Is Feature Engineering?

Feature engineering is the process of creating new features from raw data to improve machine learning model performance. In banking, feature engineering transforms raw transaction logs, customer data, and market data into predictive signals.

The Feature Engineering Mindset:

Raw Banking Data → Feature Engineering → Predictive Features
                      ↓
┌─────────────────────────────────────────────────────────────┐
│ Transaction Logs          →  RFM Scores, Spending Patterns │
│ Customer Demographics     →  Segment, Risk Tier            │
│ Account Balances          →  Utilization, Growth Rates     │
│ Loan History              →  DTI Ratio, Default Indicators │
│ Market Data               →  Volatility, Moving Averages   │
└─────────────────────────────────────────────────────────────┘

2.2 Why Feature Engineering Matters in Banking

Benefit	Banking Example
Improved Model Performance	DTI ratio predicts default better than raw income
Business Interpretability	RFM scores are intuitive for marketing teams
Regulatory Compliance	Transparent features satisfy SR 11-7 requirements
Reduced Data Requirements	Smart features capture patterns with less data
Faster Training	Fewer, better features reduce training time

SECTION 3: DOMAIN-SPECIFIC FINANCIAL FEATURES

3.1 Creating Financial Ratios

# ============= FINANCIAL RATIOS =============

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
from scipy import stats

# Create sample data
np.random.seed(42)
n = 1000

financial_data = pd.DataFrame({
    'customer_id': range(1, n+1),
    'monthly_income': np.random.normal(6000, 2000, n).clip(1500, 20000),
    'monthly_expenses': np.random.normal(4000, 1500, n).clip(500, 15000),
    'credit_card_balance': np.random.normal(5000, 3000, n).clip(0, 25000),
    'credit_limit': np.random.normal(15000, 5000, n).clip(5000, 50000),
    'loan_balance': np.random.normal(100000, 50000, n).clip(0, 400000),
    'savings_balance': np.random.normal(30000, 20000, n).clip(0, 150000),
    'investment_balance': np.random.normal(50000, 40000, n).clip(0, 300000),
    'tenure_months': np.random.exponential(36, n).astype(int).clip(1, 240),
    'num_transactions': np.random.poisson(20, n).clip(0, 80),
    'num_late_payments': np.random.poisson(0.5, n).clip(0, 8)
})

class FinancialFeatureEngineer:
    """Create domain-specific financial features."""
    
    def __init__(self, data):
        self.data = data.copy()
        self.feature_names = []
        self.created_features = {}
    
    def create_ratio_features(self):
        """Create financial ratio features."""
        
        # Debt-to-Income Ratio (DTI)
        self.data['dti_ratio'] = (
            (self.data['credit_card_balance'] + self.data['loan_balance']) / 
            self.data['monthly_income'] * 12
        ).clip(0, 10)
        self.feature_names.append('dti_ratio')
        
        # Savings Rate
        self.data['savings_rate'] = (
            self.data['savings_balance'] / (self.data['monthly_income'] * 12)
        ).clip(0, 5)
        self.feature_names.append('savings_rate')
        
        # Credit Utilization Ratio
        self.data['credit_utilization'] = (
            self.data['credit_card_balance'] / self.data['credit_limit']
        ).clip(0, 2)
        self.feature_names.append('credit_utilization')
        
        # Expense-to-Income Ratio
        self.data['expense_ratio'] = (
            self.data['monthly_expenses'] / self.data['monthly_income']
        ).clip(0, 10)
        self.feature_names.append('expense_ratio')
        
        # Loan-to-Value Ratio (simplified)
        self.data['ltv_ratio'] = (
            self.data['loan_balance'] / (self.data['savings_balance'] + self.data['investment_balance'] + 1)
        ).clip(0, 10)
        self.feature_names.append('ltv_ratio')
        
        # Net Worth to Income Ratio
        self.data['net_worth_ratio'] = (
            (self.data['savings_balance'] + self.data['investment_balance'] - 
             self.data['credit_card_balance'] - self.data['loan_balance']) /
            (self.data['monthly_income'] * 12 + 1)
        ).clip(-10, 10)
        self.feature_names.append('net_worth_ratio')
        
        return self.data
    
    def create_risk_features(self):
        """Create risk indicator features."""
        
        # High Risk Flags
        self.data['high_dti'] = (self.data['dti_ratio'] > 0.43).astype(int)
        self.data['high_utilization'] = (self.data['credit_utilization'] > 0.80).astype(int)
        self.data['low_savings'] = (self.data['savings_rate'] < 0.10).astype(int)
        self.data['has_late_payments'] = (self.data['num_late_payments'] > 0).astype(int)
        
        # Composite Risk Score
        self.data['risk_score'] = (
            self.data['high_dti'] * 4 +
            self.data['high_utilization'] * 3 +
            self.data['low_savings'] * 2 +
            self.data['has_late_payments'] * 3
        )
        
        # Risk Category
        self.data['risk_category'] = pd.cut(
            self.data['risk_score'],
            bins=[-1, 2, 5, 8, 12],
            labels=['Low', 'Medium-Low', 'Medium-High', 'High']
        )
        
        self.feature_names.extend(['high_dti', 'high_utilization', 'low_savings', 
                                   'has_late_payments', 'risk_score', 'risk_category'])
        
        return self.data
    
    def create_behavioral_features(self):
        """Create behavioral features."""
        
        # Transaction Intensity
        self.data['transaction_intensity'] = (
            self.data['num_transactions'] / (self.data['tenure_months'] / 12 + 1)
        )
        self.feature_names.append('transaction_intensity')
        
        # Tenure Category
        self.data['tenure_category'] = pd.cut(
            self.data['tenure_months'],
            bins=[0, 12, 36, 120, float('inf')],
            labels=['New', 'Regular', 'Loyal', 'Veteran']
        )
        self.feature_names.append('tenure_category')
        
        # Spending Power (disposable income)
        self.data['spending_power'] = (
            self.data['monthly_income'] - self.data['monthly_expenses'] +
            self.data['savings_balance'] * 0.01
        ).clip(0, float('inf'))
        self.feature_names.append('spending_power')
        
        # Financial Stability Score
        self.data['stability_score'] = (
            (self.data['savings_rate'] * 2) +
            (1 - self.data['credit_utilization']) * 3 +
            (1 - self.data['dti_ratio'] / 2) * 2 +
            (1 - self.data['has_late_payments']) * 3
        ).clip(0, 10)
        self.feature_names.append('stability_score')
        
        return self.data
    
    def create_all_features(self):
        """Create all financial features."""
        self.create_ratio_features()
        self.create_risk_features()
        self.create_behavioral_features()
        
        print(f"\n✅ Created {len(self.feature_names)} new financial features")
        print(f"   Total features: {len(self.data.columns)}")
        
        return self.data, self.feature_names
    
    def feature_summary(self):
        """Generate summary of created features."""
        print("\n" + "="*60)
        print("FINANCIAL FEATURE ENGINEERING SUMMARY")
        print("="*60)
        
        for feature in self.feature_names:
            if feature in self.data.columns:
                print(f"\n📊 {feature}:")
                print(f"  Type: {self.data[feature].dtype}")
                print(f"  Missing: {self.data[feature].isnull().sum()}")
                print(f"  Unique: {self.data[feature].nunique()}")
                
                if pd.api.types.is_numeric_dtype(self.data[feature]):
                    print(f"  Mean: {self.data[feature].mean():.3f}")
                    print(f"  Std: {self.data[feature].std():.3f}")
                    print(f"  Min: {self.data[feature].min():.3f}")
                    print(f"  Max: {self.data[feature].max():.3f}")

# Create features
engineer = FinancialFeatureEngineer(financial_data)
engineered_data, feature_names = engineer.create_all_features()
engineer.feature_summary()

# Visualize key features
fig, axes = plt.subplots(2, 3, figsize=(15, 10))

features_to_plot = ['dti_ratio', 'credit_utilization', 'savings_rate', 
                    'risk_score', 'transaction_intensity', 'stability_score']

for i, (ax, feature) in enumerate(zip(axes.flatten(), features_to_plot)):
    if feature in engineered_data.columns:
        engineered_data[feature].hist(bins=30, ax=ax, edgecolor='black', alpha=0.7)
        ax.set_title(f'{feature.replace("_", " ").title()}', fontsize=12)
        ax.set_xlabel('Value')
        ax.set_ylabel('Frequency')

plt.tight_layout()
plt.savefig('financial_features.png', dpi=300)
plt.show()

3.2 RFM Analysis (Recency, Frequency, Monetary)

# ============= RFM FEATURES =============

class RFMAnalyzer:
    """Create RFM (Recency, Frequency, Monetary) features."""
    
    def __init__(self, transaction_data, customer_data):
        self.transactions = transaction_data.copy()
        self.customers = customer_data.copy()
        self.rfm_data = None
    
    def calculate_rfm(self, date_col='date', amount_col='amount', customer_col='customer_id'):
        """Calculate RFM scores for each customer."""
        
        # Ensure date is datetime
        self.transactions[date_col] = pd.to_datetime(self.transactions[date_col])
        
        # Current date for recency calculation
        current_date = self.transactions[date_col].max()
        
        # Calculate RFM metrics
        rfm = self.transactions.groupby(customer_col).agg({
            date_col: lambda x: (current_date - x.max()).days,  # Recency
            customer_col: 'count',  # Frequency
            amount_col: 'sum'  # Monetary
        }).rename(columns={
            date_col: 'recency_days',
            customer_col: 'frequency_count',
            amount_col: 'monetary_sum'
        })
        
        # Calculate average transaction value
        avg_transaction = self.transactions.groupby(customer_col)[amount_col].mean()
        rfm['avg_transaction_amount'] = avg_transaction
        
        # Create RFM scores (1-5, 5 being best)
        rfm['recency_score'] = pd.qcut(rfm['recency_days'], 5, labels=[5, 4, 3, 2, 1])
        rfm['frequency_score'] = pd.qcut(rfm['frequency_count'], 5, labels=[1, 2, 3, 4, 5])
        rfm['monetary_score'] = pd.qcut(rfm['monetary_sum'], 5, labels=[1, 2, 3, 4, 5])
        
        # Combined RFM score
        rfm['rfm_score'] = (
            rfm['recency_score'].astype(int) * 100 +
            rfm['frequency_score'].astype(int) * 10 +
            rfm['monetary_score'].astype(int)
        )
        
        # RFM Segment
        def get_rfm_segment(row):
            """Assign RFM segment based on scores."""
            r = row['recency_score']
            f = row['frequency_score']
            m = row['monetary_score']
            
            if r >= 4 and f >= 4 and m >= 4:
                return 'Champions'
            elif r >= 3 and f >= 3 and m >= 3:
                return 'Loyal Customers'
            elif r >= 4 and f >= 1 and m >= 1:
                return 'Recent Customers'
            elif r >= 1 and f >= 4 and m >= 4:
                return 'Big Spenders'
            elif r >= 3 and f >= 1 and m >= 1:
                return 'Potential Loyalists'
            elif r >= 2 and f >= 2 and m >= 1:
                return 'Needs Attention'
            elif r >= 1 and f >= 1 and m >= 1:
                return 'About to Sleep'
            else:
                return 'At Risk'
        
        rfm['rfm_segment'] = rfm.apply(get_rfm_segment, axis=1)
        
        # Merge with customer data
        self.rfm_data = self.customers.merge(rfm, left_on='customer_id', right_index=True, how='left')
        
        return self.rfm_data
    
    def segment_summary(self):
        """Generate segment summary."""
        if self.rfm_data is None:
            self.calculate_rfm()
        
        print("\n" + "="*60)
        print("RFM SEGMENT SUMMARY")
        print("="*60)
        
        segment_stats = self.rfm_data.groupby('rfm_segment').agg({
            'rfm_segment': 'count',
            'monetary_sum': 'mean',
            'frequency_count': 'mean',
            'recency_days': 'mean'
        }).rename(columns={'rfm_segment': 'customer_count'})
        
        segment_stats['customer_pct'] = (segment_stats['customer_count'] / len(self.rfm_data) * 100).round(1)
        segment_stats = segment_stats.sort_values('customer_count', ascending=False)
        
        print("\n📊 Segment Overview:")
        print(segment_stats.to_string())
        
        return segment_stats
    
    def visualize_segments(self):
        """Visualize RFM segments."""
        if self.rfm_data is None:
            self.calculate_rfm()
        
        fig, axes = plt.subplots(2, 2, figsize=(14, 10))
        
        # 1. Segment distribution
        segment_counts = self.rfm_data['rfm_segment'].value_counts()
        segment_counts.plot(kind='bar', ax=axes[0, 0])
        axes[0, 0].set_title('Customer Segment Distribution', fontsize=12)
        axes[0, 0].set_xlabel('Segment')
        axes[0, 0].set_ylabel('Count')
        axes[0, 0].tick_params(axis='x', rotation=45)
        
        # 2. Monetary by segment
        self.rfm_data.boxplot(column='monetary_sum', by='rfm_segment', ax=axes[0, 1])
        axes[0, 1].set_title('Monetary Value by Segment', fontsize=12)
        axes[0, 1].set_xlabel('Segment')
        axes[0, 1].set_ylabel('Monetary Sum ($)')
        axes[0, 1].tick_params(axis='x', rotation=45)
        
        # 3. RFM score distribution
        self.rfm_data['rfm_score'].hist(bins=20, ax=axes[1, 0], edgecolor='black', alpha=0.7)
        axes[1, 0].set_title('RFM Score Distribution', fontsize=12)
        axes[1, 0].set_xlabel('RFM Score')
        axes[1, 0].set_ylabel('Count')
        
        # 4. Recency vs Frequency scatter
        scatter = axes[1, 1].scatter(
            self.rfm_data['recency_days'],
            self.rfm_data['frequency_count'],
            c=self.rfm_data['monetary_sum'],
            alpha=0.6,
            cmap='viridis'
        )
        axes[1, 1].set_title('Recency vs Frequency (colored by Monetary)', fontsize=12)
        axes[1, 1].set_xlabel('Recency (days)')
        axes[1, 1].set_ylabel('Frequency (count)')
        plt.colorbar(scatter, ax=axes[1, 1], label='Monetary Sum')
        
        plt.tight_layout()
        plt.savefig('rfm_analysis.png', dpi=300)
        plt.show()

# Create sample transaction data for RFM
np.random.seed(42)
n_transactions = 5000
n_customers = 200

rfm_customers = pd.DataFrame({
    'customer_id': range(1, n_customers + 1),
    'age': np.random.normal(45, 15, n_customers).astype(int).clip(18, 85),
    'income': np.random.normal(70000, 20000, n_customers).clip(20000, 200000)
})

rfm_transactions = pd.DataFrame({
    'customer_id': np.random.randint(1, n_customers + 1, n_transactions),
    'date': [pd.Timestamp('2024-06-01') - pd.Timedelta(days=np.random.randint(0, 180)) 
             for _ in range(n_transactions)],
    'amount': np.random.lognormal(4, 0.8, n_transactions).clip(5, 5000)
})

# Run RFM analysis
rfm_analyzer = RFMAnalyzer(rfm_transactions, rfm_customers)
rfm_results = rfm_analyzer.calculate_rfm()
rfm_analyzer.segment_summary()
rfm_analyzer.visualize_segments()

SECTION 4: TIME-BASED FEATURES

4.1 Temporal Feature Engineering

# ============= TIME-BASED FEATURES =============

class TemporalFeatureEngineer:
    """Create time-based features from datetime data."""
    
    def __init__(self, data, date_column):
        self.data = data.copy()
        self.date_column = date_column
        self.created_features = []
    
    def extract_time_features(self):
        """Extract features from datetime."""
        
        # Ensure datetime
        self.data[self.date_column] = pd.to_datetime(self.data[self.date_column])
        
        # Basic time components
        self.data['year'] = self.data[self.date_column].dt.year
        self.data['month'] = self.data[self.date_column].dt.month
        self.data['day'] = self.data[self.date_column].dt.day
        self.data['day_of_week'] = self.data[self.date_column].dt.dayofweek
        self.data['day_of_year'] = self.data[self.date_column].dt.dayofyear
        self.data['week_of_year'] = self.data[self.date_column].dt.isocalendar().week.astype(int)
        self.data['quarter'] = self.data[self.date_column].dt.quarter
        
        # Day of week name
        self.data['day_name'] = self.data[self.date_column].dt.day_name()
        
        # Flags
        self.data['is_weekend'] = (self.data['day_of_week'] >= 5).astype(int)
        self.data['is_month_start'] = (self.data[self.date_column].dt.is_month_start).astype(int)
        self.data['is_month_end'] = (self.data[self.date_column].dt.is_month_end).astype(int)
        self.data['is_quarter_start'] = (self.data[self.date_column].dt.is_quarter_start).astype(int)
        self.data['is_quarter_end'] = (self.data[self.date_column].dt.is_quarter_end).astype(int)
        
        # Cyclical encoding (for seasonality)
        self.data['month_sin'] = np.sin(2 * np.pi * self.data['month'] / 12)
        self.data['month_cos'] = np.cos(2 * np.pi * self.data['month'] / 12)
        self.data['day_sin'] = np.sin(2 * np.pi * self.data['day_of_week'] / 7)
        self.data['day_cos'] = np.cos(2 * np.pi * self.data['day_of_week'] / 7)
        
        # Track features
        time_features = ['year', 'month', 'day', 'day_of_week', 'day_of_year',
                        'week_of_year', 'quarter', 'day_name', 'is_weekend',
                        'is_month_start', 'is_month_end', 'is_quarter_start',
                        'is_quarter_end', 'month_sin', 'month_cos', 'day_sin', 'day_cos']
        
        self.created_features.extend(time_features)
        
        return self.data
    
    def create_lag_features(self, group_by_col, target_col, lags=[1, 7, 14, 30]):
        """Create lag features for time series."""
        
        for lag in lags:
            self.data[f'{target_col}_lag_{lag}'] = (
                self.data.groupby(group_by_col)[target_col].shift(lag)
            )
            self.created_features.append(f'{target_col}_lag_{lag}')
        
        return self.data
    
    def create_rolling_features(self, group_by_col, target_col, windows=[3, 7, 14, 30]):
        """Create rolling statistics."""
        
        for window in windows:
            # Rolling mean
            self.data[f'{target_col}_rolling_mean_{window}'] = (
                self.data.groupby(group_by_col)[target_col]
                .transform(lambda x: x.rolling(window, min_periods=1).mean())
            )
            self.created_features.append(f'{target_col}_rolling_mean_{window}')
            
            # Rolling std
            self.data[f'{target_col}_rolling_std_{window}'] = (
                self.data.groupby(group_by_col)[target_col]
                .transform(lambda x: x.rolling(window, min_periods=1).std())
            )
            self.created_features.append(f'{target_col}_rolling_std_{window}')
        
        return self.data
    
    def feature_summary(self):
        """Generate summary of time features."""
        print("\n" + "="*60)
        print("TIME-BASED FEATURE SUMMARY")
        print("="*60)
        
        print(f"\n📊 Created {len(self.created_features)} time features:")
        for feature in self.created_features:
            if feature in self.data.columns:
                print(f"  • {feature}")

# Create sample transaction data for time features
np.random.seed(42)
n = 5000

time_data = pd.DataFrame({
    'transaction_id': range(1, n+1),
    'customer_id': np.random.randint(1, 100, n),
    'date': [pd.Timestamp('2024-01-01') + pd.Timedelta(days=np.random.randint(0, 365)) 
             for _ in range(n)],
    'amount': np.random.lognormal(4, 0.8, n).clip(5, 5000)
})

# Create time features
time_engineer = TemporalFeatureEngineer(time_data, 'date')
time_data_with_features = time_engineer.extract_time_features()
time_engineer.feature_summary()

# Create lag and rolling features
time_data_with_features = time_engineer.create_lag_features('customer_id', 'amount', lags=[1, 7, 30])
time_data_with_features = time_engineer.create_rolling_features('customer_id', 'amount', windows=[3, 7, 30])

print("\n" + "="*60)
print("SAMPLE TIME FEATURES")
print("="*60)
print(time_data_with_features.head())

SECTION 5: INTERACTION AND POLYNOMIAL FEATURES

5.1 Creating Interaction Features

# ============= INTERACTION FEATURES =============

class InteractionFeatureEngineer:
    """Create interaction and polynomial features."""
    
    def __init__(self, data):
        self.data = data.copy()
        self.created_features = []
    
    def create_pairwise_interactions(self, feature_pairs):
        """Create pairwise interaction features."""
        
        for feat1, feat2 in feature_pairs:
            if feat1 in self.data.columns and feat2 in self.data.columns:
                interaction_name = f'{feat1}_x_{feat2}'
                self.data[interaction_name] = self.data[feat1] * self.data[feat2]
                self.created_features.append(interaction_name)
        
        return self.data
    
    def create_polynomial_features(self, features, degrees=[2, 3]):
        """Create polynomial features."""
        
        for feat in features:
            if feat in self.data.columns:
                for degree in degrees:
                    poly_name = f'{feat}_poly_{degree}'
                    self.data[poly_name] = self.data[feat] ** degree
                    self.created_features.append(poly_name)
        
        return self.data
    
    def create_ratio_features(self, numerator_features, denominator_features):
        """Create ratio features."""
        
        for num in numerator_features:
            for den in denominator_features:
                if num in self.data.columns and den in self.data.columns:
                    ratio_name = f'{num}_over_{den}'
                    self.data[ratio_name] = self.data[num] / (self.data[den] + 1)  # Avoid division by zero
                    self.created_features.append(ratio_name)
        
        return self.data
    
    def create_conditional_features(self, conditions):
        """Create features based on conditions."""
        
        for name, condition in conditions.items():
            self.data[name] = condition(self.data).astype(int)
            self.created_features.append(name)
        
        return self.data
    
    def feature_summary(self):
        """Generate summary of interaction features."""
        print("\n" + "="*60)
        print("INTERACTION FEATURE SUMMARY")
        print("="*60)
        
        print(f"\n📊 Created {len(self.created_features)} interaction features:")
        for feature in self.created_features:
            if feature in self.data.columns:
                print(f"  • {feature}")

# Sample data for interaction features
np.random.seed(42)
interaction_data = pd.DataFrame({
    'income': np.random.normal(70000, 20000, 500).clip(20000, 200000),
    'age': np.random.normal(45, 15, 500).astype(int).clip(18, 85),
    'credit_score': np.random.normal(700, 50, 500).clip(500, 850),
    'debt': np.random.normal(50000, 30000, 500).clip(0, 200000),
    'savings': np.random.normal(30000, 20000, 500).clip(0, 150000)
})

# Create interaction features
interaction_engineer = InteractionFeatureEngineer(interaction_data)

# Pairwise interactions
feature_pairs = [('income', 'age'), ('income', 'credit_score'), ('debt', 'savings')]
interaction_data = interaction_engineer.create_pairwise_interactions(feature_pairs)

# Polynomial features
polynomial_features = ['income', 'credit_score']
interaction_data = interaction_engineer.create_polynomial_features(polynomial_features, degrees=[2])

# Ratio features
numerator_features = ['debt', 'savings']
denominator_features = ['income', 'age']
interaction_data = interaction_engineer.create_ratio_features(numerator_features, denominator_features)

interaction_engineer.feature_summary()

# Show sample
print("\n" + "="*60)
print("SAMPLE INTERACTION FEATURES")
print("="*60)
print(interaction_data.head())

SECTION 6: FEATURE VALIDATION

6.1 Validating Engineered Features

# ============= FEATURE VALIDATION =============

class FeatureValidator:
    """Validate engineered features."""
    
    def __init__(self, original_data, engineered_data):
        self.original = original_data
        self.engineered = engineered_data
        self.validation_results = {}
    
    def validate_numeric_feature(self, feature):
        """Validate a numeric feature."""
        
        if feature not in self.engineered.columns:
            return {'valid': False, 'reason': 'Feature not found'}
        
        data = self.engineered[feature].dropna()
        
        if len(data) == 0:
            return {'valid': False, 'reason': 'All values are missing'}
        
        results = {
            'feature': feature,
            'valid': True,
            'missing_pct': self.engineered[feature].isnull().mean(),
            'unique_values': data.nunique(),
            'min': data.min(),
            'max': data.max(),
            'mean': data.mean(),
            'std': data.std(),
            'skewness': data.skew(),
            'has_infinite': np.isinf(data).any(),
            'has_nan': data.isnull().any()
        }
        
        # Check for invalid values
        if results['has_infinite']:
            results['valid'] = False
            results['reason'] = 'Contains infinite values'
        
        if results['has_nan']:
            results['valid'] = False
            results['reason'] = 'Contains NaN values'
        
        # Check for extreme skewness
        if abs(results['skewness']) > 5:
            results['valid'] = False
            results['reason'] = f'Extreme skewness: {results["skewness"]:.2f}'
        
        self.validation_results[feature] = results
        
        return results
    
    def validate_categorical_feature(self, feature):
        """Validate a categorical feature."""
        
        if feature not in self.engineered.columns:
            return {'valid': False, 'reason': 'Feature not found'}
        
        data = self.engineered[feature].dropna()
        
        if len(data) == 0:
            return {'valid': False, 'reason': 'All values are missing'}
        
        results = {
            'feature': feature,
            'valid': True,
            'missing_pct': self.engineered[feature].isnull().mean(),
            'unique_values': data.nunique(),
            'top_values': data.value_counts().head(5).to_dict(),
            'least_common': data.value_counts().tail(3).to_dict()
        }
        
        # Check if too many unique values
        if results['unique_values'] > len(data) * 0.5:
            results['valid'] = False
            results['reason'] = f'Too many unique values: {results["unique_values"]}'
        
        self.validation_results[feature] = results
        
        return results
    
    def validate_all_features(self):
        """Validate all engineered features."""
        
        for feature in self.engineered.columns:
            if feature in self.original.columns:
                continue  # Skip original features
            
            if pd.api.types.is_numeric_dtype(self.engineered[feature]):
                self.validate_numeric_feature(feature)
            else:
                self.validate_categorical_feature(feature)
        
        return self.validation_results
    
    def generate_report(self):
        """Generate validation report."""
        self.validate_all_features()
        
        print("\n" + "="*60)
        print("FEATURE VALIDATION REPORT")
        print("="*60)
        
        valid_count = sum(1 for r in self.validation_results.values() if r.get('valid', False))
        total_count = len(self.validation_results)
        
        print(f"\n📊 Summary: {valid_count}/{total_count} features validated successfully")
        
        for feature, results in self.validation_results.items():
            if results.get('valid', False):
                print(f"\n✅ {feature}: Valid")
                if 'missing_pct' in results:
                    print(f"  Missing: {results['missing_pct']:.1%}")
                if 'unique_values' in results:
                    print(f"  Unique: {results['unique_values']}")
            else:
                print(f"\n❌ {feature}: Invalid")
                print(f"  Reason: {results.get('reason', 'Unknown')}")
    
    def correlation_analysis(self, target_feature=None):
        """Analyze correlation between features."""
        
        numeric_cols = self.engineered.select_dtypes(include=[np.number]).columns
        
        if len(numeric_cols) < 2:
            print("Need at least 2 numeric columns for correlation analysis")
            return
        
        # Calculate correlation matrix
        corr_matrix = self.engineered[numeric_cols].corr()
        
        # Find high correlations
        high_corr_pairs = []
        for i in range(len(corr_matrix.columns)):
            for j in range(i+1, len(corr_matrix.columns)):
                if abs(corr_matrix.iloc[i, j]) > 0.8:
                    high_corr_pairs.append({
                        'feature1': corr_matrix.columns[i],
                        'feature2': corr_matrix.columns[j],
                        'correlation': corr_matrix.iloc[i, j]
                    })
        
        print("\n" + "="*60)
        print("CORRELATION ANALYSIS")
        print("="*60)
        
        if high_corr_pairs:
            print(f"\n📊 High Correlations Found (>0.8):")
            for pair in high_corr_pairs:
                print(f"  • {pair['feature1']} ↔ {pair['feature2']}: {pair['correlation']:.3f}")
        else:
            print("\n✅ No high correlations found (>0.8)")
        
        return high_corr_pairs

# Validate features
validator = FeatureValidator(financial_data, engineered_data)
validator.generate_report()
validator.correlation_analysis()

SECTION 7: BUSINESS RISK & FINANCIAL IMPACT

7.1 Regulatory Compliance for Feature Engineering

# ============= REGULATORY COMPLIANCE =============

def check_regulatory_compliance_for_features(feature_names):
    """Check if features comply with regulatory requirements."""
    
    print("\n" + "="*60)
    print("REGULATORY COMPLIANCE CHECK FOR FEATURES")
    print("="*60)
    
    requirements = {
        'SR 11-7': {
            'description': 'Model risk management requires documented feature engineering',
            'requirements': [
                'Feature engineering process documented',
                'Business rationale for each feature',
                'Testing and validation results'
            ]
        },
        'Fair Lending': {
            'description': 'Features must not discriminate',
            'requirements': [
                'No protected class information as direct features',
                'Features must be business-justified',
                'Fairness testing required'
            ]
        },
        'GDPR/CCPA': {
            'description': 'Data privacy requirements',
            'requirements': [
                'Features derived from personal data documented',
                'Right to explanation for feature usage',
                'Data minimization principle'
            ]
        }
    }
    
    for reg, details in requirements.items():
        print(f"\n📋 {reg} - {details['description']}")
        for req in details['requirements']:
            print(f"  • {req}")
    
    print("\n💡 Recommendations:")
    print("  1. Document feature engineering decisions")
    print("  2. Maintain data lineage for all features")
    print("  3. Test features for fairness and bias")
    print("  4. Review features with compliance team")
    print("  5. Keep feature engineering logs for audit")

# Run compliance check
check_regulatory_compliance_for_features(feature_names)

7.2 Business Impact of Feature Engineering

# ============= BUSINESS IMPACT =============

def analyze_feature_engineering_impact():
    """Analyze the business impact of feature engineering."""
    
    print("\n" + "="*60)
    print("FEATURE ENGINEERING BUSINESS IMPACT")
    print("="*60)
    
    impacts = {
        'Credit Scoring': {
            'benefit': 'Better risk assessment',
            'features': 'DTI ratio, credit utilization, payment history',
            'estimated_impact': 'Reduced defaults by 15-25%'
        },
        'Customer Segmentation': {
            'benefit': 'Targeted marketing',
            'features': 'RFM scores, segment categories',
            'estimated_impact': 'Increased marketing ROI by 30-50%'
        },
        'Fraud Detection': {
            'benefit': 'Better fraud identification',
            'features': 'Transaction patterns, velocity features',
            'estimated_impact': 'Reduced fraud losses by 20-40%'
        },
        'Churn Prediction': {
            'benefit': 'Customer retention',
            'features': 'Behavioral trends, recency metrics',
            'estimated_impact': 'Reduced churn by 10-20%'
        }
    }
    
    for area, details in impacts.items():
        print(f"\n📊 {area}:")
        print(f"  Benefit: {details['benefit']}")
        print(f"  Key Features: {details['features']}")
        print(f"  Estimated Impact: {details['estimated_impact']}")

analyze_feature_engineering_impact()

SECTION 8: SUMMARY FOR THE DATA PRACTITIONER

8.1 The 1-Minute Elevator Pitch

“Feature engineering transforms raw banking data into predictive signals. We create domain-specific features like debt-to-income ratios and credit utilization, RFM scores for customer segmentation, time-based features for transaction patterns, and interaction features that capture complex relationships. Well-engineered features improve model performance, enhance interpretability, and satisfy regulatory requirements. Feature engineering is where domain expertise meets data science.”

8.2 Key Takeaways

Feature engineering creates new features from raw data to improve model performance.
Financial ratios (DTI, utilization, savings rate) are powerful predictors.
RFM analysis (Recency, Frequency, Monetary) is essential for customer analytics.
Time-based features capture patterns in transaction data.
Interaction features capture relationships between variables.
Validation ensures engineered features are meaningful and correct.
Documentation is required for regulatory compliance (SR 11-7).
Business context should guide feature engineering decisions.
Feature engineering can significantly improve model performance.
Feature selection identifies the most valuable engineered features.

8.3 Recommended Next Steps

Apply feature engineering to your banking datasets
Validate engineered features
Document feature engineering decisions
Test feature impact on model performance
Build automated feature engineering pipelines

[END OF LESSON 5]

Diploma in Financial Data Analytics LESSON 5: FEATURE ENGINEERING FOR FINANCIAL DATA

Diploma in Financial Data Analytics