""" 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")