Preguntas de entrevista de Machine Learning con código (40+)

from sklearn.tree import DecisionTreeClassifier
from sklearn.model_selection import cross_val_score
import numpy as np

# Bias alto: árbol muy superficial (underfitting)
tree_shallow = DecisionTreeClassifier(max_depth=1)
scores = cross_val_score(tree_shallow, X, y, cv=5, scoring='accuracy')
print(f"Bias alto — train≈test≈bajo: {scores.mean():.3f}")

# Varianza alta: árbol sin límite (overfitting)
tree_deep = DecisionTreeClassifier()
scores = cross_val_score(tree_deep, X, y, cv=5, scoring='accuracy')
print(f"Varianza alta — train>>test: {scores.mean():.3f}")

import numpy as np

def split_dataset(X, y, train=0.7, val=0.15, test=0.15, seed=42):
    assert abs(train + val + test - 1.0) < 1e-6, "Las proporciones deben sumar 1"
    np.random.seed(seed)
    n = len(X)
    idx = np.random.permutation(n)
    t = int(n * train)
    v = int(n * (train + val))
    return (X[idx[:t]], y[idx[:t]],
            X[idx[t:v]], y[idx[t:v]],
            X[idx[v:]], y[idx[v:]])

X_tr, y_tr, X_val, y_val, X_test, y_test = split_dataset(X, y)

import torch
import torch.nn as nn

def l1_regularization(model, lambda_=1e-4):
    l1 = sum(p.abs().sum() for p in model.parameters())
    return lambda_ * l1

def l2_regularization(model, lambda_=1e-4):
    l2 = sum((p ** 2).sum() for p in model.parameters())
    return lambda_ * l2

# En el loop de entrenamiento:
loss = criterion(outputs, labels)
loss += l1_regularization(model)  # o l2
optimizer.zero_grad()
loss.backward()
optimizer.step()

import numpy as np

def sgd(X, y, lr=0.01, epochs=100, batch_size=32):
    n, d = X.shape
    w = np.zeros(d)
    b = 0.0

    for epoch in range(epochs):
        idx = np.random.permutation(n)
        X_s, y_s = X[idx], y[idx]

        for i in range(0, n, batch_size):
            xb = X_s[i:i+batch_size]
            yb = y_s[i:i+batch_size]

            # Gradiente para regresión lineal (MSE)
            preds = xb @ w + b
            error = preds - yb
            dw = (2 / len(xb)) * xb.T @ error
            db = (2 / len(xb)) * error.sum()

            w -= lr * dw
            b -= lr * db
    return w, b

import pandas as pd
import numpy as np

def remove_outliers_iqr(df, col, factor=1.5):
    Q1 = df[col].quantile(0.25)
    Q3 = df[col].quantile(0.75)
    IQR = Q3 - Q1
    lower = Q1 - factor * IQR
    upper = Q3 + factor * IQR
    return df[(df[col] >= lower) & (df[col] <= upper)]

# Z-score para distribuciones aproximadamente normales
def remove_outliers_zscore(df, col, threshold=3):
    z = (df[col] - df[col].mean()) / df[col].std()
    return df[z.abs() < threshold]

# Opción menos agresiva: clipping
df['feature'] = df['feature'].clip(lower=df['feature'].quantile(0.01),
                                    upper=df['feature'].quantile(0.99))

import numpy as np

class LogisticRegression:
    def __init__(self, lr=0.01, epochs=1000):
        self.lr = lr
        self.epochs = epochs

    def sigmoid(self, z):
        return 1 / (1 + np.exp(-np.clip(z, -500, 500)))

    def fit(self, X, y):
        n, d = X.shape
        self.w = np.zeros(d)
        self.b = 0.0
        for _ in range(self.epochs):
            z = X @ self.w + self.b
            p = self.sigmoid(z)
            dw = (1/n) * X.T @ (p - y)
            db = (1/n) * (p - y).sum()
            self.w -= self.lr * dw
            self.b -= self.lr * db
        return self

    def predict_proba(self, X):
        return self.sigmoid(X @ self.w + self.b)

    def predict(self, X, threshold=0.5):
        return (self.predict_proba(X) >= threshold).astype(int)

import numpy as np

def gini_impurity(y):
    if len(y) == 0:
        return 0
    classes, counts = np.unique(y, return_counts=True)
    probs = counts / len(y)
    return 1 - np.sum(probs ** 2)

def gini_split(y_left, y_right):
    n = len(y_left) + len(y_right)
    return (len(y_left)/n) * gini_impurity(y_left) + \
           (len(y_right)/n) * gini_impurity(y_right)

# Encontrar el mejor split para una feature continua
def best_split(X_col, y):
    thresholds = np.unique(X_col)
    best_gini = float('inf')
    best_thresh = None
    for t in thresholds:
        left = y[X_col <= t]
        right = y[X_col > t]
        g = gini_split(left, right)
        if g < best_gini:
            best_gini = g
            best_thresh = t
    return best_thresh, best_gini

from sklearn.ensemble import RandomForestClassifier
import numpy as np

# Random Forest = bagging de árboles con feature randomness
rf = RandomForestClassifier(
    n_estimators=100,
    max_features='sqrt',   # solo sqrt(n_features) en cada split
    max_samples=0.8,       # 80% del training set por árbol (bagging)
    random_state=42
)
rf.fit(X_train, y_train)

# Feature importance: promedio de reducción de impureza
importance = pd.Series(rf.feature_importances_, index=feature_names)
importance.sort_values().plot.barh()

from sklearn.svm import SVC
from sklearn.preprocessing import StandardScaler
from sklearn.pipeline import Pipeline

# SVM es sensible a la escala: siempre escalar
svm_pipeline = Pipeline([
    ('scaler', StandardScaler()),
    ('svm', SVC(kernel='rbf', C=1.0, gamma='scale', probability=True))
])

svm_pipeline.fit(X_train, y_train)

import numpy as np

def kmeans(X, k, max_iter=300, tol=1e-4, seed=42):
    np.random.seed(seed)
    idx = np.random.choice(len(X), k, replace=False)
    centroids = X[idx].copy()

    for _ in range(max_iter):
        dists = np.linalg.norm(X[:, None] - centroids[None], axis=2)
        labels = np.argmin(dists, axis=1)

        new_centroids = np.array([X[labels == j].mean(axis=0)
                                   if (labels == j).any() else centroids[j]
                                   for j in range(k)])

        if np.linalg.norm(new_centroids - centroids) < tol:
            break
        centroids = new_centroids

    return labels, centroids

labels, centroids = kmeans(X, k=3)

import numpy as np

def confusion_matrix_manual(y_true, y_pred):
    TP = ((y_true == 1) & (y_pred == 1)).sum()
    TN = ((y_true == 0) & (y_pred == 0)).sum()
    FP = ((y_true == 0) & (y_pred == 1)).sum()
    FN = ((y_true == 1) & (y_pred == 0)).sum()
    return TP, TN, FP, FN

def metrics(y_true, y_pred):
    TP, TN, FP, FN = confusion_matrix_manual(y_true, y_pred)
    precision = TP / (TP + FP + 1e-9)
    recall    = TP / (TP + FN + 1e-9)
    f1        = 2 * precision * recall / (precision + recall + 1e-9)
    accuracy  = (TP + TN) / len(y_true)
    return {'precision': precision, 'recall': recall, 'f1': f1, 'accuracy': accuracy}

# Ejemplo: fraude (1% positivos)
y_true = np.array([0]*99 + [1])
y_pred = np.zeros(100, dtype=int)
print(metrics(y_true, y_pred))  # accuracy=0.99, f1=0.0

import numpy as np

def roc_auc_manual(y_true, y_scores):
    thresholds = np.sort(np.unique(y_scores))[::-1]
    tprs, fprs = [0], [0]
    pos = (y_true == 1).sum()
    neg = (y_true == 0).sum()

    for t in thresholds:
        preds = (y_scores >= t).astype(int)
        tp = ((preds == 1) & (y_true == 1)).sum()
        fp = ((preds == 1) & (y_true == 0)).sum()
        tprs.append(tp / pos)
        fprs.append(fp / neg)

    tprs.append(1); fprs.append(1)
    auc = np.trapz(tprs, fprs)
    return abs(auc)

auc = roc_auc_manual(y_test, model.predict_proba(X_test)[:, 1])
print(f"AUC-ROC: {auc:.4f}")

from sklearn.model_selection import StratifiedKFold, cross_val_score
from sklearn.ensemble import GradientBoostingClassifier

skf = StratifiedKFold(n_splits=5, shuffle=True, random_state=42)
model = GradientBoostingClassifier()

scores = cross_val_score(model, X, y, cv=skf, scoring='roc_auc')
print(f"AUC: {scores.mean():.3f} ± {scores.std():.3f}")

def train_with_early_stopping(model, X_tr, y_tr, X_val, y_val,
                               patience=10, metric_fn=None):
    best_val = float('inf')
    best_epoch = 0
    best_weights = None
    no_improve = 0

    for epoch in range(1000):
        model.partial_fit(X_tr, y_tr)
        val_loss = metric_fn(model, X_val, y_val)

        if val_loss < best_val - 1e-4:
            best_val = val_loss
            best_epoch = epoch
            best_weights = copy.deepcopy(model.__dict__)
            no_improve = 0
        else:
            no_improve += 1

        if no_improve >= patience:
            print(f"Early stop en epoch {epoch}, mejor: {best_epoch}")
            model.__dict__.update(best_weights)
            break

    return model

from sklearn.preprocessing import StandardScaler, MinMaxScaler, RobustScaler

# StandardScaler: media 0, std 1 — para algoritmos que asumen distribución normal
ss = StandardScaler()

# MinMaxScaler: [0,1] — para redes neuronales con sigmoid/tanh de output
mm = MinMaxScaler()

# RobustScaler: usa IQR — ideal cuando hay outliers que no querés eliminar
rs = RobustScaler()

# SIEMPRE: fit solo en train, transform en val/test
ss.fit(X_train)
X_train_s = ss.transform(X_train)
X_test_s  = ss.transform(X_test)

from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.impute import SimpleImputer
from sklearn.linear_model import LogisticRegression

# MAL: leakage — los parámetros del scaler ven el test set
X_scaled = StandardScaler().fit_transform(X)
X_train, X_test, y_train, y_test = train_test_split(X_scaled, y)

# BIEN: el pipeline aplica fit solo al training fold
pipe = Pipeline([
    ('imputer', SimpleImputer(strategy='median')),
    ('scaler', StandardScaler()),
    ('clf', LogisticRegression())
])
pipe.fit(X_train, y_train)
pipe.score(X_test, y_test)

import pandas as pd
from sklearn.preprocessing import OrdinalEncoder, OneHotEncoder
from category_encoders import TargetEncoder

# One-Hot: para categóricas sin orden con cardinalidad baja (<20)
ohe = OneHotEncoder(sparse_output=False, handle_unknown='ignore')

# Ordinal: cuando HAY orden real (ej: bajo/medio/alto)
oe = OrdinalEncoder(categories=[['bajo', 'medio', 'alto']])

# Target Encoding: para alta cardinalidad (ciudad, ID de producto)
te = TargetEncoder(smoothing=10)

# Para cardinalidad altísima (>1000): feature hashing
from sklearn.feature_extraction import FeatureHasher
fh = FeatureHasher(n_features=64, input_type='string')

import pandas as pd
import numpy as np
from sklearn.impute import SimpleImputer, KNNImputer

def smart_impute(df, strategy='auto'):
    df = df.copy()
    for col in df.columns:
        if df[col].isnull().sum() == 0:
            continue
        if df[col].dtype in ['float64', 'int64']:
            df[col] = SimpleImputer(strategy='median').fit_transform(
                df[[col]]).ravel()
        else:
            df[col] = df[col].fillna(df[col].mode()[0])
    return df

# Alternativa potente: KNN imputer para variables correlacionadas
knn_imp = KNNImputer(n_neighbors=5)
X_imputed = knn_imp.fit_transform(X_train)

from sklearn.ensemble import BaggingClassifier, AdaBoostClassifier
from sklearn.tree import DecisionTreeClassifier

# Bagging: modelos paralelos, independientes, en subsets aleatorios
bagging = BaggingClassifier(
    estimator=DecisionTreeClassifier(max_depth=10),
    n_estimators=100,
    max_samples=0.8,
    bootstrap=True
)

# Boosting: modelos secuenciales, cada uno aprende de los errores del anterior
boosting = AdaBoostClassifier(
    estimator=DecisionTreeClassifier(max_depth=1),
    n_estimators=100,
    learning_rate=0.1
)

import optuna
import xgboost as xgb
from sklearn.model_selection import cross_val_score

def objective(trial):
    params = {
        'max_depth': trial.suggest_int('max_depth', 3, 10),
        'learning_rate': trial.suggest_float('learning_rate', 1e-3, 0.3, log=True),
        'n_estimators': trial.suggest_int('n_estimators', 50, 500),
        'subsample': trial.suggest_float('subsample', 0.5, 1.0),
        'colsample_bytree': trial.suggest_float('colsample_bytree', 0.5, 1.0),
        'min_child_weight': trial.suggest_int('min_child_weight', 1, 10),
        'reg_alpha': trial.suggest_float('reg_alpha', 1e-5, 10, log=True),
        'eval_metric': 'logloss',
        'use_label_encoder': False
    }
    model = xgb.XGBClassifier(**params, random_state=42)
    return cross_val_score(model, X_train, y_train, cv=5, scoring='roc_auc').mean()

study = optuna.create_study(direction='maximize')
study.optimize(objective, n_trials=100)
print(f"Mejores params: {study.best_params}")

import numpy as np

class MLP:
    def __init__(self, layers):
        self.weights = []
        self.biases = []
        for i in range(len(layers)-1):
            w = np.random.randn(layers[i], layers[i+1]) * np.sqrt(2/layers[i])
            b = np.zeros((1, layers[i+1]))
            self.weights.append(w)
            self.biases.append(b)

    def relu(self, z): return np.maximum(0, z)
    def relu_grad(self, z): return (z > 0).astype(float)
    def sigmoid(self, z): return 1 / (1 + np.exp(-z))

    def forward(self, X):
        self.zs, self.as_ = [], [X]
        a = X
        for i, (w, b) in enumerate(zip(self.weights, self.biases)):
            z = a @ w + b
            self.zs.append(z)
            a = self.relu(z) if i < len(self.weights)-1 else self.sigmoid(z)
            self.as_.append(a)
        return a

    def backward(self, y, lr=0.01):
        n = len(y)
        delta = self.as_[-1] - y.reshape(-1, 1)
        for i in reversed(range(len(self.weights))):
            dw = self.as_[i].T @ delta / n
            db = delta.mean(axis=0, keepdims=True)
            self.weights[i] -= lr * dw
            self.biases[i] -= lr * db
            if i > 0:
                delta = (delta @ self.weights[i].T) * self.relu_grad(self.zs[i-1])

import torch
import torch.nn as nn

class MLPConBN(nn.Module):
    def __init__(self, in_dim, hidden, out_dim):
        super().__init__()
        self.net = nn.Sequential(
            nn.Linear(in_dim, hidden),
            nn.BatchNorm1d(hidden),
            nn.ReLU(),
            nn.Dropout(0.3),
            nn.Linear(hidden, hidden),
            nn.BatchNorm1d(hidden),
            nn.ReLU(),
            nn.Linear(hidden, out_dim)
        )

    def forward(self, x):
        return self.net(x)

import torch
import torch.nn as nn
from torch.utils.data import DataLoader, TensorDataset

def train_epoch(model, loader, optimizer, criterion, device):
    model.train()
    total_loss = 0
    for X_batch, y_batch in loader:
        X_batch, y_batch = X_batch.to(device), y_batch.to(device)
        optimizer.zero_grad()
        outputs = model(X_batch)
        loss = criterion(outputs, y_batch)
        loss.backward()
        torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)
        optimizer.step()
        total_loss += loss.item()
    return total_loss / len(loader)

def eval_epoch(model, loader, criterion, device):
    model.eval()
    total_loss = 0
    with torch.no_grad():
        for X_batch, y_batch in loader:
            X_batch, y_batch = X_batch.to(device), y_batch.to(device)
            outputs = model(X_batch)
            total_loss += criterion(outputs, y_batch).item()
    return total_loss / len(loader)

device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
model = MLPConBN(X_train.shape[1], 128, 1).to(device)
optimizer = torch.optim.AdamW(model.parameters(), lr=1e-3, weight_decay=1e-4)
criterion = nn.BCEWithLogitsLoss()
scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=50)

from sklearn.feature_extraction.text import TfidfVectorizer
from sentence_transformers import SentenceTransformer

corpus = ["La reunión fue cancelada", "El equipo tuvo una junta", "El partido fue suspendido"]

# TF-IDF: bag of words ponderado — rápido, interpretable, no entiende semántica
tfidf = TfidfVectorizer(ngram_range=(1,2), max_features=10000)
X_tfidf = tfidf.fit_transform(corpus)

# Embeddings: representación densa — captura "reunión" ≈ "junta"
model = SentenceTransformer('paraphrase-multilingual-mpnet-base-v2')
X_emb = model.encode(corpus)

from sklearn.metrics.pairwise import cosine_similarity
sims = cosine_similarity(X_emb)
print(f"Sim(reunión, junta): {sims[0,1]:.3f}")  # ~0.85

import faiss
import numpy as np
from sentence_transformers import SentenceTransformer

encoder = SentenceTransformer('paraphrase-multilingual-mpnet-base-v2')
docs = ["Python tutorial for beginners", "Machine learning crash course",
        "Deep learning with PyTorch", "Data structures and algorithms"]

embeddings = encoder.encode(docs, normalize_embeddings=True)
dim = embeddings.shape[1]

index = faiss.IndexFlatIP(dim)
index.add(embeddings.astype(np.float32))

query = "intro a ML para principiantes"
q_emb = encoder.encode([query], normalize_embeddings=True).astype(np.float32)
scores, indices = index.search(q_emb, k=3)
for i, (score, idx) in enumerate(zip(scores[0], indices[0])):
    print(f"{i+1}. [{score:.3f}] {docs[idx]}")

from scipy.stats import ks_2samp, chi2_contingency
import numpy as np

def detect_drift_numeric(ref_data, prod_data, threshold=0.05):
    stat, p_value = ks_2samp(ref_data, prod_data)
    return {'statistic': stat, 'p_value': p_value, 'drift': p_value < threshold}

def detect_drift_categorical(ref_data, prod_data):
    ref_counts = pd.Series(ref_data).value_counts()
    prod_counts = pd.Series(prod_data).value_counts()
    all_cats = set(ref_counts.index) | set(prod_counts.index)
    ref_freq = [ref_counts.get(c, 0) for c in all_cats]
    prod_freq = [prod_counts.get(c, 0) for c in all_cats]
    _, p_value, _, _ = chi2_contingency([ref_freq, prod_freq])
    return {'p_value': p_value, 'drift': p_value < 0.05}

drift = detect_drift_numeric(train_data['age'], prod_data['age'])
if drift['drift']:
    print(f"DRIFT DETECTADO en 'age' (p={drift['p_value']:.4f}) — reentrenar")

import mlflow
import mlflow.sklearn
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import roc_auc_score

mlflow.set_experiment("churn-prediction")

with mlflow.start_run(run_name="rf-v2"):
    params = {'n_estimators': 200, 'max_depth': 8, 'min_samples_leaf': 10}
    mlflow.log_params(params)

    model = RandomForestClassifier(**params, random_state=42)
    model.fit(X_train, y_train)

    train_auc = roc_auc_score(y_train, model.predict_proba(X_train)[:, 1])
    val_auc = roc_auc_score(y_val, model.predict_proba(X_val)[:, 1])
    mlflow.log_metric("train_auc", train_auc)
    mlflow.log_metric("val_auc", val_auc)
    mlflow.log_metric("overfit_gap", train_auc - val_auc)

    mlflow.sklearn.log_model(model, "model",
                              registered_model_name="ChurnRF",
                              input_example=X_train[:5])

import numpy as np
from scipy.stats import ttest_ind

class ModelABTest:
    def __init__(self, model_a, model_b, split=0.5):
        self.model_a = model_a
        self.model_b = model_b
        self.split = split
        self.results_a, self.results_b = [], []

    def route(self, user_id):
        return 'A' if hash(str(user_id)) % 100 < self.split * 100 else 'B'

    def predict(self, X, user_id):
        model = self.model_a if self.route(user_id) == 'A' else self.model_b
        return model.predict(X)

    def log_outcome(self, user_id, metric_value):
        if self.route(user_id) == 'A':
            self.results_a.append(metric_value)
        else:
            self.results_b.append(metric_value)

    def significance_test(self, alpha=0.05):
        stat, p = ttest_ind(self.results_a, self.results_b)
        winner = 'A' if np.mean(self.results_a) > np.mean(self.results_b) else 'B'
        significant = p < alpha
        return {'p_value': p, 'winner': winner if significant else 'no winner yet',
                'mean_a': np.mean(self.results_a), 'mean_b': np.mean(self.results_b)}

import numpy as np

class MatrixFactorization:
    def __init__(self, n_factors=20, lr=0.01, reg=0.1, epochs=50):
        self.k = n_factors
        self.lr = lr
        self.reg = reg
        self.epochs = epochs

    def fit(self, R):
        n_users, n_items = R.shape
        self.P = np.random.normal(0, 0.1, (n_users, self.k))
        self.Q = np.random.normal(0, 0.1, (n_items, self.k))
        mask = ~np.isnan(R)

        for epoch in range(self.epochs):
            for u, i in zip(*np.where(mask)):
                err = R[u, i] - self.P[u] @ self.Q[i]
                self.P[u] += self.lr * (err * self.Q[i] - self.reg * self.P[u])
                self.Q[i] += self.lr * (err * self.P[u] - self.reg * self.Q[i])

    def predict(self, user, item):
        return self.P[user] @ self.Q[item]

    def recommend(self, user, top_k=10, seen_items=None):
        scores = self.P[user] @ self.Q.T
        if seen_items:
            scores[seen_items] = -np.inf
        return np.argsort(scores)[::-1][:top_k]

import numpy as np

def precision_at_k(recommended, relevant, k):
    recs_k = set(recommended[:k])
    return len(recs_k & set(relevant)) / k

def recall_at_k(recommended, relevant, k):
    recs_k = set(recommended[:k])
    rel = set(relevant)
    return len(recs_k & rel) / len(rel) if rel else 0

def ndcg_at_k(recommended, relevant, k):
    def dcg(items, relevant_set, k):
        return sum(1/np.log2(i+2) for i, item in enumerate(items[:k])
                   if item in relevant_set)
    ideal = sorted(recommended[:k], key=lambda x: x in set(relevant), reverse=True)
    idcg = dcg(ideal, set(relevant), k)
    return dcg(recommended, set(relevant), k) / idcg if idcg > 0 else 0

recs = [3, 1, 5, 7, 2]
relevant = {1, 3, 6}
print(f"P@3: {precision_at_k(recs, relevant, 3):.3f}")
print(f"NDCG@5: {ndcg_at_k(recs, relevant, 5):.3f}")

import torch
import torch.nn as nn
import torch.nn.functional as F
import math

class MultiHeadAttention(nn.Module):
    def __init__(self, d_model, n_heads):
        super().__init__()
        assert d_model % n_heads == 0
        self.d_k = d_model // n_heads
        self.n_heads = n_heads
        self.W_q = nn.Linear(d_model, d_model)
        self.W_k = nn.Linear(d_model, d_model)
        self.W_v = nn.Linear(d_model, d_model)
        self.W_o = nn.Linear(d_model, d_model)

    def split_heads(self, x, batch):
        return x.view(batch, -1, self.n_heads, self.d_k).transpose(1, 2)

    def forward(self, Q, K, V, mask=None):
        batch = Q.size(0)
        Q = self.split_heads(self.W_q(Q), batch)
        K = self.split_heads(self.W_k(K), batch)
        V = self.split_heads(self.W_v(V), batch)

        scores = Q @ K.transpose(-2, -1) / math.sqrt(self.d_k)
        if mask is not None:
            scores = scores.masked_fill(mask == 0, -1e9)
        attn = F.softmax(scores, dim=-1)

        out = (attn @ V).transpose(1, 2).contiguous()
        out = out.view(batch, -1, self.n_heads * self.d_k)
        return self.W_o(out), attn

import torch.nn as nn

class SentimentLSTM(nn.Module):
    def __init__(self, vocab_size, embed_dim, hidden_dim, n_layers):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, embed_dim, padding_idx=0)
        self.lstm = nn.LSTM(embed_dim, hidden_dim,
                            num_layers=n_layers,
                            batch_first=True,
                            dropout=0.3,
                            bidirectional=True)
        self.fc = nn.Linear(hidden_dim * 2, 1)
        self.dropout = nn.Dropout(0.3)

    def forward(self, x):
        emb = self.dropout(self.embedding(x))
        out, (hidden, cell) = self.lstm(emb)
        hidden = torch.cat([hidden[-2], hidden[-1]], dim=1)
        return self.fc(self.dropout(hidden))

class FraudDetectionSystem:
    def __init__(self):
        self.feature_store = FeatureStore()
        self.model = LightGBMModel.load('fraud_model_v3')
        self.rules = RuleEngine()

    def score_transaction(self, txn: dict) -> dict:
        features = self.feature_store.get_features(
            user_id=txn['user_id'],
            window=['1m', '1h', '24h']
        )

        rule_verdict = self.rules.evaluate(txn, features)
        if rule_verdict['action'] == 'block':
            return rule_verdict

        score = self.model.predict_proba(features)[0][1]
        threshold = self.get_threshold(txn['merchant_category'])
        action = 'block' if score > 0.9 else 'review' if score > threshold else 'approve'

        return {'action': action, 'score': score, 'features': features}

from prefect import flow, task

@task(retries=2)
def extract_training_data(start_date, end_date):
    return query_feature_store(start_date, end_date)

@task
def validate_data(df):
    assert df.shape[0] > MIN_ROWS, "Datos insuficientes"
    assert df['label'].mean() > 0.001, "Sin positivos suficientes"
    return df

@task
def train_model(df, best_params):
    model = xgb.XGBClassifier(**best_params)
    model.fit(df.drop('label', axis=1), df['label'])
    return model

@task
def evaluate_and_promote(model, holdout):
    auc = roc_auc_score(holdout['label'],
                        model.predict_proba(holdout.drop('label', axis=1))[:, 1])
    current_auc = get_production_model_metric('auc')
    if auc > current_auc - 0.005:
        promote_to_production(model)
        return True
    return False

@flow(name="weekly-retraining")
def retrain_pipeline():
    df = extract_training_data(weeks_back=8)
    df = validate_data(df)
    model = train_model(df, load_best_params())
    evaluate_and_promote(model, load_holdout())

import numpy as np

def bootstrap_ci(data, statistic_fn, n_bootstrap=10000, ci=0.95, seed=42):
    np.random.seed(seed)
    n = len(data)
    bootstrap_stats = [statistic_fn(np.random.choice(data, size=n, replace=True))
                       for _ in range(n_bootstrap)]

    alpha = (1 - ci) / 2
    return {
        'estimate': statistic_fn(data),
        'ci_lower': np.percentile(bootstrap_stats, alpha * 100),
        'ci_upper': np.percentile(bootstrap_stats, (1 - alpha) * 100),
        'ci_level': ci
    }

import numpy as np

class GaussianNaiveBayes:
    def fit(self, X, y):
        self.classes = np.unique(y)
        self.priors = {}
        self.means = {}
        self.stds = {}
        for c in self.classes:
            X_c = X[y == c]
            self.priors[c] = len(X_c) / len(X)
            self.means[c] = X_c.mean(axis=0)
            self.stds[c]  = X_c.std(axis=0) + 1e-9

    def gaussian_pdf(self, x, mean, std):
        return np.exp(-((x - mean) ** 2) / (2 * std ** 2)) / (np.sqrt(2 * np.pi) * std)

    def predict(self, X):
        preds = []
        for x in X:
            posteriors = {c: np.log(self.priors[c]) + np.sum(
                np.log(self.gaussian_pdf(x, self.means[c], self.stds[c])))
                for c in self.classes}
            preds.append(max(posteriors, key=posteriors.get))
        return np.array(preds)

# CÓDIGO CON BUG
X_scaled = StandardScaler().fit_transform(X)  # BUG: fit en todo el dataset
X_train, X_test, y_train, y_test = train_test_split(X_scaled, y, test_size=0.2)

# CORRECCIÓN
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2)
scaler = StandardScaler()
X_train_s = scaler.fit_transform(X_train)
X_test_s  = scaler.transform(X_test)  # solo transform

# 1. Leakage de features temporales
df.sort_values('date', inplace=True)  # siempre splitear por tiempo

# 2. Train-test distribution shift
from scipy.stats import ks_2samp
for col in features:
    stat, p = ks_2samp(train[col].dropna(), prod[col].dropna())
    if p < 0.05:
        print(f"DRIFT: {col} (p={p:.4f})")

# 3. Label leak
corr_with_target = X_train.corrwith(pd.Series(y_train))
suspicious = corr_with_target[corr_with_target.abs() > 0.9]
print(f"Features sospechosas: {suspicious.index.tolist()}")

# 4. Concept drift
print(f"Train positive rate: {y_train.mean():.4f}")
print(f"Prod positive rate: {y_prod.mean():.4f}")

import json
import hashlib
from datetime import datetime

class PredictionLogger:
    def __init__(self, model_name, model_version, storage_backend):
        self.model_name = model_name
        self.model_version = model_version
        self.storage = storage_backend

    def log(self, input_features: dict, prediction: float,
            confidence: float, user_id: str = None):
        record = {
            'timestamp': datetime.utcnow().isoformat(),
            'model_name': self.model_name,
            'model_version': self.model_version,
            'prediction_id': hashlib.md5(
                f"{user_id}{datetime.utcnow()}".encode()).hexdigest()[:16],
            'user_id': user_id,
            'prediction': round(float(prediction), 6),
            'confidence': round(float(confidence), 6),
            'feature_hash': hashlib.sha256(
                json.dumps(input_features, sort_keys=True).encode()).hexdigest()[:16],
        }
        self.storage.write(record)
        return record['prediction_id']

import pandas as pd
import numpy as np
from sklearn.feature_selection import mutual_info_classif, VarianceThreshold

# 1. Entender la estructura
print(df.info(memory_usage='deep'))
print(df.isnull().sum().sort_values(ascending=False).head(20))

# 2. Eliminar features con varianza cero
vt = VarianceThreshold(threshold=0.01)
X_filtered = vt.fit_transform(X)

# 3. Eliminar features altamente correlacionadas
corr_matrix = pd.DataFrame(X_filtered).corr().abs()
upper = corr_matrix.where(np.triu(np.ones(corr_matrix.shape), k=1).astype(bool))
to_drop = [col for col in upper.columns if any(upper[col] > 0.95)]
X_reduced = pd.DataFrame(X_filtered).drop(columns=to_drop)

# 4. Mutual information para feature importance
mi = mutual_info_classif(X_reduced, y, random_state=42)
top_features = np.argsort(mi)[::-1][:50]

def build_churn_features(user_id, events_df, billing_df, support_df):
    features = {}

    for window in [7, 30, 90]:
        recent = events_df[events_df['user_id'] == user_id].tail(window)
        features[f'logins_{window}d'] = recent[recent['event'] == 'login'].shape[0]
        features[f'core_actions_{window}d'] = recent[recent['core_feature'] == 1].shape[0]
        features[f'days_active_{window}d'] = recent['date'].nunique()

    features['engagement_trend'] = (
        features['logins_7d'] / (features['logins_30d'] / 4 + 1e-6)
    )

    b = billing_df[billing_df['user_id'] == user_id]
    features['days_since_last_payment'] = (pd.Timestamp.now() - b['paid_at'].max()).days
    features['failed_payments_last_3m'] = b[b['status'] == 'failed']['amount'].count()
    features['mrr'] = b['mrr'].iloc[-1] if len(b) else 0

    s = support_df[support_df['user_id'] == user_id]
    features['open_tickets'] = s[s['status'] == 'open'].shape[0]
    features['tickets_last_30d'] = s[s['created_at'] > pd.Timestamp.now() - pd.Timedelta('30d')].shape[0]
    features['account_age_days'] = (pd.Timestamp.now() - b['created_at'].min()).days

    return features

from sklearn.calibration import CalibratedClassifierCV, calibration_curve
import matplotlib.pyplot as plt

def plot_calibration(y_true, y_prob, model_name="Modelo"):
    frac_positives, mean_pred = calibration_curve(y_true, y_prob, n_bins=10)
    plt.plot(mean_pred, frac_positives, 's-', label=model_name)
    plt.plot([0,1], [0,1], 'k--', label='Perfecto')
    plt.xlabel('Probabilidad predicha')
    plt.ylabel('Fracción real de positivos')
    plt.legend()

uncalibrated = xgb.XGBClassifier().fit(X_train, y_train)

cal_platt = CalibratedClassifierCV(uncalibrated, method='sigmoid', cv='prefit')
cal_platt.fit(X_val, y_val)

cal_isotonic = CalibratedClassifierCV(uncalibrated, method='isotonic', cv='prefit')
cal_isotonic.fit(X_val, y_val)

# Platt Scaling: asume relación sigmoide — funciona con pocos datos de calibración
# Isotonic Regression: más flexible, necesita ~1000 ejemplos mínimo