feature_audit.py

"""
Feature Audit: Detect Data Leakage in Battery SOH Prediction Features

This script audits all 52 features for causality violations:
- Uses only data ≤ current cycle
- No future information leakage
- Document each feature's causality status
"""

import pandas as pd
import numpy as np
from typing import Dict, List, Tuple
import warnings
warnings.filterwarnings('ignore')


class FeatureAuditor:
    """Audit features for data leakage and causality violations."""
    
    def __init__(self):
        self.leakage_issues = []
        self.safe_features = []
        self.warnings = []
        
    def audit_feature(self, feature_name: str, calculation_method: str, 
                     uses_future_data: bool, description: str) -> Dict:
        """Audit a single feature."""
        status = "❌ LEAKAGE" if uses_future_data else "✅ SAFE"
        result = {
            'feature': feature_name,
            'status': status,
            'leakage': uses_future_data,
            'method': calculation_method,
            'description': description
        }
        
        if uses_future_data:
            self.leakage_issues.append(result)
        else:
            self.safe_features.append(result)
            
        return result
    
    def generate_audit_report(self) -> str:
        """Generate comprehensive audit report."""
        report = []
        report.append("="*80)
        report.append("FEATURE CAUSALITY AUDIT REPORT")
        report.append("="*80)
        report.append(f"\nDate: {pd.Timestamp.now().strftime('%Y-%m-%d %H:%M:%S')}")
        report.append(f"\nTotal Features Audited: {len(self.leakage_issues) + len(self.safe_features)}")
        report.append(f"❌ Features with Leakage: {len(self.leakage_issues)}")
        report.append(f"✅ Causal Features: {len(self.safe_features)}")
        
        if self.leakage_issues:
            report.append("\n" + "="*80)
            report.append("🚨 DATA LEAKAGE ISSUES FOUND")
            report.append("="*80)
            for issue in self.leakage_issues:
                report.append(f"\n❌ {issue['feature']}")
                report.append(f"   Method: {issue['method']}")
                report.append(f"   Issue: {issue['description']}")
                report.append(f"   Action Required: FIX OR REMOVE")
        
        report.append("\n" + "="*80)
        report.append("✅ SAFE CAUSAL FEATURES")
        report.append("="*80)
        
        # Group by category
        categories = {}
        for feature in self.safe_features:
            category = feature['method'].split('_')[0] if '_' in feature['method'] else 'Other'
            if category not in categories:
                categories[category] = []
            categories[category].append(feature)
        
        for category, features in categories.items():
            report.append(f"\n[{category.upper()}]")
            for feature in features:
                report.append(f"  ✅ {feature['feature']}")
        
        report.append("\n" + "="*80)
        report.append("RECOMMENDATIONS")
        report.append("="*80)
        report.append("\n1. Remove or fix all features with ❌ LEAKAGE status")
        report.append("2. Retrain model with only causal features")
        report.append("3. Re-validate on independent test set")
        report.append("4. Report honest R² score (even if lower)")
        report.append("5. Document uncertainty quantification methodology")
        
        return "\n".join(report)


def audit_all_features():
    """Audit all features in the engineering pipeline."""
    auditor = FeatureAuditor()
    
    # ========================================================================
    # 1. RAW MEASUREMENTS - SAFE (uses only current cycle)
    # ========================================================================
    auditor.audit_feature(
        'voltage_mean', 'current_cycle', False,
        'Mean voltage within current cycle - CAUSAL'
    )
    auditor.audit_feature(
        'current_mean', 'current_cycle', False,
        'Mean current within current cycle - CAUSAL'
    )
    auditor.audit_feature(
        'temperature_mean', 'current_cycle', False,
        'Mean temperature within current cycle - CAUSAL'
    )
    auditor.audit_feature(
        'capacity', 'current_cycle', False,
        'Capacity at current cycle - CAUSAL'
    )
    
    # Normalization - CHECK: uses min/max across ALL cycles
    auditor.audit_feature(
        'voltage_norm', 'minmax_normalization', True,
        '⚠️ NORMALIZATION LEAKAGE: Uses min/max across all cycles (future data) - FIX: Use only past cycles or fixed bounds'
    )
    auditor.audit_feature(
        'current_norm', 'minmax_normalization', True,
        '⚠️ NORMALIZATION LEAKAGE: Uses min/max across all cycles - FIX: Use only past cycles or fixed bounds'
    )
    auditor.audit_feature(
        'temperature_norm', 'minmax_normalization', True,
        '⚠️ NORMALIZATION LEAKAGE: Uses min/max across all cycles - FIX: Use only past cycles or fixed bounds'
    )
    
    # ========================================================================
    # 2. STATISTICAL FEATURES - SAFE (within current cycle)
    # ========================================================================
    for stat in ['mean', 'std', 'min', 'max', 'range', 'median']:
        for signal in ['voltage', 'current', 'temperature']:
            auditor.audit_feature(
                f'{signal}_{stat}', 'current_cycle', False,
                f'{stat.capitalize()} {signal} within current cycle - CAUSAL'
            )
    
    # ========================================================================
    # 3. TEMPORAL FEATURES - CHECK
    # ========================================================================
    auditor.audit_feature(
        'cycles_from_start', 'temporal', False,
        'Cycles since start - CAUSAL (only uses past)'
    )
    auditor.audit_feature(
        'cycles_to_end', 'temporal', True,
        '🚨 MAJOR LEAKAGE: Uses max(cycle_index) which requires future knowledge - REMOVE'
    )
    auditor.audit_feature(
        'days_since_start', 'temporal', False,
        'Days since start (estimated) - CAUSAL'
    )
    auditor.audit_feature(
        'cumulative_time', 'temporal', False,
        'Cumulative time up to current cycle - CAUSAL'
    )
    
    # ========================================================================
    # 4. DEGRADATION FEATURES - CHECK ROLLING WINDOWS
    # ========================================================================
    auditor.audit_feature(
        'soh_fade_rate', 'diff_backward', False,
        'SOH difference from previous cycle - CAUSAL (uses only past)'
    )
    auditor.audit_feature(
        'soh_rolling_mean_5', 'rolling_window', False,
        'Rolling mean (window=5, center=False) - CAUSAL if backward-looking only'
    )
    auditor.audit_feature(
        'soh_rolling_std_5', 'rolling_window', False,
        'Rolling std (window=5, center=False) - CAUSAL if backward-looking only'
    )
    auditor.audit_feature(
        'soh_ema', 'exponential_moving_average', False,
        'Exponential moving average (backward) - CAUSAL'
    )
    auditor.audit_feature(
        'capacity_fade_rate', 'diff_backward', False,
        'Capacity change from previous cycle - CAUSAL'
    )
    
    # ========================================================================
    # 5. PHYSICS-INFORMED FEATURES - SAFE (current cycle)
    # ========================================================================
    auditor.audit_feature(
        'energy_throughput', 'current_cycle', False,
        'Energy throughput within current cycle - CAUSAL'
    )
    auditor.audit_feature(
        'power_mean', 'current_cycle', False,
        'Mean power (V×I) in current cycle - CAUSAL'
    )
    auditor.audit_feature(
        'temperature_stress', 'current_cycle', False,
        'Temperature stress indicators in current cycle - CAUSAL'
    )
    auditor.audit_feature(
        'charge_discharge_asymmetry', 'cumulative', True,
        '⚠️ POTENTIAL LEAKAGE: Uses sum across all cycles - FIX: Use only past cycles or current cycle ratio'
    )
    
    return auditor


if __name__ == "__main__":
    print("Starting Feature Causality Audit...")
    auditor = audit_all_features()
    
    report = auditor.generate_audit_report()
    print("\n" + report)
    
    # Save report
    with open('./data_leakage_audit.txt', 'w') as f:
        f.write(report)
    
    print(f"\n✓ Audit report saved to: ./data_leakage_audit.txt")
    print(f"✓ Found {len(auditor.leakage_issues)} features with leakage")
    print(f"✓ Found {len(auditor.safe_features)} causal features")