""" FreshMarket Analytics — Clasificador de Intención de Compra

Cliente: FreshMarket Analytics (SaaS B2B para retailers de alimentación) Tarea: Predecir categoría de producto más probable en próxima visita a partir de embeddings de comportamiento del cliente.

Stack: Python 3.11 · PyTorch 2.2 · producción lista para CPU y GPU

Patrones aplicados (CULTIVA IA / patrones-de-pytorch): ✓ Código agnóstico al dispositivo (CPU/GPU sin hardcoding) ✓ Reproducibilidad completa (semillas en torch/numpy/random/cudnn) ✓ Gestión explícita de formas de tensores en forward() ✓ nn.Module limpio con init / forward separados ✓ Inicialización de pesos correcta (Kaiming/He) ✓ Training loop con mixed precision (AMP) + gradient clipping ✓ Validation loop con @torch.no_grad() ✓ DataLoader optimizado (pin_memory, persistent_workers) ✓ Dataset custom con type hints ✓ Checkpointing completo (epoch, optimizer, loss) ✓ torch.compile para inferencia rápida (PyTorch 2.0+) ✓ Anti-patrones documentados y evitados """

from future import annotations

import csv import os import random import time from pathlib import Path from typing import Optional

import numpy as np import torch import torch.nn as nn import torch.nn.functional as F from torch.utils.data import DataLoader, Dataset

---------------------------------------------------------------------------

0. REPRODUCIBILIDAD COMPLETA

---------------------------------------------------------------------------

def set_seed(seed: int = 42) -> None: """Fija todas las fuentes de aleatoriedad para resultados reproducibles.""" torch.manual_seed(seed) torch.cuda.manual_seed_all(seed) np.random.seed(seed) random.seed(seed) torch.backends.cudnn.deterministic = True torch.backends.cudnn.benchmark = False # benchmark=True rompe reproducibilidad

---------------------------------------------------------------------------

1. DISPOSITIVO AGNÓSTICO

---------------------------------------------------------------------------

def get_device() -> torch.device: """Detecta el mejor dispositivo disponible sin hardcoding.""" if torch.cuda.is_available(): return torch.device("cuda") # Apple Silicon MPS — ML engineers con Mac if torch.backends.mps.is_available(): return torch.device("mps") return torch.device("cpu")

---------------------------------------------------------------------------

2. DATASET CUSTOM CON TYPE HINTS

---------------------------------------------------------------------------

CATEGORIES = [ "Frutas", "Verduras", "Lácteos", "Carnes", "Pescado", "Pan", "Bebidas", "Snacks", "Limpieza", "Higiene", "Mascotas", "Congelados", "Conservas", "Cereales", "Dulces", "Alcohol", "Café", "Especias", "Pasta", "Arroz", "Aceites", "Salsas", "Refrigerados", "Otros", ] NUM_CLASSES = len(CATEGORIES) # 24 EMBEDDING_DIM = 128 # dimensión del vector de comportamiento SESSION_DIM = 16 # features de sesión (día, hora, RFM, etc.)

class PurchaseIntentDataset(Dataset): """ Dataset que carga vectores de comportamiento de clientes desde CSV.

CSV esperado: customer_id, emb_0..emb_127, sess_0..sess_15, category_id
Para el ejemplo genera datos sintéticos si no existe el CSV.
"""

def __init__(
    self,
    csv_path: str | Path,
    transform=None,
    n_synthetic: int = 2000,
) -> None:
    self.transform = transform
    csv_path = Path(csv_path)

    if csv_path.exists():
        self._load_from_csv(csv_path)
    else:
        # Datos sintéticos para demostración
        print(f"[Dataset] CSV no encontrado → generando {n_synthetic} muestras sintéticas")
        self._generate_synthetic(n_synthetic)

def _load_from_csv(self, path: Path) -> None:
    embeddings, sessions, labels = [], [], []
    with open(path, newline="") as f:
        reader = csv.DictReader(f)
        for row in reader:
            emb = [float(row[f"emb_{i}"]) for i in range(EMBEDDING_DIM)]
            sess = [float(row[f"sess_{i}"]) for i in range(SESSION_DIM)]
            embeddings.append(emb)
            sessions.append(sess)
            labels.append(int(row["category_id"]))
    self.embeddings = torch.tensor(embeddings, dtype=torch.float32)
    self.sessions   = torch.tensor(sessions,   dtype=torch.float32)
    self.labels     = torch.tensor(labels,     dtype=torch.long)

def _generate_synthetic(self, n: int) -> None:
    # Embeddings con distribución por categoría (centros separados)
    set_seed(99)
    centers = torch.randn(NUM_CLASSES, EMBEDDING_DIM) * 2.0
    labels  = torch.randint(0, NUM_CLASSES, (n,))
    noise   = torch.randn(n, EMBEDDING_DIM) * 0.8
    self.embeddings = centers[labels] + noise
    self.sessions   = torch.randn(n, SESSION_DIM)
    self.labels     = labels

def __len__(self) -> int:
    return len(self.labels)

def __getitem__(self, idx: int) -> tuple[tuple[torch.Tensor, torch.Tensor], int]:
    emb  = self.embeddings[idx]   # (128,)
    sess = self.sessions[idx]     # (16,)
    lbl  = self.labels[idx].item()

    if self.transform:
        emb, sess = self.transform(emb, sess)

    return (emb, sess), lbl

---------------------------------------------------------------------------

3. ARQUITECTURA DEL MODELO — nn.Module LIMPIO

---------------------------------------------------------------------------

class PurchaseIntentClassifier(nn.Module): """ Red neuronal para clasificar intención de compra.

Combina embedding de comportamiento del cliente con features de sesión
a través de una rama dual con fusión final.

Arquitectura:
    customer_embedding (128) ──► BehaviorEncoder ──► (256)
                                                            ├─► FusionMLP ──► (24 clases)
    session_features   (16)  ──► SessionEncoder  ──► (64)  ┘
"""

def __init__(
    self,
    emb_dim: int = EMBEDDING_DIM,
    sess_dim: int = SESSION_DIM,
    num_classes: int = NUM_CLASSES,
    dropout: float = 0.3,
) -> None:
    super().__init__()

    # Rama de comportamiento del cliente
    self.behavior_encoder = nn.Sequential(
        nn.Linear(emb_dim, 256),
        nn.LayerNorm(256),
        nn.ReLU(),
        nn.Dropout(dropout),
        nn.Linear(256, 256),
        nn.LayerNorm(256),
        nn.ReLU(),
    )

    # Rama de features de sesión
    self.session_encoder = nn.Sequential(
        nn.Linear(sess_dim, 64),
        nn.LayerNorm(64),
        nn.ReLU(),
        nn.Dropout(dropout),
    )

    # Capa de fusión
    fusion_dim = 256 + 64  # 320
    self.fusion = nn.Sequential(
        nn.Linear(fusion_dim, 128),
        nn.ReLU(),
        nn.Dropout(dropout * 0.5),
        nn.Linear(128, num_classes),
    )

    # Inicialización de pesos
    self.apply(self._init_weights)

def _init_weights(self, module: nn.Module) -> None:
    """Inicialización Kaiming para capas lineales, ones/zeros para LayerNorm."""
    if isinstance(module, nn.Linear):
        nn.init.kaiming_normal_(module.weight, mode="fan_out", nonlinearity="relu")
        if module.bias is not None:
            nn.init.zeros_(module.bias)
    elif isinstance(module, nn.LayerNorm):
        nn.init.ones_(module.weight)
        nn.init.zeros_(module.bias)

def forward(
    self,
    customer_emb: torch.Tensor,
    session_feats: torch.Tensor,
) -> torch.Tensor:
    # customer_emb:  (batch_size, 128)
    # session_feats: (batch_size, 16)

    beh = self.behavior_encoder(customer_emb)   # → (batch_size, 256)
    sess = self.session_encoder(session_feats)   # → (batch_size, 64)

    fused = torch.cat([beh, sess], dim=1)        # → (batch_size, 320)
    logits = self.fusion(fused)                  # → (batch_size, 24)

    return logits   # raw logits — la pérdida CrossEntropy aplica softmax internamente

---------------------------------------------------------------------------

4. TRAINING LOOP CON MIXED PRECISION + GRADIENT CLIPPING

---------------------------------------------------------------------------

def train_one_epoch( model: nn.Module, dataloader: DataLoader, optimizer: torch.optim.Optimizer, criterion: nn.Module, device: torch.device, scaler: torch.amp.GradScaler | None = None, epoch: int = 0, ) -> float: """Un epoch de entrenamiento con AMP y gradient clipping.""" model.train() # Activa dropout y batch norm en modo entrenamiento total_loss = 0.0 n_batches = len(dataloader)

for batch_idx, ((emb, sess), target) in enumerate(dataloader):
    # Mover datos al dispositivo (el modelo ya está en device)
    emb    = emb.to(device)     # (B, 128)
    sess   = sess.to(device)    # (B, 16)
    target = target.to(device)  # (B,)

    # zero_grad con set_to_none=True es más eficiente que .zero_grad()
    optimizer.zero_grad(set_to_none=True)

    # Mixed precision training (2× velocidad en CUDA)
    use_amp = scaler is not None and device.type == "cuda"
    with torch.amp.autocast("cuda", enabled=use_amp):
        logits = model(emb, sess)          # (B, 24)
        loss = criterion(logits, target)   # escalar

    if scaler is not None and use_amp:
        scaler.scale(loss).backward()
        scaler.unscale_(optimizer)
        # Gradient clipping previene explosión de gradientes
        torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)
        scaler.step(optimizer)
        scaler.update()
    else:
        loss.backward()
        torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)
        optimizer.step()

    total_loss += loss.item()  # .item() DESPUÉS de backward — no antes

    if batch_idx % 10 == 0:
        pct = 100.0 * batch_idx / n_batches
        print(f"  Epoch {epoch:02d} [{pct:5.1f}%] loss={loss.item():.4f}")

return total_loss / n_batches

---------------------------------------------------------------------------

5. VALIDATION LOOP CON @torch.no_grad()

---------------------------------------------------------------------------

@torch.no_grad() # Más eficiente que envolver en bloque with def evaluate( model: nn.Module, dataloader: DataLoader, criterion: nn.Module, device: torch.device, ) -> tuple[float, float, float]: """ Evalúa el modelo. Devuelve (loss, top1_accuracy, top3_accuracy). SIEMPRE llama model.eval() — desactiva dropout y usa running stats de BN. """ model.eval() # CRÍTICO: sin esto, dropout sigue activo en validación total_loss = 0.0 correct_top1 = 0 correct_top3 = 0 total = 0

for (emb, sess), target in dataloader:
    emb    = emb.to(device)
    sess   = sess.to(device)
    target = target.to(device)

    logits = model(emb, sess)                          # (B, 24)
    loss = criterion(logits, target)
    total_loss += loss.item()

    # Top-1
    pred_top1 = logits.argmax(dim=1)                   # (B,)
    correct_top1 += (pred_top1 == target).sum().item()

    # Top-3
    _, pred_top3 = logits.topk(3, dim=1)               # (B, 3)
    correct_top3 += (pred_top3 == target.unsqueeze(1)).any(dim=1).sum().item()

    total += target.size(0)

avg_loss = total_loss / len(dataloader)
acc_top1 = correct_top1 / total
acc_top3 = correct_top3 / total

return avg_loss, acc_top1, acc_top3

---------------------------------------------------------------------------

6. CHECKPOINTING COMPLETO

---------------------------------------------------------------------------

def save_checkpoint( model: nn.Module, optimizer: torch.optim.Optimizer, epoch: int, val_loss: float, acc_top1: float, path: str | Path, ) -> None: """Guarda el estado completo para reanudar entrenamiento.""" torch.save( { "epoch": epoch, "model_state_dict": model.state_dict(), # solo pesos, no la clase "optimizer_state_dict": optimizer.state_dict(), "val_loss": val_loss, "acc_top1": acc_top1, }, path, ) print(f" [Checkpoint] Guardado en {path} (epoch={epoch}, acc_top1={acc_top1:.3f})")

def load_checkpoint( path: str | Path, model: nn.Module, optimizer: Optional[torch.optim.Optimizer] = None, ) -> dict: """Carga checkpoint. map_location='cpu' funciona en cualquier dispositivo.""" checkpoint = torch.load(path, map_location="cpu", weights_only=True) model.load_state_dict(checkpoint["model_state_dict"]) if optimizer is not None: optimizer.load_state_dict(checkpoint["optimizer_state_dict"]) print(f" [Checkpoint] Cargado desde epoch {checkpoint['epoch']}") return checkpoint

---------------------------------------------------------------------------

7. INFERENCIA RÁPIDA CON torch.compile

---------------------------------------------------------------------------

def compile_for_inference(model: nn.Module) -> nn.Module: """ Compila el modelo con torch.compile para inferencia rápida. Solo disponible en PyTorch 2.0+. Si falla, devuelve el modelo sin compilar. """ try: compiled = torch.compile(model, mode="reduce-overhead") print(" [compile] Modelo compilado con mode='reduce-overhead'") return compiled except Exception as e: print(f" [compile] No disponible ({e}), usando modelo sin compilar") return model

---------------------------------------------------------------------------

8. PIPELINE COMPLETO — PUNTO DE ENTRADA

---------------------------------------------------------------------------

def main() -> None: print("=" * 60) print("FreshMarket Analytics — Pipeline de Entrenamiento") print("Patrón: patrones-de-pytorch (CULTIVA IA / MLOps)") print("=" * 60)

# 0. Semilla y dispositivo
set_seed(42)
device = get_device()
print(f"\n[Config] Dispositivo: {device}")
print(f"[Config] Clases: {NUM_CLASSES} categorías de producto")

# 1. Datasets y DataLoaders optimizados
data_path = Path(__file__).parent / "input" / "customer_behavior.csv"
train_dataset = PurchaseIntentDataset(data_path, n_synthetic=1600)
val_dataset   = PurchaseIntentDataset(data_path, n_synthetic=400)

# DataLoader optimizado para producción
train_loader = DataLoader(
    train_dataset,
    batch_size=64,
    shuffle=True,
    num_workers=0,            # 0 para compatibilidad multiplataforma en demo
    pin_memory=(device.type == "cuda"),   # Solo útil en CUDA
    persistent_workers=False,
    drop_last=True,           # Tamaños de batch consistentes para LayerNorm
)
val_loader = DataLoader(
    val_dataset,
    batch_size=128,
    shuffle=False,            # No mezclar en validación
    num_workers=0,
    pin_memory=(device.type == "cuda"),
)
print(f"[Data] Train: {len(train_dataset)} muestras | Val: {len(val_dataset)} muestras")

# 2. Modelo, optimizer, loss
model = PurchaseIntentClassifier(dropout=0.3).to(device)  # .to(device) UNA sola vez

n_params = sum(p.numel() for p in model.parameters() if p.requires_grad)
print(f"\n[Modelo] PurchaseIntentClassifier — {n_params:,} parámetros entrenables")
print(model)

optimizer = torch.optim.AdamW(model.parameters(), lr=1e-3, weight_decay=1e-4)
scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=10, eta_min=1e-5)
criterion = nn.CrossEntropyLoss(label_smoothing=0.1)

# Mixed precision scaler (solo CUDA)
scaler = torch.amp.GradScaler("cuda") if device.type == "cuda" else None

# 3. Training loop
best_val_loss = float("inf")
checkpoint_dir = Path(__file__).parent / "checkpoints"
checkpoint_dir.mkdir(exist_ok=True)

print("\n[Entrenamiento] Iniciando (5 epochs de demo)...")
for epoch in range(1, 6):
    t0 = time.time()

    train_loss = train_one_epoch(
        model, train_loader, optimizer, criterion, device, scaler, epoch=epoch
    )
    val_loss, acc_top1, acc_top3 = evaluate(model, val_loader, criterion, device)
    scheduler.step()

    elapsed = time.time() - t0
    print(
        f"\nEpoch {epoch:02d} | "
        f"train_loss={train_loss:.4f} | "
        f"val_loss={val_loss:.4f} | "
        f"top1={acc_top1:.3f} | "
        f"top3={acc_top3:.3f} | "
        f"lr={scheduler.get_last_lr()[0]:.2e} | "
        f"{elapsed:.1f}s"
    )

    # Guardar mejor checkpoint
    if val_loss < best_val_loss:
        best_val_loss = val_loss
        save_checkpoint(
            model, optimizer, epoch, val_loss, acc_top1,
            checkpoint_dir / "best_model.pt",
        )

# 4. Cargar mejor modelo y compilar para inferencia
print("\n[Inferencia] Cargando mejor checkpoint...")
best_model = PurchaseIntentClassifier().to(device)
load_checkpoint(checkpoint_dir / "best_model.pt", best_model)
best_model = compile_for_inference(best_model)

# 5. Inferencia de ejemplo — batch de 4 clientes
print("\n[Inferencia] Predicción para 4 clientes de ejemplo:")
best_model.eval()
with torch.no_grad():
    sample_emb  = torch.randn(4, EMBEDDING_DIM).to(device)   # (4, 128)
    sample_sess = torch.randn(4, SESSION_DIM).to(device)     # (4, 16)

    t_start = time.perf_counter()
    logits = best_model(sample_emb, sample_sess)             # (4, 24)
    t_end = time.perf_counter()

    probs = F.softmax(logits, dim=1)                          # (4, 24)
    top3_probs, top3_idx = probs.topk(3, dim=1)              # (4, 3)

    for i in range(4):
        preds = [
            f"{CATEGORIES[top3_idx[i, j]]}: {top3_probs[i, j]:.1%}"
            for j in range(3)
        ]
        print(f"  Cliente {i+1}: {' | '.join(preds)}")

    ms = (t_end - t_start) * 1000
    print(f"\n  Latencia batch=4: {ms:.2f}ms ({ms/4:.2f}ms por cliente)")

print("\n[OK] Pipeline completado.")
print("     Patrón aplicado: reproducible · agnóstico · tipado · óptimo")

if name == "main": main()