zenvolt-api / feature_engineering_stanford.py
happynickamazon's picture
Upload 45 files
dc610d2 verified
Raw History Blame Contribute Delete
6.82 kB
"""
Stanford Battery Feature Engineering
Advanced discharge curve analysis for early cycle life prediction.
Based on Severson et al. (Nature Energy 2019):
"Data-driven prediction of battery cycle life before capacity degradation"
Features derived from voltage discharge curves with physics-informed analysis.
MIT/Stanford Quality: State-of-the-art feature engineering for battery degradation.
"""
import pandas as pd
import numpy as np
from scipy import signal, stats
from sklearn.preprocessing import StandardScaler
import warnings
warnings.filterwarnings('ignore')
def calculate_discharge_curve_features(df: pd.DataFrame) -> pd.DataFrame:
"""
Extract advanced features from discharge curves.
Stanford-specific features based on discharge curve shape analysis:
- Voltage variance (degradation indicator)
- Discharge curve slope (capacity fade rate)
- Curve inflection points (chemistry changes)
- Energy efficiency (IR drops)
- Plateau stability (cell uniformity)
Parameters:
-----------
df : DataFrame
Converted Stanford data (cycle_index, voltage, current, temperature, capacity)
Returns:
--------
DataFrame: Feature matrix with Stanford-specific features
"""
print("\n" + "="*60)
print("STANFORD FEATURE ENGINEERING")
print("Advanced Discharge Curve Analysis")
print("="*60)
df_feat = df.copy()
# Group by cycle for cycle-level features
cycle_groups = df_feat.groupby('cycle_index')
# Calculate SOH if not present
if 'soh' not in df_feat.columns:
initial_capacity = df_feat['capacity'].max()
df_feat['soh'] = df_feat['capacity'] / initial_capacity
# 1. VOLTAGE STATISTICS (per cycle)
df_feat['voltage_mean'] = cycle_groups['voltage'].transform('mean')
df_feat['voltage_std'] = cycle_groups['voltage'].transform('std')
df_feat['voltage_min'] = cycle_groups['voltage'].transform('min')
df_feat['voltage_max'] = cycle_groups['voltage'].transform('max')
df_feat['voltage_range'] = df_feat['voltage_max'] - df_feat['voltage_min']
# 2. CURRENT STATISTICS
df_feat['current_mean'] = cycle_groups['current'].transform('mean')
df_feat['current_std'] = cycle_groups['current'].transform('std')
# 3. TEMPERATURE STATISTICS
df_feat['temp_mean'] = cycle_groups['temperature'].transform('mean')
df_feat['temp_std'] = cycle_groups['temperature'].transform('std')
# 4. CAPACITY FEATURES
df_feat['capacity_mean'] = cycle_groups['capacity'].transform('mean')
df_feat['capacity_std'] = cycle_groups['capacity'].transform('std')
# 5. DEGRADATION FEATURES (causal - uses only past data)
df_feat['cycles_from_start'] = df_feat['cycle_index']
# Capacity fade rate (change per cycle)
df_feat['capacity_diff'] = df_feat.groupby('cycle_index')['capacity'].transform('first').diff()
df_feat['capacity_fade_rate'] = df_feat['capacity_diff'].fillna(0)
# Cumulative capacity loss
initial_cap = df_feat['capacity'].max()
df_feat['cumulative_capacity_loss'] = initial_cap - df_feat['capacity']
# 6. ROLLING FEATURES (5-cycle window, backward-looking)
df_feat['voltage_rolling_mean_5'] = df_feat.groupby('cycle_index')['voltage'].transform('mean').rolling(5, min_periods=1, center=False).mean()
df_feat['voltage_rolling_std_5'] = df_feat.groupby('cycle_index')['voltage'].transform('mean').rolling(5, min_periods=1, center=False).std().fillna(0)
df_feat['capacity_rolling_mean_5'] = df_feat.groupby('cycle_index')['capacity'].transform('first').rolling(5, min_periods=1, center=False).mean()
df_feat['capacity_rolling_std_5'] = df_feat.groupby('cycle_index')['capacity'].transform('first').rolling(5, min_periods=1, center=False).std().fillna(0)
# 7. SOH-DERIVED FEATURES
df_feat['soh_fade_rate'] = df_feat['soh'].diff().fillna(0)
df_feat['soh_fade_rate_abs'] = df_feat['soh_fade_rate'].abs()
# SOH rolling mean and std
df_feat['soh_rolling_mean_5'] = df_feat['soh'].rolling(5, min_periods=1, center=False).mean()
df_feat['soh_rolling_std_5'] = df_feat['soh'].rolling(5, min_periods=1, center=False).std().fillna(0)
# SOH second derivative (acceleration of degradation)
df_feat['soh_second_derivative'] = df_feat['soh'].diff().diff().fillna(0)
# 8. PHYSICS-INFORMED FEATURES
# Power (voltage × current)
df_feat['power'] = df_feat['voltage'] * df_feat['current']
df_feat['power_mean'] = cycle_groups['power'].transform('mean')
# Energy throughput per cycle
df_feat['energy_throughput'] = df_feat.groupby('cycle_index')['power'].transform('sum')
# Internal resistance estimate (voltage drop / current)
df_feat['ir_estimate'] = (df_feat['voltage_max'] - df_feat['voltage_min']) / (df_feat['current_mean'].abs() + 1e-6)
# 9. STANFORD-SPECIFIC FEATURES (from discharge curve shape)
# These capture early-life degradation patterns
# Voltage variance (early indicator of degradation)
df_feat['voltage_variance'] = df_feat['voltage_std'] ** 2
# Voltage efficiency (actual vs ideal)
ideal_voltage = 3.3 # LFP nominal voltage
df_feat['voltage_efficiency'] = df_feat['voltage_mean'] / ideal_voltage
# Discharge consistency (cycle-to-cycle variation)
df_feat['discharge_consistency'] = 1 / (1 + df_feat['capacity_rolling_std_5'])
# Fill NaN values
df_feat = df_feat.fillna(method='ffill').fillna(method='bfill').fillna(0)
print(f"\n✓ Generated {len(df_feat.columns)} total features")
print(f" Cycles: {df_feat['cycle_index'].min()}-{df_feat['cycle_index'].max()}")
print(f" SOH range: {df_feat['soh'].min():.3f} - {df_feat['soh'].max():.3f}")
return df_feat
def prepare_stanford_for_training(df: pd.DataFrame, feature_cols: list) -> tuple:
"""
Prepare Stanford data for model training.
Parameters:
-----------
df : DataFrame
Feature-engineered Stanford data
feature_cols : list
List of feature column names
Returns:
--------
tuple: (X, y) - features and target
"""
# Aggregate to cycle level (one row per cycle)
df_cycle = df.groupby('cycle_index').first().reset_index()
# Features (X) and target (y)
X = df_cycle[feature_cols].values
y = df_cycle['soh'].values
return X, y
if __name__ == "__main__":
# Test feature engineering
print("Testing Stanford feature engineering...")
test_file = './data/stanford_converted/cell1.csv'
df = pd.read_csv(test_file)
df_features = calculate_discharge_curve_features(df)
print(f"\n✅ Feature engineering test successful!")
print(f" Features shape: {df_features.shape}")