pipu-music/services/recommender/pipu_rec/train/pipeline.py

"""Ночной пайплайн (docs/recommendations.md#8): события → implicit_score → ALS → проверка метрик →
векторы в track_embeddings.cf / user_profiles_rec.taste_cf и запись в model_versions."""

from __future__ import annotations

import datetime as dt
import json
import logging
import threading
import time
from collections import Counter
from typing import Any

import numpy as np
import psycopg

from pipu_rec.train.als import ALSModel, normalize, train_als
from pipu_rec.train.metrics import evaluate
from pipu_rec.train.signals import DISLIKE, LIKE, Interaction, event_weight, implicit_scores

log = logging.getLogger(__name__)

HOLDOUT_DAYS = 7
# Новая модель активируется, только если NDCG@10 не хуже активной более чем на 2%.
MAX_REGRESSION = 0.02
MIN_INTERACTIONS = 50
ADVISORY_LOCK = 0x70697075  # «pipu»: один прогон обучения на кластер


def load_interactions(conn: psycopg.Connection[Any]) -> list[Interaction]:
    out: list[Interaction] = []
    for user, track, ts, kind, played, duration, in_wave in conn.execute(
        """
        SELECT e.user_id::text, e.track_id::text, extract(epoch FROM e.occurred_at)::float8, e.type,
            e.played_ms, t.duration_ms, (e.wave_session_id IS NOT NULL OR e.context_type = 'wave')
        FROM play_events e JOIN tracks t ON t.id = e.track_id
        WHERE e.type IN ('play_end', 'skip') AND e.occurred_at > now() - interval '365 days'
        """
    ):
        w = event_weight(kind, played, duration, bool(in_wave))
        if w is not None:
            out.append(Interaction(user, track, float(ts), w))
    for user, track, ts, reaction in conn.execute(
        "SELECT user_id::text, track_id::text, extract(epoch FROM created_at)::float8, reaction::text "
        "FROM track_reactions"
    ):
        out.append(Interaction(user, track, float(ts), LIKE if reaction == "like" else DISLIKE))
    return out


def holdout(
    interactions: list[Interaction], now: float, days: int = HOLDOUT_DAYS
) -> tuple[dict[str, dict[str, float]], dict[str, set[str]]]:
    """Разбиение по времени: обучение — до отсечки, тест — треки, впервые понравившиеся после неё."""
    cutoff = now - days * 86400
    train = implicit_scores(interactions, now, before=cutoff)
    after: dict[str, dict[str, float]] = {}
    for it in interactions:
        if it.ts >= cutoff:
            row = after.setdefault(it.user, {})
            row[it.track] = row.get(it.track, 0.0) + it.weight
    relevant = {u: {t for t, w in c.items() if w > 0 and t not in train.get(u, {})} for u, c in after.items()}
    return train, {u: r for u, r in relevant.items() if r}


def offline_metrics(interactions: list[Interaction], now: float) -> dict[str, Any]:
    """ALS против популярности на отложенной неделе."""
    train, relevant = holdout(interactions, now)
    model = train_als(train)
    if model is None or not relevant:
        return {"users": 0}
    pop = Counter(t for row in train.values() for t, s in row.items() if s > 0)
    popular = [t for t, _ in pop.most_common()]

    def rank_als(u: str) -> list[str]:
        seen = train.get(u, {})
        order = np.argsort(-model.scores(u))
        return [model.items[i] for i in order if model.items[i] not in seen][:50]

    def rank_pop(u: str) -> list[str]:
        seen = train.get(u, {})
        return [t for t in popular if t not in seen][:50]

    users = list(relevant)
    return {"als": evaluate(users, relevant, rank_als), "popularity": evaluate(users, relevant, rank_pop)}


def vec_text(v: np.ndarray) -> str:
    return "[" + ",".join(f"{x:.6f}" for x in v.tolist()) + "]"


def active_metrics(conn: psycopg.Connection[Any], kind: str) -> dict[str, Any] | None:
    row = conn.execute(
        "SELECT metrics FROM model_versions WHERE kind = %s AND is_active ORDER BY created_at DESC LIMIT 1",
        (kind,),
    ).fetchone()
    return row[0] if row and row[0] else None


def should_activate(new: dict[str, Any], current: dict[str, Any] | None) -> bool:
    """Без активной модели или без тестовых данных — активируем; иначе NDCG@10 не хуже на > 2%."""
    if current is None:
        return True
    new_ndcg = new.get("als", {}).get("ndcg@10")
    cur_ndcg = (current.get("offline") or {}).get("als", {}).get("ndcg@10")
    if new_ndcg is None or not cur_ndcg:
        return True
    return float(new_ndcg) >= float(cur_ndcg) * (1 - MAX_REGRESSION)


def save_als(conn: psycopg.Connection[Any], model: ALSModel, version: str, metrics: dict[str, Any]) -> None:
    items = normalize(model.item_factors)
    users = normalize(model.user_factors)
    with conn.transaction():
        with conn.cursor() as cur:
            cur.executemany(
                """
                INSERT INTO track_embeddings (track_id, cf, updated_at) VALUES (%s, %s::vector, now())
                ON CONFLICT (track_id) DO UPDATE SET cf = EXCLUDED.cf, updated_at = now()
                """,
                [(t, vec_text(items[i])) for i, t in enumerate(model.items)],
            )
            cur.execute(
                "UPDATE track_embeddings SET cf = NULL "
                "WHERE cf IS NOT NULL AND NOT (track_id::text = ANY(%s))",
                (model.items,),
            )
            cur.executemany(
                """
                INSERT INTO user_profiles_rec (user_id, taste_cf, updated_at) VALUES (%s, %s::vector, now())
                ON CONFLICT (user_id) DO UPDATE SET taste_cf = EXCLUDED.taste_cf, updated_at = now()
                """,
                [(u, vec_text(users[i])) for i, u in enumerate(model.users)],
            )
        conn.execute("UPDATE model_versions SET is_active = false WHERE kind = 'als' AND is_active")
        conn.execute(
            "INSERT INTO model_versions (id, kind, metrics, is_active) VALUES (%s, 'als', %s, true)",
            (version, json.dumps(metrics)),
        )


def train_all(dsn: str, now: float | None = None) -> dict[str, Any]:
    """Один прогон обучения. Возвращает сводку (её же пишет в лог и model_versions)."""
    now = now or time.time()
    with psycopg.connect(dsn) as conn:
        if not conn.execute("SELECT pg_try_advisory_lock(%s)", (ADVISORY_LOCK,)).fetchone()[0]:  # type: ignore[index]
            return {"skipped": "another training is running"}
        try:
            interactions = load_interactions(conn)
            if len(interactions) < MIN_INTERACTIONS:
                return {"skipped": f"only {len(interactions)} interactions"}
            started = time.monotonic()
            offline = offline_metrics(interactions, now)
            model = train_als(implicit_scores(interactions, now))
            if model is None:
                return {"skipped": "no data"}
            version = "als-" + dt.datetime.fromtimestamp(now, dt.UTC).strftime("%Y%m%d%H%M")
            metrics = {
                "offline": offline,
                "users": len(model.users),
                "items": len(model.items),
                "interactions": len(interactions),
                "train_s": round(time.monotonic() - started, 1),
            }
            if not should_activate(offline, active_metrics(conn, "als")):
                conn.execute(
                    "INSERT INTO model_versions (id, kind, metrics, is_active) VALUES (%s, 'als', %s, false) "
                    "ON CONFLICT (id) DO NOTHING",
                    (version, json.dumps(metrics)),
                )
                log.warning("als %s not activated: offline metrics regressed %s", version, offline)
                return {"version": version, "activated": False, **metrics}
            save_als(conn, model, version, metrics)
            log.info("als %s activated: %s", version, metrics)
            return {"version": version, "activated": True, **metrics}
        finally:
            conn.execute("SELECT pg_advisory_unlock(%s)", (ADVISORY_LOCK,))


class NightlyTrainer:
    """Обучение раз в сутки в hour:00 по местному времени сервера; при первом запуске без модели — сразу.
    on_done вызывается после успешной активации (горячая подмена каталога в памяти)."""

    def __init__(self, dsn: str, hour: int = 3, on_done: Any = None) -> None:
        self.dsn = dsn
        self.hour = hour
        self.on_done = on_done
        self._stop = threading.Event()

    def start(self) -> None:
        threading.Thread(target=self._loop, name="nightly-trainer", daemon=True).start()

    def stop(self) -> None:
        self._stop.set()

    def seconds_until_next(self, now: dt.datetime) -> float:
        target = now.replace(hour=self.hour, minute=0, second=0, microsecond=0)
        if target <= now:
            target += dt.timedelta(days=1)
        return (target - now).total_seconds()

    def _run(self) -> None:
        try:
            result = train_all(self.dsn)
            log.info("training finished: %s", result)
            if result.get("activated") and self.on_done:
                self.on_done()
        except Exception:
            log.exception("training failed")

    def _loop(self) -> None:
        try:
            with psycopg.connect(self.dsn) as conn:
                never = active_metrics(conn, "als") is None
        except Exception:
            never = False
        if never:
            self._run()
        while not self._stop.wait(self.seconds_until_next(dt.datetime.now())):
            self._run()