--- name: python-data-scientist description: | Guide for applied machine learning with scikit-learn covering feature engineering, model selection, pipeline construction, evaluation, hyperparameter tuning, and production-ready model patterns. Use when the user asks about python data scientist, related techniques, best practices, or needs guidance in this domain. Do NOT use when the request is outside the scope of python data scientist or requires a different specialized skill. license: Apache-2.0 metadata: author: foundry-skills version: "1.0.0" tags: "data-science statistics guide python testing performing-arts" category: "data-analysis" subcategory: "statistics-modeling" depends: "" disclaimer: "none" difficulty: "intermediate" --- # Python Data Scientist You are an expert applied data scientist who builds robust machine learning pipelines with scikit-learn, engineering features methodically, selecting models systematically, and evaluating results rigorously. ## When to Use **Use this skill when:** - User asks about python data scientist techniques or best practices - User needs guidance on python data scientist concepts - User wants to implement or improve their approach to python data scientist **Do NOT use when:** - The request falls outside the scope of python data scientist - User needs a different specialized skill for their specific situation - The topic requires professional consultation beyond general guidance ## Project Structure ``` ml-project/ ├── data/ │ ├── raw/ # Immutable original data │ ├── processed/ # Cleaned, feature-engineered data │ └── external/ # Third-party reference data ├── src/ │ ├── data/ │ │ ├── ingestion.py # Data loading │ │ └── validation.py # Schema checks │ ├── features/ │ │ ├── engineering.py # Feature transforms │ │ └── selection.py # Feature selection │ ├── models/ │ │ ├── train.py # Training pipeline │ │ ├── evaluate.py # Metrics and reports │ │ └── predict.py # Inference │ └── utils/ │ └── config.py # Hyperparameters, paths ├── notebooks/ │ ├── 01-exploration.ipynb │ └── 02-modeling.ipynb ├── models/ # Serialized model artifacts ├── tests/ └── pyproject.toml ``` ## Feature Engineering ### Numeric Features ```python from sklearn.pipeline import Pipeline from sklearn.preprocessing import ( StandardScaler, MinMaxScaler, RobustScaler, PowerTransformer, QuantileTransformer ) from sklearn.impute import SimpleImputer import numpy as np numeric_pipeline = Pipeline([ ('imputer', SimpleImputer(strategy='median')), ('scaler', RobustScaler()), # Robust to outliers ]) # When to use each scaler: # StandardScaler - Normal-ish data, linear models, SVMs # MinMaxScaler - Neural networks, bounded features # RobustScaler - Data with outliers # PowerTransformer - Skewed distributions (Box-Cox, Yeo-Johnson) # QuantileTransformer - Force uniform or normal distribution ``` ### Categorical Features ```python from sklearn.preprocessing import ( OneHotEncoder, OrdinalEncoder, TargetEncoder ) # One-hot: low cardinality (<15 categories) ohe = OneHotEncoder( drop='if_binary', # Drop redundant column for binary handle_unknown='ignore', # Handle unseen categories at predict time sparse_output=True, # Memory efficient for high-dim min_frequency=0.01, # Group rare categories ) # Ordinal: ordered categories ordinal = OrdinalEncoder( categories=[['low', 'medium', 'high', 'critical']], handle_unknown='use_encoded_value', unknown_value=-1, ) # Target encoding: high cardinality (cities, zip codes) target_enc = TargetEncoder( smooth='auto', # Regularization target_type='continuous', ) ``` ### Date/Time Features ```python def extract_datetime_features(df, col): """Extract useful features from a datetime column.""" df = df.copy() dt = df[col] df[f'{col}_year'] = dt.dt.year df[f'{col}_month'] = dt.dt.month df[f'{col}_day_of_week'] = dt.dt.dayofweek df[f'{col}_hour'] = dt.dt.hour df[f'{col}_is_weekend'] = dt.dt.dayofweek.isin([5, 6]).astype(int) df[f'{col}_quarter'] = dt.dt.quarter df[f'{col}_day_of_year'] = dt.dt.dayofyear # Cyclical encoding for periodic features df[f'{col}_month_sin'] = np.sin(2 * np.pi * dt.dt.month / 12) df[f'{col}_month_cos'] = np.cos(2 * np.pi * dt.dt.month / 12) df[f'{col}_hour_sin'] = np.sin(2 * np.pi * dt.dt.hour / 24) df[f'{col}_hour_cos'] = np.cos(2 * np.pi * dt.dt.hour / 24) return df ``` ### Custom Transformer ```python from sklearn.base import BaseEstimator, TransformerMixin class InteractionFeatures(BaseEstimator, TransformerMixin): """Create interaction features between specified column pairs.""" def __init__(self, interaction_pairs): self.interaction_pairs = interaction_pairs def fit(self, X, y=None): return self def transform(self, X): X = X.copy() for col_a, col_b in self.interaction_pairs: X[f'{col_a}_x_{col_b}'] = X[col_a] * X[col_b] X[f'{col_a}_div_{col_b}'] = X[col_a] / X[col_b].replace(0, np.nan) return X def get_feature_names_out(self, input_features=None): names = list(input_features) if input_features else [] for col_a, col_b in self.interaction_pairs: names.extend([f'{col_a}_x_{col_b}', f'{col_a}_div_{col_b}']) return names ``` ## Pipeline Construction ### Full ML Pipeline ```python from sklearn.compose import ColumnTransformer from sklearn.pipeline import Pipeline from sklearn.ensemble import GradientBoostingClassifier # Define column groups numeric_features = ['age', 'income', 'tenure_months', 'num_products'] categorical_features = ['region', 'plan_type', 'channel'] # Preprocessing preprocessor = ColumnTransformer( transformers=[ ('num', Pipeline([ ('imputer', SimpleImputer(strategy='median')), ('scaler', StandardScaler()), ]), numeric_features), ('cat', Pipeline([ ('imputer', SimpleImputer(strategy='constant', fill_value='unknown')), ('encoder', OneHotEncoder(handle_unknown='ignore', sparse_output=False)), ]), categorical_features), ], remainder='drop', verbose_feature_names_out=False, ) # Full pipeline pipeline = Pipeline([ ('preprocessor', preprocessor), ('classifier', GradientBoostingClassifier( n_estimators=200, max_depth=5, learning_rate=0.1, random_state=42, )), ]) # Fit pipeline.fit(X_train, y_train) # Predict y_pred = pipeline.predict(X_test) y_proba = pipeline.predict_proba(X_test)[:, 1] ``` ## Model Selection Guide ### By Problem Type | Problem | Start With | Then Try | When to Use | |---------|-----------|----------|-------------| | **Binary classification** | Logistic Regression | GBM, Random Forest | Churn, fraud, click | | **Multi-class** | Random Forest | GBM, SVM | Category prediction | | **Regression** | Ridge/Lasso | GBM, Random Forest | Revenue, pricing | | **Ranking** | LambdaMART (LightGBM) | XGBoost ranker | Search, recommendations | | **Anomaly detection** | Isolation Forest | One-class SVM, LOF | Fraud, outliers | | **Clustering** | K-Means | DBSCAN, HDBSCAN | Segmentation | | **Time series** | ARIMA/Prophet | LSTM, LightGBM | Forecasting | ### Model Comparison ```python from sklearn.model_selection import cross_val_score from sklearn.linear_model import LogisticRegression from sklearn.ensemble import ( RandomForestClassifier, GradientBoostingClassifier ) models = { 'Logistic Regression': LogisticRegression(max_iter=1000, random_state=42), 'Random Forest': RandomForestClassifier(n_estimators=200, random_state=42), 'Gradient Boosting': GradientBoostingClassifier(n_estimators=200, random_state=42), } results = {} for name, model in models.items(): pipe = Pipeline([ ('preprocessor', preprocessor), ('model', model), ]) scores = cross_val_score(pipe, X_train, y_train, cv=5, scoring='roc_auc') results[name] = { 'mean_auc': scores.mean(), 'std_auc': scores.std(), 'scores': scores, } print(f"{name}: AUC = {scores.mean():.4f} (+/- {scores.std():.4f})") ``` ## Model Evaluation ### Classification Metrics ```python from sklearn.metrics import ( classification_report, confusion_matrix, roc_auc_score, average_precision_score, roc_curve, precision_recall_curve ) def evaluate_classifier(y_true, y_pred, y_proba, model_name='Model'): """Comprehensive classification evaluation.""" print(f"=== {model_name} Evaluation ===\n") # Classification report print(classification_report(y_true, y_pred, digits=3)) # AUC metrics roc_auc = roc_auc_score(y_true, y_proba) avg_precision = average_precision_score(y_true, y_proba) print(f"ROC AUC: {roc_auc:.4f}") print(f"Average Precision: {avg_precision:.4f}") # Confusion matrix cm = confusion_matrix(y_true, y_pred) print(f"\nConfusion Matrix:") print(f" TN={cm[0,0]:5d} FP={cm[0,1]:5d}") print(f" FN={cm[1,0]:5d} TP={cm[1,1]:5d}") return {'roc_auc': roc_auc, 'avg_precision': avg_precision} ``` ### Regression Metrics ```python from sklearn.metrics import ( mean_absolute_error, mean_squared_error, r2_score, mean_absolute_percentage_error ) def evaluate_regressor(y_true, y_pred, model_name='Model'): """Comprehensive regression evaluation.""" mae = mean_absolute_error(y_true, y_pred) rmse = mean_squared_error(y_true, y_pred, squared=False) r2 = r2_score(y_true, y_pred) mape = mean_absolute_percentage_error(y_true, y_pred) print(f"=== {model_name} Evaluation ===") print(f"MAE: {mae:.4f}") print(f"RMSE: {rmse:.4f}") print(f"R2: {r2:.4f}") print(f"MAPE: {mape:.2%}") return {'mae': mae, 'rmse': rmse, 'r2': r2, 'mape': mape} ``` ## Hyperparameter Tuning ### Randomized Search ```python from sklearn.model_selection import RandomizedSearchCV from scipy.stats import randint, uniform param_distributions = { 'classifier__n_estimators': randint(100, 500), 'classifier__max_depth': randint(3, 10), 'classifier__learning_rate': uniform(0.01, 0.3), 'classifier__subsample': uniform(0.6, 0.4), 'classifier__min_samples_leaf': randint(5, 50), } search = RandomizedSearchCV( pipeline, param_distributions, n_iter=50, cv=5, scoring='roc_auc', random_state=42, n_jobs=-1, verbose=1, ) search.fit(X_train, y_train) print(f"Best AUC: {search.best_score_:.4f}") print(f"Best params: {search.best_params_}") best_model = search.best_estimator_ ``` ### Optuna Integration ```python import optuna from sklearn.model_selection import cross_val_score def objective(trial): params = { 'classifier__n_estimators': trial.suggest_int('n_estimators', 100, 500), 'classifier__max_depth': trial.suggest_int('max_depth', 3, 10), 'classifier__learning_rate': trial.suggest_float('learning_rate', 0.01, 0.3, log=True), 'classifier__subsample': trial.suggest_float('subsample', 0.6, 1.0), 'classifier__min_samples_leaf': trial.suggest_int('min_samples_leaf', 5, 50), } pipe = pipeline.set_params(**params) scores = cross_val_score(pipe, X_train, y_train, cv=5, scoring='roc_auc') return scores.mean() study = optuna.create_study(direction='maximize') study.optimize(objective, n_trials=100) print(f"Best AUC: {study.best_value:.4f}") print(f"Best params: {study.best_params}") ``` ## Feature Selection ```python from sklearn.feature_selection import ( SelectKBest, f_classif, mutual_info_classif ) from sklearn.inspection import permutation_importance # Method 1: Statistical tests selector = SelectKBest(score_func=f_classif, k=20) X_selected = selector.fit_transform(X_train_processed, y_train) # Method 2: Model-based importance model.fit(X_train_processed, y_train) importances = pd.DataFrame({ 'feature': feature_names, 'importance': model.feature_importances_, }).sort_values('importance', ascending=False) # Method 3: Permutation importance (model-agnostic) perm_importance = permutation_importance( model, X_test_processed, y_test, n_repeats=10, random_state=42, scoring='roc_auc' ) perm_df = pd.DataFrame({ 'feature': feature_names, 'importance_mean': perm_importance.importances_mean, 'importance_std': perm_importance.importances_std, }).sort_values('importance_mean', ascending=False) ``` ## Cross-Validation Strategies ```python from sklearn.model_selection import ( StratifiedKFold, TimeSeriesSplit, GroupKFold, RepeatedStratifiedKFold ) # Imbalanced classification cv = StratifiedKFold(n_splits=5, shuffle=True, random_state=42) # Time series (no data leakage) cv = TimeSeriesSplit(n_splits=5, gap=7) # 7-day gap between train/test # Grouped data (e.g., same user in train and test) cv = GroupKFold(n_splits=5) scores = cross_val_score(pipeline, X, y, cv=cv, groups=user_ids) # More robust estimate cv = RepeatedStratifiedKFold(n_splits=5, n_repeats=3, random_state=42) ``` ## Model Serialization ```python import joblib from pathlib import Path from datetime import datetime def save_model(pipeline, metrics, model_dir='models'): """Save model with metadata.""" model_dir = Path(model_dir) model_dir.mkdir(exist_ok=True) timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') model_path = model_dir / f'model_{timestamp}.joblib' meta_path = model_dir / f'model_{timestamp}_meta.json' # Save model joblib.dump(pipeline, model_path) # Save metadata import json metadata = { 'timestamp': timestamp, 'metrics': metrics, 'features': list(pipeline.named_steps['preprocessor'].get_feature_names_out()), 'model_type': type(pipeline.named_steps['classifier']).__name__, 'model_path': str(model_path), } with open(meta_path, 'w') as f: json.dump(metadata, f, indent=2, default=str) print(f"Model saved to {model_path}") return model_path ``` ## Common Pitfalls | Pitfall | Symptom | Fix | |---------|---------|-----| | Data leakage | Unrealistically high test performance | Fit preprocessing only on train data | | Target leakage | Feature contains future information | Audit feature timestamps | | Class imbalance | High accuracy, low recall | Use stratified CV, SMOTE, class weights | | Overfitting | Train >> test performance | Regularization, simpler model, more data | | Feature scale issues | Linear model ignores some features | Scale all numeric features | | Missing value patterns | Model fails on new data | Handle unknowns in encoders | | Train/serve skew | Good offline, bad online | Use same pipeline for train and predict | | Temporal leakage | Random CV on time series | Use TimeSeriesSplit | ## Process 1. **Gather information.** Ask the user clarifying questions to understand their specific situation, goals, and constraints 2. **Analyze context.** Review the information provided and identify key factors relevant to python data scientist 3. **Develop recommendations.** Apply domain expertise to create actionable guidance tailored to the user's needs 4. **Present structured output.** Deliver findings in the output format below with clear next steps 5. **Address follow-ups.** Answer additional questions and refine recommendations based on feedback ## Output Format ```template ## Python Data Scientist Analysis ### Assessment [Key findings and observations] ### Recommendations 1. [Primary recommendation] 2. [Secondary recommendation] 3. [Additional suggestions] ### Action Items - [ ] [First action step] - [ ] [Second action step] - [ ] [Follow-up task] ``` ## Edge Cases - **Incomplete information:** Ask clarifying questions before proceeding with recommendations - **Conflicting requirements:** Prioritize the most critical constraint and note trade-offs - **Out of scope requests:** Redirect to appropriate specialized skill or professional resource - **Beginner vs advanced:** Adjust depth and terminology based on user's experience level ## Example **Input:** "Help me with python data scientist for my current situation" **Output:** Based on your situation, here is a structured approach to python data scientist: 1. **Assessment:** Evaluate your current state and identify key areas for improvement 2. **Strategy:** Develop a targeted plan based on best practices 3. **Implementation:** Execute the plan with specific, measurable steps 4. **Review:** Monitor progress and adjust as needed