train_gpu.py

"""Standalone GPU training script for RunPod.

Trains sentence-transformer embeddings and a Neural Collaborative Filtering
(NCF) model on exported movie/rating data.  Fully self-contained -- does NOT
import any app modules so it can run on a bare RunPod instance with only
PyTorch and sentence-transformers installed.

Usage:
    python train_gpu.py
    python train_gpu.py --data-dir /data/export --output-dir /data/models
    python train_gpu.py --device cuda:0 --epochs 80 --batch-size 512
"""

from __future__ import annotations

import argparse
import json
import time
from pathlib import Path
from typing import Any

import numpy as np
import torch
import torch.nn as nn
from torch.utils.data import DataLoader, TensorDataset
from sentence_transformers import SentenceTransformer


# ---------------------------------------------------------------------------
# Document builder -- mirrors feature_engineering.build_movie_document
# ---------------------------------------------------------------------------

def build_movie_document(movie: dict[str, Any]) -> str:
    """Build a natural-language document from movie metadata.

    Uses natural language format optimized for sentence transformers,
    matching ``app.ml.embedding_engine.build_movie_document``.
    """
    parts: list[str] = []

    if movie.get("overview"):
        parts.append(movie["overview"])

    if movie.get("genres"):
        genres = movie["genres"] if isinstance(movie["genres"], list) else []
        if genres:
            parts.append(f"Genres: {', '.join(str(g) for g in genres)}.")

    if movie.get("director"):
        parts.append(f"Director: {movie['director']}.")

    if movie.get("cast"):
        cast = movie["cast"] if isinstance(movie["cast"], list) else []
        cast_names: list[str] = []
        for c in cast[:5]:
            if isinstance(c, dict):
                name = c.get("name")
                if name:
                    cast_names.append(str(name))
            elif isinstance(c, str):
                cast_names.append(c)
        if cast_names:
            parts.append(f"Cast: {', '.join(cast_names)}.")

    return " ".join(parts)


# ---------------------------------------------------------------------------
# Embedding training
# ---------------------------------------------------------------------------

def train_embeddings(
    movies: list[dict[str, Any]],
    output_dir: Path,
    device: str,
) -> None:
    """Encode all movies with a sentence-transformer and persist the result.

    Saves:
        - ``embeddings.npy``       -- (N, D) float32 array
        - ``embeddings_metadata.json`` -- movie id list + model info
    """
    print(f"\n{'='*60}")
    print("Training embeddings")
    print(f"{'='*60}")
    t0 = time.time()

    print(f"Loading SentenceTransformer('all-MiniLM-L6-v2') on {device} ...")
    model = SentenceTransformer("all-MiniLM-L6-v2", device=device)

    print(f"Building documents for {len(movies)} movies ...")
    documents = [build_movie_document(m) for m in movies]

    empty_count = sum(1 for d in documents if not d.strip())
    if empty_count:
        print(f"  Warning: {empty_count} movies produced empty documents")

    print("Encoding ...")
    embeddings = model.encode(
        documents,
        batch_size=64,
        show_progress_bar=True,
        convert_to_numpy=True,
    )

    embeddings_path = output_dir / "embeddings.npy"
    np.save(embeddings_path, embeddings)
    print(f"Saved embeddings ({embeddings.shape}) to {embeddings_path}")

    metadata = {
        "model_name": "all-MiniLM-L6-v2",
        "embedding_dim": int(embeddings.shape[1]),
        "n_movies": len(movies),
        "movie_ids": [m["id"] for m in movies],
    }
    metadata_path = output_dir / "embeddings_metadata.json"
    with open(metadata_path, "w") as f:
        json.dump(metadata, f, indent=2)
    print(f"Saved metadata to {metadata_path}")

    elapsed = time.time() - t0
    print(f"Embeddings done in {elapsed:.1f}s")


# ---------------------------------------------------------------------------
# NCF model definition
# ---------------------------------------------------------------------------

class NCFModel(nn.Module):
    """Neural Collaborative Filtering (GMF + MLP).

    Architecture mirrors the project's collaborative model design:
      - GMF path:  element-wise product of user/item embeddings
      - MLP path:  concatenated embeddings -> [128, 64, 32] with ReLU + Dropout
      - Output:    sigmoid(combined) * 9 + 1  (maps to 1-10 rating scale)
    """

    def __init__(
        self,
        n_users: int,
        n_items: int,
        embed_dim: int = 32,
        mlp_layers: tuple[int, ...] = (128, 64, 32),
        dropout: float = 0.2,
    ):
        super().__init__()

        # GMF embeddings
        self.gmf_user_embed = nn.Embedding(n_users, embed_dim)
        self.gmf_item_embed = nn.Embedding(n_items, embed_dim)

        # MLP embeddings
        self.mlp_user_embed = nn.Embedding(n_users, embed_dim)
        self.mlp_item_embed = nn.Embedding(n_items, embed_dim)

        # MLP layers
        mlp_modules: list[nn.Module] = []
        input_dim = embed_dim * 2
        for hidden_dim in mlp_layers:
            mlp_modules.append(nn.Linear(input_dim, hidden_dim))
            mlp_modules.append(nn.ReLU())
            mlp_modules.append(nn.Dropout(dropout))
            input_dim = hidden_dim
        self.mlp = nn.Sequential(*mlp_modules)

        # Final prediction layer (GMF output + last MLP layer output)
        self.output_layer = nn.Linear(embed_dim + mlp_layers[-1], 1)
        self.sigmoid = nn.Sigmoid()

        self._init_weights()

    def _init_weights(self) -> None:
        for module in self.modules():
            if isinstance(module, nn.Embedding):
                nn.init.normal_(module.weight, std=0.01)
            elif isinstance(module, nn.Linear):
                nn.init.xavier_uniform_(module.weight)
                if module.bias is not None:
                    nn.init.zeros_(module.bias)

    def forward(self, user_ids: torch.Tensor, item_ids: torch.Tensor) -> torch.Tensor:
        # GMF path
        gmf_user = self.gmf_user_embed(user_ids)
        gmf_item = self.gmf_item_embed(item_ids)
        gmf_out = gmf_user * gmf_item  # element-wise product

        # MLP path
        mlp_user = self.mlp_user_embed(user_ids)
        mlp_item = self.mlp_item_embed(item_ids)
        mlp_input = torch.cat([mlp_user, mlp_item], dim=-1)
        mlp_out = self.mlp(mlp_input)

        # Combine and predict
        combined = torch.cat([gmf_out, mlp_out], dim=-1)
        logit = self.output_layer(combined)
        return self.sigmoid(logit).squeeze(-1) * 9.0 + 1.0


# ---------------------------------------------------------------------------
# NCF training
# ---------------------------------------------------------------------------

MIN_RATINGS_FOR_NCF = 50


def train_ncf(
    ratings: list[dict[str, Any]],
    output_dir: Path,
    device: str,
    epochs: int,
    batch_size: int,
) -> None:
    """Train a Neural Collaborative Filtering model and save to disk.

    Saves:
        - ``ncf_model.pt``         -- model state_dict
        - ``ncf_metadata.json``    -- id mappings + hyper-parameters
    """
    print(f"\n{'='*60}")
    print("Training NCF model")
    print(f"{'='*60}")

    if len(ratings) < MIN_RATINGS_FOR_NCF:
        print(
            f"Skipping NCF: only {len(ratings)} ratings "
            f"(minimum {MIN_RATINGS_FOR_NCF} required)"
        )
        return

    t0 = time.time()

    # ── Build ID mappings ────────────────────────────────────────────────
    user_ids_set = sorted({r["user_id"] for r in ratings})
    movie_ids_set = sorted({r["movie_id"] for r in ratings})

    user_to_idx = {uid: idx for idx, uid in enumerate(user_ids_set)}
    movie_to_idx = {mid: idx for idx, mid in enumerate(movie_ids_set)}

    n_users = len(user_ids_set)
    n_items = len(movie_ids_set)
    print(f"Users: {n_users}  |  Items: {n_items}  |  Ratings: {len(ratings)}")

    # ── Prepare tensors ──────────────────────────────────────────────────
    user_indices = torch.tensor(
        [user_to_idx[r["user_id"]] for r in ratings], dtype=torch.long
    )
    item_indices = torch.tensor(
        [movie_to_idx[r["movie_id"]] for r in ratings], dtype=torch.long
    )
    rating_values = torch.tensor(
        [r["rating"] for r in ratings], dtype=torch.float32
    )

    # ── Train / validation split (90/10) ─────────────────────────────────
    n_total = len(ratings)
    perm = torch.randperm(n_total)
    split = int(n_total * 0.9)

    train_idx, val_idx = perm[:split], perm[split:]

    train_ds = TensorDataset(
        user_indices[train_idx], item_indices[train_idx], rating_values[train_idx]
    )
    val_ds = TensorDataset(
        user_indices[val_idx], item_indices[val_idx], rating_values[val_idx]
    )

    train_loader = DataLoader(train_ds, batch_size=batch_size, shuffle=True)
    val_loader = DataLoader(val_ds, batch_size=batch_size)

    print(f"Train: {len(train_ds)}  |  Val: {len(val_ds)}")

    # ── Instantiate model ────────────────────────────────────────────────
    model = NCFModel(n_users=n_users, n_items=n_items).to(device)
    criterion = nn.MSELoss()
    optimizer = torch.optim.Adam(model.parameters(), lr=0.001)

    total_params = sum(p.numel() for p in model.parameters())
    print(f"Model parameters: {total_params:,}")

    # ── Training loop with early stopping ────────────────────────────────
    best_val_loss = float("inf")
    patience = 5
    patience_counter = 0
    best_state = None

    for epoch in range(1, epochs + 1):
        # Train
        model.train()
        train_loss_sum = 0.0
        train_count = 0
        for u_batch, i_batch, r_batch in train_loader:
            u_batch = u_batch.to(device)
            i_batch = i_batch.to(device)
            r_batch = r_batch.to(device)

            optimizer.zero_grad()
            predictions = model(u_batch, i_batch)
            loss = criterion(predictions, r_batch)
            loss.backward()
            optimizer.step()

            train_loss_sum += loss.item() * len(r_batch)
            train_count += len(r_batch)

        train_loss = train_loss_sum / train_count

        # Validate
        model.eval()
        val_loss_sum = 0.0
        val_count = 0
        with torch.no_grad():
            for u_batch, i_batch, r_batch in val_loader:
                u_batch = u_batch.to(device)
                i_batch = i_batch.to(device)
                r_batch = r_batch.to(device)

                predictions = model(u_batch, i_batch)
                loss = criterion(predictions, r_batch)
                val_loss_sum += loss.item() * len(r_batch)
                val_count += len(r_batch)

        val_loss = val_loss_sum / max(val_count, 1)

        # Log every epoch (useful on GPU runs where you tail the log)
        print(
            f"  Epoch {epoch:3d}/{epochs}  "
            f"train_loss={train_loss:.4f}  val_loss={val_loss:.4f}"
        )

        # Early stopping check
        if val_loss < best_val_loss:
            best_val_loss = val_loss
            patience_counter = 0
            best_state = {k: v.cpu().clone() for k, v in model.state_dict().items()}
        else:
            patience_counter += 1
            if patience_counter >= patience:
                print(f"  Early stopping at epoch {epoch} (patience={patience})")
                break

    # ── Save model ───────────────────────────────────────────────────────
    if best_state is None:
        print("Warning: no checkpoint saved (training may have failed)")
        return

    model_path = output_dir / "ncf_model.pt"
    torch.save(best_state, model_path)
    print(f"Saved best model (val_loss={best_val_loss:.4f}) to {model_path}")

    metadata = {
        "n_users": n_users,
        "n_items": n_items,
        "embed_dim": 32,
        "mlp_layers": [128, 64, 32],
        "dropout": 0.2,
        "best_val_loss": float(best_val_loss),
        "user_id_to_idx": {str(k): v for k, v in user_to_idx.items()},
        "movie_id_to_idx": {str(k): v for k, v in movie_to_idx.items()},
    }
    metadata_path = output_dir / "ncf_metadata.json"
    with open(metadata_path, "w") as f:
        json.dump(metadata, f, indent=2)
    print(f"Saved metadata to {metadata_path}")

    elapsed = time.time() - t0
    print(f"NCF training done in {elapsed:.1f}s")


# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------

def main() -> None:
    parser = argparse.ArgumentParser(
        description="Train movie-recommendation models on GPU (RunPod)."
    )
    parser.add_argument(
        "--data-dir",
        type=Path,
        default=Path(__file__).resolve().parent.parent / "data" / "export",
        help="Directory containing movies.json and ratings.json",
    )
    parser.add_argument(
        "--output-dir",
        type=Path,
        default=Path(__file__).resolve().parent.parent / "data" / "models",
        help="Directory to write trained model artefacts",
    )
    parser.add_argument(
        "--device",
        type=str,
        default="cuda:0",
        help="Torch device (default: cuda:0)",
    )
    parser.add_argument(
        "--epochs",
        type=int,
        default=50,
        help="Max training epochs for NCF (default: 50)",
    )
    parser.add_argument(
        "--batch-size",
        type=int,
        default=256,
        help="Batch size for NCF training (default: 256)",
    )
    args = parser.parse_args()

    # ── Resolve paths ────────────────────────────────────────────────────
    data_dir: Path = args.data_dir
    output_dir: Path = args.output_dir
    output_dir.mkdir(parents=True, exist_ok=True)

    # ── Validate device ──────────────────────────────────────────────────
    device = args.device
    if device.startswith("cuda") and not torch.cuda.is_available():
        print(f"Warning: {device} requested but CUDA is not available, falling back to cpu")
        device = "cpu"
    print(f"Using device: {device}")

    # ── Load data ────────────────────────────────────────────────────────
    movies_path = data_dir / "movies.json"
    ratings_path = data_dir / "ratings.json"

    if not movies_path.exists():
        print(f"Error: {movies_path} not found. Run export_data.py first.")
        raise SystemExit(1)

    with open(movies_path) as f:
        movies: list[dict[str, Any]] = json.load(f)

    ratings: list[dict[str, Any]] = []
    if ratings_path.exists():
        with open(ratings_path) as f:
            ratings = json.load(f)
    else:
        print(f"Warning: {ratings_path} not found, skipping NCF training")

    print(f"\nLoaded {len(movies)} movies, {len(ratings)} ratings from {data_dir}")

    # ── Train ────────────────────────────────────────────────────────────
    overall_start = time.time()

    train_embeddings(movies, output_dir, device)
    train_ncf(ratings, output_dir, device, args.epochs, args.batch_size)

    total = time.time() - overall_start
    print(f"\nAll done in {total:.1f}s")


if __name__ == "__main__":
    main()