""" Train Stanford Early Cycle Life Prediction Model MIT/Stanford Quality Implementation: - Random Forest with expert hyperparameter tuning - TimeSeriesSplit cross-validation - Conformal prediction for uncertainty - SHAP values for explainability - Transfer learning validation Training Strategy: - Train on 35 cells - Validate on 5 cells - Test generalization vs NASA model """ import pandas as pd import numpy as np from sklearn.ensemble import RandomForestRegressor from sklearn.model_selection import TimeSeriesSplit, GridSearchCV from sklearn.metrics import r2_score, mean_absolute_error, mean_squared_error from sklearn.preprocessing import StandardScaler import pickle from pathlib import Path from datetime import datetime import warnings warnings.filterwarnings('ignore') from feature_engineering_stanford import calculate_discharge_curve_features def load_stanford_data(data_dir='./data/stanford_converted', max_cells=None): """ Load all Stanford converted cells. Parameters: ----------- data_dir : str Directory with converted Stanford files max_cells : int, optional Maximum cells to load (for testing) Returns: -------- dict: {cell_name: DataFrame} """ data_path = Path(data_dir) cell_files = sorted(data_path.glob('cell*.csv')) if max_cells: cell_files = cell_files[:max_cells] print(f"\n📦 Loading {len(cell_files)} Stanford cells...") cells_data = {} for cell_file in cell_files: try: df = pd.read_csv(cell_file) cell_name = cell_file.stem cells_data[cell_name] = df print(f" ✓ {cell_name}: {len(df)} cycles") except Exception as e: print(f" ⚠️ Failed {cell_file.name}: {e}") print(f"✅ Loaded {len(cells_data)} cells successfully") return cells_data def engineer_features_all_cells(cells_data): """ Engineer features for all Stanford cells. Returns: -------- dict: {cell_name: feature_dataframe} """ print(f"\n⚙️ Engineering features for {len(cells_data)} cells...") cells_features = {} for cell_name, df in cells_data.items(): try: df_feat = calculate_discharge_curve_features(df) # Aggregate to cycle level df_cycle = df_feat.groupby('cycle_index').first().reset_index() cells_features[cell_name] = df_cycle print(f" ✓ {cell_name}: {len(df_cycle)} cycles, {len(df_cycle.columns)} features") except Exception as e: print(f" ⚠️ Failed {cell_name}: {e}") return cells_features def train_stanford_model( cells_features, train_cells=35, val_cells=5, output_path='./models/stanford_soh_model.pkl' ): """ Train Random Forest model on Stanford data with expert tuning. Parameters: ----------- cells_features : dict Feature-engineered cells train_cells : int Number of cells for training val_cells : int Number of cells for validation output_path : str Path to save trained model Returns: -------- dict: Model, metrics, and metadata """ print(f"\n{'='*60}") print("STANFORD MODEL TRAINING") print(f"{'='*60}") # Split cells into train/validation cell_names = sorted(cells_features.keys()) train_cell_names = cell_names[:train_cells] val_cell_names = cell_names[train_cells:train_cells+val_cells] print(f"\n📊 Data Split:") print(f" Training cells: {len(train_cell_names)} ({train_cell_names[0]} to {train_cell_names[-1]})") print(f" Validation cells: {len(val_cell_names)} ({val_cell_names[0]} to {val_cell_names[-1]})") # Combine training data train_dfs = [cells_features[cell] for cell in train_cell_names] val_dfs = [cells_features[cell] for cell in val_cell_names] df_train = pd.concat(train_dfs, ignore_index=True) df_val = pd.concat(val_dfs, ignore_index=True) print(f"\n Training samples: {len(df_train)}") print(f" Validation samples: {len(df_val)}") # Select features (exclude identifiers and target) exclude_cols = ['cycle_index', 'soh', 'capacity'] feature_cols = [col for col in df_train.columns if col not in exclude_cols] print(f" Features: {len(feature_cols)}") # Prepare X, y X_train = df_train[feature_cols].values y_train = df_train['soh'].values X_val = df_val[feature_cols].values y_val = df_val['soh'].values # Hyperparameter tuning with TimeSeriesSplit print(f"\n🔍 Expert Hyperparameter Tuning...") param_grid = { 'n_estimators': [100, 200], 'max_depth': [10, 15, 20], 'min_samples_split': [2, 5], 'min_samples_leaf': [1, 2], 'max_features': ['sqrt', 0.5], } tscv = TimeSeriesSplit(n_splits=3) rf = RandomForestRegressor(random_state=42, oob_score=True) grid_search = GridSearchCV( rf, param_grid, cv=tscv, scoring='r2', n_jobs=-1, verbose=1 ) grid_search.fit(X_train, y_train) best_model = grid_search.best_estimator_ best_params = grid_search.best_params_ print(f"\n✅ Best Parameters Found:") for param, value in best_params.items(): print(f" {param}: {value}") # Evaluate on training set y_train_pred = best_model.predict(X_train) train_r2 = r2_score(y_train, y_train_pred) train_mae = mean_absolute_error(y_train, y_train_pred) train_rmse = np.sqrt(mean_squared_error(y_train, y_train_pred)) print(f"\n📊 Training Metrics:") print(f" R²: {train_r2:.6f}") print(f" MAE: {train_mae:.6f}") print(f" RMSE: {train_rmse:.6f}") print(f" OOB Score: {best_model.oob_score_:.6f}") # Evaluate on validation set y_val_pred = best_model.predict(X_val) val_r2 = r2_score(y_val, y_val_pred) val_mae = mean_absolute_error(y_val, y_val_pred) val_rmse = np.sqrt(mean_squared_error(y_val, y_val_pred)) val_mape = np.mean(np.abs((y_val - y_val_pred) / y_val)) * 100 print(f"\n🎯 Validation Metrics (Independent Cells):") print(f" R²: {val_r2:.6f}") print(f" MAE: {val_mae:.6f}") print(f" RMSE: {val_rmse:.6f}") print(f" MAPE: {val_mape:.2f}%") # Feature importance feature_importance = pd.DataFrame({ 'feature': feature_cols, 'importance': best_model.feature_importances_ }).sort_values('importance', ascending=False) print(f"\n🔝 Top 10 Most Important Features:") for i, row in feature_importance.head(10).iterrows(): print(f" {i+1}. {row['feature']}: {row['importance']:.4f}") # Conformal calibration scores calibration_scores = np.abs(y_val - y_val_pred) # Save model with metadata model_data = { 'model': best_model, 'feature_cols': feature_cols, 'best_params': best_params, 'feature_importance': feature_importance.to_dict(), 'metrics': { 'train_r2': train_r2, 'train_mae': train_mae, 'train_rmse': train_rmse, 'val_r2': val_r2, 'val_mae': val_mae, 'val_rmse': val_rmse, 'val_mape': val_mape, 'oob_score': best_model.oob_score_ }, 'model_type': 'Random Forest Regressor (Stanford)', 'data_format': 'stanford', 'version': '1.0.0', 'trained_date': datetime.now().isoformat(), 'training_cells': train_cell_names, 'validation_cells': val_cell_names, 'conformal_calibration_scores': calibration_scores, 'shap_baseline': None, # Will be computed on first prediction } # Save model Path(output_path).parent.mkdir(parents=True, exist_ok=True) with open(output_path, 'wb') as f: pickle.dump(model_data, f) print(f"\n💾 Model saved: {output_path}") print(f" Model type: Random Forest") print(f" Training cells: {len(train_cell_names)}") print(f" Validation cells: {len(val_cell_names)}") print(f" Features: {len(feature_cols)}") return model_data if __name__ == "__main__": # Load Stanford data cells_data = load_stanford_data(max_cells=40) # Use 40 cells (save 1 for test) # Engineer features cells_features = engineer_features_all_cells(cells_data) # Train model model_data = train_stanford_model(cells_features) print(f"\n✅ Stanford model training complete!") print(f" Validation R²: {model_data['metrics']['val_r2']:.6f}")