feat(recommender): nightly ALS with offline metrics and CF candidates in the wave (M6)

- Implicit scores from play events and reactions per the signals table
  (90-day half-life, weaker skips outside the wave, dislikes override).
- numpy ALS for implicit feedback; time-split evaluation on the last
  7 days (NDCG@10, Recall@50) against a popularity baseline; a new model
  is activated only if NDCG@10 does not regress by more than 2%.
- Vectors are written to track_embeddings.cf and user_profiles_rec.taste_cf,
  versions to model_versions; NightlyTrainer runs at TRAIN_HOUR and on
  first start, `pipu-rec train` runs it manually.
- Wave: cf and item_cf candidate sources and a 0.10 cf term in the v1
  score; impressions are tagged heuristic-v1+cf.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

services/recommender/pipu_rec/train/pipeline.py

@@ -0,0 +1,220 @@
+"""Ночной пайплайн (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()