V6.5-7ds cleanup: remove legacy src/bigru_t/data/data_augmentation.py

src/bigru_t/data/data_augmentation.py

@@ -1,161 +0,0 @@
-"""data_augmentation.py — Data Augmentation para BiGRU_T_version V6.
-
-Adaptado de gru-ring-v13-9-2/source/xavante_work/flexnet/data_augmentation.py.
-
-Estratégias:
-  1. Synonym replacement (placeholder — usa mapeamento estático simples)
-  2. Random deletion (remove tokens com prob p)
-  3. Random swap (troca 2 tokens de posição)
-  4. Token dropout (substitui por PAD com prob p)
-  5. Back-translation simulada (aplica ruído gaussiano nos embeddings)
-  6. Mixup (combina 2 samples: x_mix = λ·x_1 + (1-λ)·x_2)
-
-V6 melhoria vs V13.9.2:
-  * Vetorização total (sem loops Python sobre tokens)
-  * Mixup com λ ~ Beta(α, α) — diferenciável
-  * Compatível com tokens int (não altera vocab) para estratégias 1-4
-  * Compatível com embeddings (float) para estratégias 5-6
-"""
-from __future__ import annotations
-
-import torch
-import torch.nn.functional as F
-from typing import Optional
-
-
-def random_deletion_tokens(
-    input_ids: torch.Tensor,
-    pad_id: int = 0,
-    p: float = 0.1,
-) -> torch.Tensor:
-    """Deleta tokens aleatoriamente (substitui por PAD).
-
-    Args:
-        input_ids: (batch, T) — tokens int
-        pad_id: ID do token PAD
-        p: probabilidade de deletar cada token
-
-    Returns:
-        (batch, T) — tokens com alguns substituídos por pad_id
-    """
-    if p <= 0:
-        return input_ids
-    mask = torch.bernoulli(torch.full_like(input_ids, p, dtype=torch.float)).bool()
-    return input_ids.masked_fill(mask, pad_id)
-
-
-def random_swap_tokens(
-    input_ids: torch.Tensor,
-    n_swaps: int = 1,
-) -> torch.Tensor:
-    """Troca n pares de tokens de posição (por batch).
-
-    Args:
-        input_ids: (batch, T) — tokens int
-        n_swaps: nº de trocas por sample
-    """
-    if n_swaps <= 0:
-        return input_ids
-    batch, T = input_ids.shape
-    out = input_ids.clone()
-    for _ in range(n_swaps):
-        i = torch.randint(0, T, (batch,), device=input_ids.device)
-        j = torch.randint(0, T, (batch,), device=input_ids.device)
-        # Vectorized swap
-        rows = torch.arange(batch, device=input_ids.device)
-        tmp = out[rows, i].clone()
-        out[rows, i] = out[rows, j]
-        out[rows, j] = tmp
-    return out
-
-
-def token_dropout(
-    input_ids: torch.Tensor,
-    pad_id: int = 0,
-    p: float = 0.1,
-) -> torch.Tensor:
-    """Token dropout — alias para random_deletion_tokens (mesma semântica)."""
-    return random_deletion_tokens(input_ids, pad_id=pad_id, p=p)
-
-
-def embedding_noise(
-    embeddings: torch.Tensor,
-    noise_std: float = 0.01,
-) -> torch.Tensor:
-    """Adiciona ruído gaussiano aos embeddings (back-translation simulada).
-
-    Args:
-        embeddings: (batch, T, d)
-        noise_std: desvio padrão do ruído
-    """
-    if noise_std <= 0:
-        return embeddings
-    noise = torch.randn_like(embeddings) * noise_std
-    return embeddings + noise
-
-
-def mixup(
-    x: torch.Tensor,
-    y: torch.Tensor,
-    alpha: float = 0.2,
-    num_classes: Optional[int] = None,
-) -> tuple[torch.Tensor, torch.Tensor, torch.Tensor, float]:
-    """Mixup: x_mix = λ·x_1 + (1-λ)·x_2, y_mix = λ·y_1 + (1-λ)·y_2.
-
-    Args:
-        x: (batch, ...) — features (tokens ou embeddings)
-        y: (batch,) ou (batch, C) — labels (int ou one-hot)
-        alpha: parâmetro da Beta(α, α) para amostrar λ
-        num_classes: nº de classes (necessário se y for int)
-
-    Returns:
-        x_mixed, y_mixed, perm, lam
-    """
-    if alpha <= 0:
-        if y.dim() == 1 and num_classes is not None:
-            y_oh = F.one_hot(y, num_classes=num_classes).float()
-            return x, y_oh, torch.arange(x.size(0), device=x.device), 1.0
-        return x, y.float() if y.dim() == 1 else y, torch.arange(x.size(0), device=x.device), 1.0
-
-    # λ ~ Beta(α, α)
-    lam = float(torch.distributions.Beta(alpha, alpha).sample().item())
-    batch = x.size(0)
-    perm = torch.randperm(batch, device=x.device)
-    x_mixed = lam * x + (1.0 - lam) * x[perm]
-
-    # y_mixed
-    if y.dim() == 1 and num_classes is not None:
-        y_oh = F.one_hot(y, num_classes=num_classes).float()
-        y_perm_oh = F.one_hot(y[perm], num_classes=num_classes).float()
-        y_mixed = lam * y_oh + (1.0 - lam) * y_perm_oh
-    else:
-        y_mixed = lam * y.float() + (1.0 - lam) * y[perm].float()
-
-    return x_mixed, y_mixed, perm, lam
-
-
-def augment_batch(
-    input_ids: torch.Tensor,
-    pad_id: int = 0,
-    p_delete: float = 0.1,
-    n_swaps: int = 1,
-    p_dropout: float = 0.05,
-) -> torch.Tensor:
-    """Pipeline de augmentação para tokens (sem mudar vocab).
-
-    Aplica: random_deletion → random_swap → token_dropout (sequencialmente).
-    """
-    out = random_deletion_tokens(input_ids, pad_id=pad_id, p=p_delete)
-    out = random_swap_tokens(out, n_swaps=n_swaps)
-    out = token_dropout(out, pad_id=pad_id, p=p_dropout)
-    return out
-
-
-__all__ = [
-    "random_deletion_tokens",
-    "random_swap_tokens",
-    "token_dropout",
-    "embedding_noise",
-    "mixup",
-    "augment_batch",
-]