senior-data-scientist

name	senior-data-scientist
description	Skill de cientista de dados sênior de nível mundial especializada em modelagem estatística, design de experimentos, inferência causal e analytics preditivo. Cobre teste A/B (dimensionamento de amostra, testes z de duas proporções, correção de Bonferroni), difference-in-differences, pipelines de feature engineering (Scikit-learn, XGBoost), avaliação de modelo com validação cruzada (AUC-ROC, AUC-PR, SHAP) e rastreamento de experimentos MLflow — usando Python (NumPy, Pandas, Scikit-learn), R e SQL. Use quando projetar ou analisar experimentos controlados, construir e avaliar modelos de classificação ou regressão, realizar análise causal em dados observacionais, fazer feature engineering para datasets tabulares estruturados ou traduzir achados estatísticos em decisões de negócio baseadas em dados.
agents	["claude-code"]

Cientista de Dados Sênior

Skill de cientista de dados sênior de nível mundial para sistemas de IA/ML/Dados em produção.

Fluxos de Trabalho Principais

1. Projetar um Teste A/B

import numpy as np
from scipy import stats

def calculate_sample_size(baseline_rate, mde, alpha=0.05, power=0.8):
    """
    Calcular o tamanho de amostra necessário por variante.
    baseline_rate: taxa de conversão atual (ex.: 0.10)
    mde: efeito mínimo detectável (relativo, ex.: 0.05 = 5% de lift)
    """
    p1 = baseline_rate
    p2 = baseline_rate * (1 + mde)
    effect_size = abs(p2 - p1) / np.sqrt((p1 * (1 - p1) + p2 * (1 - p2)) / 2)
    z_alpha = stats.norm.ppf(1 - alpha / 2)
    z_beta = stats.norm.ppf(power)
    n = ((z_alpha + z_beta) / effect_size) ** 2
    return int(np.ceil(n))

def analyze_experiment(control, treatment, alpha=0.05):
    """
    Executar teste z de duas proporções e retornar resultados estruturados.
    control/treatment: dicts com 'conversions' e 'visitors'.
    """
    p_c = control["conversions"] / control["visitors"]
    p_t = treatment["conversions"] / treatment["visitors"]
    pooled = (control["conversions"] + treatment["conversions"]) / (control["visitors"] + treatment["visitors"])
    se = np.sqrt(pooled * (1 - pooled) * (1 / control["visitors"] + 1 / treatment["visitors"]))
    z = (p_t - p_c) / se
    p_value = 2 * (1 - stats.norm.cdf(abs(z)))
    ci_low = (p_t - p_c) - stats.norm.ppf(1 - alpha / 2) * se
    ci_high = (p_t - p_c) + stats.norm.ppf(1 - alpha / 2) * se
    return {
        "lift": (p_t - p_c) / p_c,
        "p_value": p_value,
        "significant": p_value < alpha,
        "ci_95": (ci_low, ci_high),
    }

# --- Checklist do experimento ---
# 1. Definir UMA métrica primária e pré-registrar métricas secundárias.
# 2. Calcular tamanho de amostra ANTES de iniciar: calculate_sample_size(0.10, 0.05)
# 3. Randomizar no nível do usuário (não da sessão) para evitar vazamento.
# 4. Executar por pelo menos 1 ciclo de negócio completo (tipicamente 2 semanas).
# 5. Verificar sample ratio mismatch: abs(n_control - n_treatment) / expected < 0.01
# 6. Analisar com analyze_experiment() e reportar lift + IC, não apenas p-value.
# 7. Aplicar correção de Bonferroni para múltiplas métricas: alpha / n_metrics

2. Construir um Pipeline de Feature Engineering

import pandas as pd
import numpy as np
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.impute import SimpleImputer
from sklearn.compose import ColumnTransformer

def build_feature_pipeline(numeric_cols, categorical_cols, date_cols=None):
    """
    Retorna um ColumnTransformer pronto para ajuste com dados tabulares estruturados.
    """
    numeric_pipeline = Pipeline([
        ("impute", SimpleImputer(strategy="median")),
        ("scale",  StandardScaler()),
    ])
    categorical_pipeline = Pipeline([
        ("impute", SimpleImputer(strategy="most_frequent")),
        ("encode", OneHotEncoder(handle_unknown="ignore", sparse_output=False)),
    ])
    transformers = [
        ("num", numeric_pipeline, numeric_cols),
        ("cat", categorical_pipeline, categorical_cols),
    ]
    return ColumnTransformer(transformers, remainder="drop")

def add_time_features(df, date_col):
    """Extrair features cíclicas e de lag de uma coluna datetime."""
    df = df.copy()
    df[date_col] = pd.to_datetime(df[date_col])
    df["dow_sin"] = np.sin(2 * np.pi * df[date_col].dt.dayofweek / 7)
    df["dow_cos"] = np.cos(2 * np.pi * df[date_col].dt.dayofweek / 7)
    df["month_sin"] = np.sin( * np.pi * df[date_col].dt.month / )
    df[] = np.cos( * np.pi * df[date_col].dt.month / )
    df[] = (df[date_col].dt.dayofweek >= ).astype()
     df

3. Treinar, Avaliar e Selecionar um Modelo de Predição

from sklearn.model_selection import StratifiedKFold, cross_validate
from sklearn.metrics import make_scorer, roc_auc_score, average_precision_score
import xgboost as xgb
import mlflow

SCORERS = {
    "roc_auc":  make_scorer(roc_auc_score, needs_proba=True),
    "avg_prec": make_scorer(average_precision_score, needs_proba=True),
}

def evaluate_model(model, X, y, cv=5):
    """
    Validação cruzada retornando média ± desvio padrão para cada scorer.
    Use StratifiedKFold para classificação para preservar o balanceamento de classes.
    """
    cv_results = cross_validate(
        model, X, y,
        cv=StratifiedKFold(n_splits=cv, shuffle=True, random_state=42),
        scoring=SCORERS,
        return_train_score=True,
    )
    summary = {}
    for metric in SCORERS:
        test_scores = cv_results[f"test_{metric}"]
        summary[metric] = {"mean": test_scores.mean(), "std": test_scores.std()}
        # Sinalizar overfitting: grande gap entre pontuação de treino e teste
        train_mean = cv_results[f"train_{metric}"].mean()
        summary[metric]["overfit_gap"] = train_mean - test_scores.mean()
    return summary

def train_and_log(model, X_train, y_train, X_test, y_test, run_name):
    """Treinar modelo e registrar todos os artefatos no MLflow."""
    with mlflow.start_run(run_name=run_name):
        model.fit(X_train, y_train)
        proba = model.predict_proba(X_test)[:, 1]
        metrics = {
            :  roc_auc_score(y_test, proba),
            : average_precision_score(y_test, proba),
        }
        mlflow.log_params(model.get_params())
        mlflow.log_metrics(metrics)
        mlflow.sklearn.log_model(model, )
         metrics

4. Inferência Causal: Difference-in-Differences

import statsmodels.formula.api as smf

def diff_in_diff(df, outcome, treatment_col, post_col, controls=None):
    """
    Estimar ATT via OLS DiD com covariáveis opcionais.
    df deve ter: outcome, treatment_col (0/1), post_col (0/1).
    Retorna o coeficiente de interação (treatment × post) e seu p-value.
    """
    covariates = " + ".join(controls) if controls else ""
    formula = (
        f"{outcome} ~ {treatment_col} * {post_col}"
        + (f" + {covariates}" if covariates else "")
    )
    result = smf.ols(formula, data=df).fit(cov_type="HC3")
    interaction = f"{treatment_col}:{post_col}"
    return {
        "att":     result.params[interaction],
        "p_value": result.pvalues[interaction],
        "ci_95":   result.conf_int().loc[interaction].tolist(),
        "summary": result.summary(),
    }

# --- Checklist de inferência causal ---
# 1. Validar tendências paralelas no pré-período antes de confiar nas estimativas DiD.
# 2. Usar erros padrão robustos HC3 para lidar com heterocedasticidade.
# 3. Para dados em painel, clusterizar SEs no nível da unidade (adicionar parâmetro groups= ao fit).
# 4. Considerar propensity score matching se os grupos diferirem no baseline.
# 5. Reportar o ATT com intervalo de confiança, não apenas significância estatística.

Documentação de Referência

• Métodos Estatísticos: references/statistical_methods_advanced.md
• Frameworks de Design de Experimentos: references/experiment_design_frameworks.md
• Padrões de Feature Engineering: references/feature_engineering_patterns.md

Comandos Comuns

# Testes e linting
python -m pytest tests/ -v --cov=src/
python -m black src/ && python -m pylint src/

# Treinamento e avaliação
python scripts/train.py --config prod.yaml
python scripts/evaluate.py --model best.pth

# Implantação
docker build -t service:v1 .
kubectl apply -f k8s/
helm upgrade service ./charts/

# Monitoramento e saúde
kubectl logs -f deployment/service
python scripts/health_check.py