""" Causal Feature Engineering for Battery SOH Prediction FIXED VERSION: All features use only data ≤ current cycle Removes data leakage from: - cycles_to_end (used max cycle - future knowledge) - Normalization (used all cycles - future knowledge) - charge_discharge_asymmetry (used all cycles) """ import pandas as pd import numpy as np from sklearn.preprocessing import MinMaxScaler import warnings warnings.filterwarnings('ignore') def calculate_energy_throughput(df): """Calculate energy throughput: ∫(voltage * current) dt per cycle.""" if 'time' not in df.columns: df = df.copy() df['time'] = df.groupby('cycle_index').cumcount() df['power'] = df['voltage'] * df['current'] energy_per_cycle = df.groupby('cycle_index').apply( lambda x: np.trapz(x['power'], x['time']) if len(x) > 1 else 0 ) return energy_per_cycle def engineer_features_causal(df, initial_capacity, use_fixed_bounds=True): """ Engineer CAUSAL features - uses only data ≤ current cycle. FIXES: 1. Removed cycles_to_end (data leakage) 2. Fixed normalization to use only past cycles or fixed bounds 3. Fixed charge_discharge_asymmetry to use only past cycles 4. Rolling windows verified backward-only (center=False) Parameters: ----------- df : DataFrame Preprocessed DataFrame with cycle_index, voltage, current, temperature, capacity, soh initial_capacity : float Initial capacity value use_fixed_bounds : bool If True, use fixed physical bounds for normalization. If False, use expanding window. Returns: -------- DataFrame: Feature matrix with causal features only """ print("\n" + "="*60) print("CAUSAL FEATURE ENGINEERING (No Data Leakage)") print("="*60) df_feat = df.copy() if 'cycle_index' not in df_feat.columns: print("Error: cycle_index column not found") return None # Sort by cycle_index df_feat = df_feat.sort_values('cycle_index').reset_index(drop=True) cycle_groups = df_feat.groupby('cycle_index') features = pd.DataFrame() features['cycle_index'] = cycle_groups['cycle_index'].first() # ======================================================================== # 1. RAW MEASUREMENTS (causal - current cycle only) # ======================================================================== print("\n1. Raw Measurements (causal)...") if 'voltage' in df_feat.columns: features['voltage_mean'] = cycle_groups['voltage'].mean() # FIX: Use fixed bounds (0-5V typical) or expanding window if use_fixed_bounds: features['voltage_norm'] = (features['voltage_mean'] - 0) / (5.0 - 0) # Fixed 0-5V range else: # Expanding window: use only past cycles features['voltage_norm'] = features['voltage_mean'].expanding().apply( lambda x: (x.iloc[-1] - x.min()) / (x.max() - x.min()) if x.max() > x.min() else 0.5 ) if 'current' in df_feat.columns: features['current_mean'] = cycle_groups['current'].mean() # FIX: Use fixed bounds or expanding window if use_fixed_bounds: # Typical range: -10A to +10A (adjust based on battery specs) features['current_norm'] = (features['current_mean'] + 10) / 20 else: features['current_norm'] = features['current_mean'].expanding().apply( lambda x: (x.iloc[-1] - x.min()) / (x.max() - x.min()) if x.max() > x.min() else 0.5 ) features['is_charge'] = (features['current_mean'] > 0).astype(int) features['is_discharge'] = (features['current_mean'] < 0).astype(int) if 'temperature' in df_feat.columns: features['temperature_mean'] = cycle_groups['temperature'].mean() # FIX: Use fixed bounds (0-50°C typical) or expanding window if use_fixed_bounds: features['temperature_norm'] = (features['temperature_mean'] - 0) / (50.0 - 0) else: features['temperature_norm'] = features['temperature_mean'].expanding().apply( lambda x: (x.iloc[-1] - x.min()) / (x.max() - x.min()) if x.max() > x.min() else 0.5 ) if 'capacity' in df_feat.columns: features['capacity'] = cycle_groups['capacity'].mean() features['capacity_norm'] = features['capacity'] / initial_capacity if initial_capacity > 0 else features['capacity'] if 'soh' in df_feat.columns: features['soh'] = cycle_groups['soh'].mean() # ======================================================================== # 2. STATISTICAL FEATURES (causal - current cycle only) # ======================================================================== print("2. Statistical Features (causal)...") for col in ['voltage', 'current', 'temperature']: if col in df_feat.columns: features[f'{col}_mean'] = cycle_groups[col].mean() features[f'{col}_std'] = cycle_groups[col].std().fillna(0) features[f'{col}_min'] = cycle_groups[col].min() features[f'{col}_max'] = cycle_groups[col].max() features[f'{col}_range'] = features[f'{col}_max'] - features[f'{col}_min'] features[f'{col}_median'] = cycle_groups[col].median() if 'current' in df_feat.columns and 'capacity' in df_feat.columns: charge_capacity = df_feat[df_feat['current'] > 0].groupby('cycle_index')['capacity'].mean() discharge_capacity = df_feat[df_feat['current'] < 0].groupby('cycle_index')['capacity'].mean() features['charge_discharge_diff'] = (charge_capacity - discharge_capacity).fillna(0) if 'voltage' in df_feat.columns and 'current' in df_feat.columns: try: energy = calculate_energy_throughput(df_feat) features['energy_throughput'] = features['cycle_index'].map(energy).fillna(0) except: features['power_mean'] = features['voltage_mean'] * features['current_mean'] features['energy_throughput'] = features['power_mean'].abs() # ======================================================================== # 3. TEMPORAL FEATURES (causal - removed cycles_to_end) # ======================================================================== print("3. Temporal Features (causal)...") features['cycles_from_start'] = features['cycle_index'] - features['cycle_index'].min() # REMOVED: cycles_to_end (data leakage - uses max cycle) if 'time' in df_feat.columns: features['cumulative_time'] = cycle_groups['time'].sum() features['time_per_cycle'] = cycle_groups['time'].mean() else: features['cumulative_time'] = features['cycles_from_start'] features['time_per_cycle'] = 1.0 features['days_since_start'] = features['cycles_from_start'] * 0.5 # ======================================================================== # 4. DEGRADATION FEATURES (causal - backward-looking only) # ======================================================================== print("4. Degradation Features (causal)...") if 'soh' in features.columns: # FIXED: Backward-looking diff (uses only past) features['soh_fade_rate'] = features['soh'].diff().fillna(0) features['soh_fade_rate_abs'] = features['soh_fade_rate'].abs() # FIXED: Rolling windows with center=False (backward-only) features['soh_rolling_mean_5'] = features['soh'].rolling(window=5, min_periods=1, center=False).mean() features['soh_rolling_std_5'] = features['soh'].rolling(window=5, min_periods=1, center=False).std().fillna(0) features['soh_ema'] = features['soh'].ewm(span=5, adjust=False).mean() features['soh_second_derivative'] = features['soh_fade_rate'].diff().fillna(0) features['cumulative_capacity_loss'] = (1 - features['soh']) * initial_capacity if initial_capacity > 0 else (1 - features['soh']) if 'capacity' in features.columns: features['capacity_fade_rate'] = features['capacity'].diff().fillna(0) features['capacity_fade_rate_pct'] = features['capacity'].pct_change().fillna(0) # ======================================================================== # 5. PHYSICS-INFORMED FEATURES (causal) # ======================================================================== print("5. Physics-Informed Features (causal)...") # FIXED: charge_discharge_asymmetry - use only current cycle ratio, not cumulative if 'current' in df_feat.columns: # Calculate per-cycle ratio instead of cumulative charge_by_cycle = df_feat[df_feat['current'] > 0].groupby('cycle_index')['current'].sum() discharge_by_cycle = df_feat[df_feat['current'] < 0].groupby('cycle_index')['current'].sum().abs() features['charge_discharge_asymmetry'] = (charge_by_cycle / discharge_by_cycle).fillna(1.0) features['charge_discharge_asymmetry'] = features['cycle_index'].map(features['charge_discharge_asymmetry']).fillna(1.0) if 'temperature' in df_feat.columns: temp_stress = cycle_groups['temperature'].apply(lambda x: (x > 40).sum()) features['temperature_stress'] = temp_stress / cycle_groups['temperature'].count() features['max_temperature'] = cycle_groups['temperature'].max() features['min_temperature'] = cycle_groups['temperature'].min() features['temperature_range'] = features['max_temperature'] - features['min_temperature'] if 'voltage' in df_feat.columns and 'current' in df_feat.columns: discharge_voltage_var = df_feat[df_feat['current'] < 0].groupby('cycle_index')['voltage'].var().fillna(0) features['voltage_variance_discharge'] = features['cycle_index'].map(discharge_voltage_var).fillna(0) if 'voltage_mean' in features.columns and 'current_mean' in features.columns: if 'power_mean' not in features.columns: features['power_mean'] = features['voltage_mean'] * features['current_mean'] features['power_abs'] = features['power_mean'].abs() # ======================================================================== # 6. TARGET VARIABLES (for training only - not used in production) # ======================================================================== print("6. Target Variables (for training only)...") if 'soh' in features.columns: # These are TARGETS, not features - only used during training features['soh_10_cycles_ahead'] = features['soh'].shift(-10).fillna(features['soh'].iloc[-1]) eol_threshold = 0.80 features['remaining_cycles_to_eol'] = None for idx, row in features.iterrows(): future_soh = features[features['cycle_index'] > row['cycle_index']]['soh'] eol_cycles = future_soh[future_soh < eol_threshold] if len(eol_cycles) > 0: eol_cycle = eol_cycles.index[0] features.loc[idx, 'remaining_cycles_to_eol'] = features.loc[eol_cycle, 'cycle_index'] - row['cycle_index'] else: # Estimate based on degradation rate if EOL not reached current_degradation = row['soh_fade_rate'] if 'soh_fade_rate' in row else 0 if current_degradation < 0: cycles_to_eol = (row['soh'] - eol_threshold) / abs(current_degradation) features.loc[idx, 'remaining_cycles_to_eol'] = max(0, cycles_to_eol) else: features.loc[idx, 'remaining_cycles_to_eol'] = 1000 # Large number if not degrading features['remaining_cycles_to_eol'] = features['remaining_cycles_to_eol'].fillna(0) # Fill any remaining NaN values features = features.fillna(0) print(f"\n✓ Causal feature engineering complete: {len(features)} cycles, {len(features.columns)} features") print(f" Removed data leakage features: cycles_to_end") print(f" Fixed normalization: Using fixed bounds or expanding window") return features def engineer_features_for_all_batteries_causal(dataframes_dict, initial_capacities_dict, use_fixed_bounds=True): """Engineer causal features for all batteries.""" all_features = {} for battery_id, df in dataframes_dict.items(): print(f"\n{'='*60}") print(f"Processing {battery_id} (Causal Features)") print(f"{'='*60}") initial_capacity = initial_capacities_dict.get(battery_id, df['capacity'].max() if 'capacity' in df.columns else 1.0) features = engineer_features_causal(df, initial_capacity, use_fixed_bounds) if features is not None: all_features[battery_id] = features return all_features if __name__ == "__main__": import sys sys.path.append('.') from unified_data_pipeline import preprocess_all_datasets dataframes, summary = preprocess_all_datasets() if dataframes: initial_capacities = {} for battery_id, df in dataframes.items(): initial_capacities[battery_id] = df['capacity'].max() if 'capacity' in df.columns else 1.0 all_features = engineer_features_for_all_batteries_causal(dataframes, initial_capacities) import os os.makedirs('./data/features_causal', exist_ok=True) for battery_id, features in all_features.items(): output_path = f'./data/features_causal/{battery_id}_features_causal.csv' features.to_csv(output_path, index=False) print(f"\nSaved causal features to: {output_path}")