| 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
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()),
])
Categorical Features
from sklearn.preprocessing import (
OneHotEncoder, OrdinalEncoder, TargetEncoder
)
ohe = OneHotEncoder(
drop='if_binary',
handle_unknown='ignore',
sparse_output=True,
min_frequency=0.01,
)
ordinal = OrdinalEncoder(
categories=[['low', 'medium', 'high', 'critical']],
handle_unknown='use_encoded_value',
unknown_value=-1,
)
target_enc = TargetEncoder(
smooth='auto',
target_type='continuous',
)
Date/Time Features
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
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
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
from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
from sklearn.ensemble import GradientBoostingClassifier
numeric_features = ['age', 'income', 'tenure_months', 'num_products']
categorical_features = ['region', 'plan_type', 'channel']
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,
)
pipeline = Pipeline([
('preprocessor', preprocessor),
('classifier', GradientBoostingClassifier(
n_estimators=200,
max_depth=5,
learning_rate=0.1,
random_state=42,
)),
])
pipeline.fit(X_train, y_train)
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
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
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")
print(classification_report(y_true, y_pred, digits=3))
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}")
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
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
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
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
from sklearn.feature_selection import (
SelectKBest, f_classif, mutual_info_classif
)
from sklearn.inspection import permutation_importance
selector = SelectKBest(score_func=f_classif, k=20)
X_selected = selector.fit_transform(X_train_processed, y_train)
model.fit(X_train_processed, y_train)
importances = pd.DataFrame({
'feature': feature_names,
'importance': model.feature_importances_,
}).sort_values('importance', ascending=False)
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
from sklearn.model_selection import (
StratifiedKFold, TimeSeriesSplit, GroupKFold,
RepeatedStratifiedKFold
)
cv = StratifiedKFold(n_splits=5, shuffle=True, random_state=42)
cv = TimeSeriesSplit(n_splits=5, gap=7)
cv = GroupKFold(n_splits=5)
scores = cross_val_score(pipeline, X, y, cv=cv, groups=user_ids)
cv = RepeatedStratifiedKFold(n_splits=5, n_repeats=3, random_state=42)
Model Serialization
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'
joblib.dump(pipeline, model_path)
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
- Gather information. Ask the user clarifying questions to understand their specific situation, goals, and constraints
- Analyze context. Review the information provided and identify key factors relevant to python data scientist
- Develop recommendations. Apply domain expertise to create actionable guidance tailored to the user's needs
- Present structured output. Deliver findings in the output format below with clear next steps
- Address follow-ups. Answer additional questions and refine recommendations based on feedback
Output Format
## 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:
- Assessment: Evaluate your current state and identify key areas for improvement
- Strategy: Develop a targeted plan based on best practices
- Implementation: Execute the plan with specific, measurable steps
- Review: Monitor progress and adjust as needed