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