""" Production Feature Engineering for Prediction When predicting SOH, we don't have SOH yet - need to: 1. Predict SOH iteratively (for rolling features) 2. Or use only features that don't require SOH 3. Or bootstrap initial SOH estimate This version handles production prediction scenarios. """ import pandas as pd import numpy as np from typing import Optional, Dict import warnings warnings.filterwarnings('ignore') from feature_engineering_causal import calculate_energy_throughput def engineer_features_production( df: pd.DataFrame, initial_capacity: float, predicted_soh_series: Optional[pd.Series] = None, use_fixed_bounds: bool = True ): """ Engineer causal features for production prediction. When SOH is not available (production prediction): - Uses capacity to estimate SOH: soh_estimate = capacity / initial_capacity - Creates rolling features based on estimated SOH - Or uses predicted_soh_series if available (iterative prediction) Parameters: ----------- df : DataFrame BMS data (may not have 'soh' column) initial_capacity : float Initial battery capacity predicted_soh_series : Series, optional Predicted SOH values from previous iterations (for iterative prediction) use_fixed_bounds : bool Use fixed bounds for normalization Returns: -------- DataFrame: Features ready for model prediction """ df_feat = df.copy() if 'cycle_index' not in df_feat.columns: raise ValueError("cycle_index column required") # 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) # ======================================================================== if 'voltage' in df_feat.columns: features['voltage_mean'] = cycle_groups['voltage'].mean() if use_fixed_bounds: features['voltage_norm'] = (features['voltage_mean'] - 0) / (5.0 - 0) else: 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() if use_fixed_bounds: 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() 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'] # ======================================================================== # 2. ESTIMATE OR USE SOH # ======================================================================== # If SOH not in data, estimate from capacity if 'soh' not in df_feat.columns: if predicted_soh_series is not None: # Use predicted SOH from previous iteration features['soh'] = features['cycle_index'].map(predicted_soh_series).fillna(features['capacity_norm']) else: # Estimate SOH from capacity (simple approach for production) features['soh'] = features['capacity_norm'].clip(0, 1) else: features['soh'] = cycle_groups['soh'].mean() # ======================================================================== # 3. STATISTICAL FEATURES (causal - current cycle only) # ======================================================================== 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() # ======================================================================== # 4. TEMPORAL FEATURES (causal) # ======================================================================== features['cycles_from_start'] = features['cycle_index'] - features['cycle_index'].min() 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 # ======================================================================== # 5. DEGRADATION FEATURES (causal - backward-looking only) # ======================================================================== # These use SOH, which we've estimated above features['soh_fade_rate'] = features['soh'].diff().fillna(0) features['soh_fade_rate_abs'] = features['soh_fade_rate'].abs() 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) # ======================================================================== # 6. PHYSICS-INFORMED FEATURES (causal) # ======================================================================== if 'current' in df_feat.columns: 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() # Fill any remaining NaN values features = features.fillna(0) return features def engineer_features_iterative( df: pd.DataFrame, initial_capacity: float, model, feature_cols: list, max_iterations: int = 3 ): """ Iterative feature engineering for production prediction. Approach: 1. Estimate SOH from capacity initially 2. Engineer features with estimated SOH 3. Predict SOH 4. Use predicted SOH to re-engineer features 5. Re-predict SOH (iterative refinement) Parameters: ----------- df : DataFrame BMS data (no SOH column) initial_capacity : float Initial capacity model : RandomForestRegressor Trained model feature_cols : list Feature columns expected by model max_iterations : int Maximum iterations for refinement Returns: -------- DataFrame: Features with refined SOH predictions """ predicted_soh = None for iteration in range(max_iterations): # Engineer features with current SOH estimate features = engineer_features_production( df, initial_capacity, predicted_soh_series=predicted_soh, use_fixed_bounds=True ) # Predict SOH X = features[feature_cols].values new_predicted_soh = model.predict(X) predicted_soh = pd.Series(new_predicted_soh, index=features['cycle_index']) # Check convergence if iteration > 0: change = np.abs(new_predicted_soh - old_predicted_soh).max() if change < 0.001: # Converged break old_predicted_soh = new_predicted_soh.copy() # Final feature engineering with converged SOH features = engineer_features_production( df, initial_capacity, predicted_soh_series=predicted_soh, use_fixed_bounds=True ) return features