Spaces:
Runtime error
Runtime error
| """ | |
| 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") | |