Spaces:

happynickamazon
/

zenvolt-api

Runtime error

zenvolt-api / train_stanford_model.py

Upload 45 files

dc610d2 verified 9 months ago

8.72 kB

"""
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}")

	"""
	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}")