python-data-scientist

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()),  # 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

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

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

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

# 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

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

    # 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

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

# 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

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

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

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

## 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