feat(recommender): LightGBM lambdarank ranker with impression features (M6)

The wave logs each served candidate's feature vector to rec_impression_features.
Nightly training joins them with impression outcomes, validates on the last week
and activates the model only when it beats the v1 order and the active model.
The service loads the active ranker and uses its order in place of the v1 score.

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

services/recommender/pipu_rec/wave/engine.py

@@ -2,7 +2,10 @@
 
 from __future__ import annotations
 
+import math
+import time
 from dataclasses import dataclass, field
+from typing import Protocol
 
 import numpy as np
 from numpy.typing import NDArray
@@ -114,6 +117,55 @@ class Pick:
     id: str
     source: str
     score: float
+    features: list[float] | None = None  # вектор признаков в момент показа (для обучения ранжировщика)
+
+
+# Источники кандидатов (one-hot в признаках ранжировщика; порядок фиксирован).
+SOURCES = ["seed", "fresh", "audio_sim", "cf", "item_cf", "artist", "rediscover", "popular", "explore"]
+
+# Признаки ранжировщика v2 (docs/recommendations.md#6.3). Порядок — часть формата модели:
+# менять только вместе с FEATURES_VERSION (старые модели тогда не загружаются).
+FEATURE_NAMES = [
+    "sim_session",
+    "sim_taste",
+    "cf",
+    "artist_affinity",
+    "popularity",
+    "v1_score",
+    "context_fit",
+    "is_new",
+    "liked_recent",
+    "bandit_mean",
+    *[f"src_{s}" for s in SOURCES],
+    "energy",
+    "bpm",
+    "danceability",
+    "valence",
+    "hour_sin",
+    "hour_cos",
+    "position",
+    "skips_in_row",
+    "has_activity",
+    "has_mood",
+]
+FEATURES_VERSION = 1
+
+
+class Ranker(Protocol):
+    """Обученный ранжировщик: оценки кандидатов по матрице признаков (строки — кандидаты)."""
+
+    version: str
+
+    def predict(self, x: NDArray[np.float64]) -> NDArray[np.float64]: ...
+
+
+@dataclass
+class Context:
+    """Контекст показа для признаков ранжировщика."""
+
+    hour: float = field(default_factory=lambda: time.localtime().tm_hour + time.localtime().tm_min / 60)
+    has_activity: bool = False
+    has_mood: bool = False
 
 
 # Вес бонуса бандита к оценке (θ − 0.5 ∈ [−0.5, 0.5]).
@@ -182,10 +234,66 @@ def cf_scores(index: CatalogIndex, profile: Profile) -> NDArray[np.float64] | No
     return np.where(index.has_cf, index.cf @ u, 0.0).astype(np.float64)
 
 
-def model_version(index: CatalogIndex, profile: Profile) -> str:
+def model_version(index: CatalogIndex, profile: Profile, ranker: Ranker | None = None) -> str:
+    if ranker is not None:
+        return ranker.version
     return MODEL_VER + (CF_SUFFIX if cf_scores(index, profile) is not None else "")
 
 
+def candidate_features(
+    index: CatalogIndex,
+    profile: Profile,
+    session: Session,
+    cand: NDArray[np.intp],
+    sources: dict[int, str],
+    sim_session: NDArray[np.float64],
+    sim_taste: NDArray[np.float64],
+    cf_score: NDArray[np.float64] | None,
+    artist_aff: NDArray[np.float64],
+    v1: NDArray[np.float64],
+    context_fit: NDArray[np.float64],
+    ctx: Context,
+) -> NDArray[np.float64]:
+    """Матрица признаков (кандидаты × FEATURE_NAMES)."""
+    m = len(cand)
+    x = np.zeros((m, len(FEATURE_NAMES)), dtype=np.float64)
+
+    def feat(name: str) -> NDArray[np.float64]:
+        v = index.features.get(name)
+        if v is None:
+            return np.full(m, -1.0)
+        return np.nan_to_num(v[cand], nan=-1.0)
+
+    col = {n: k for k, n in enumerate(FEATURE_NAMES)}
+    x[:, col["sim_session"]] = sim_session[cand]
+    x[:, col["sim_taste"]] = sim_taste[cand]
+    x[:, col["cf"]] = cf_score[cand] if cf_score is not None else 0.0
+    x[:, col["artist_affinity"]] = artist_aff[cand]
+    x[:, col["popularity"]] = index.popularity[cand]
+    x[:, col["v1_score"]] = v1
+    x[:, col["context_fit"]] = context_fit
+    x[:, col["is_new"]] = [index.ids[i] not in profile.played for i in cand]
+    x[:, col["liked_recent"]] = [index.ids[i] in profile.liked_recent for i in cand]
+    for j, i in enumerate(cand):
+        src = sources[int(i)]
+        ab = profile.bandit.get(src)
+        x[j, col["bandit_mean"]] = ab[0] / (ab[0] + ab[1]) if ab else 0.5
+        if src in SOURCES:
+            x[j, col[f"src_{src}"]] = 1.0
+    x[:, col["energy"]] = feat("energy")
+    bpm = feat("bpm")
+    x[:, col["bpm"]] = np.where(bpm > 0, bpm / 200.0, -1.0)
+    x[:, col["danceability"]] = feat("danceability")
+    x[:, col["valence"]] = feat("valence")
+    x[:, col["hour_sin"]] = math.sin(2 * math.pi * ctx.hour / 24)
+    x[:, col["hour_cos"]] = math.cos(2 * math.pi * ctx.hour / 24)
+    x[:, col["position"]] = min(len(session.served) / 50.0, 1.0)
+    x[:, col["skips_in_row"]] = min(session.skips_in_row / 3.0, 1.0)
+    x[:, col["has_activity"]] = float(ctx.has_activity)
+    x[:, col["has_mood"]] = float(ctx.has_mood)
+    return x
+
+
 def _top(scores: NDArray[np.float64], mask: NDArray[np.bool_], k: int) -> NDArray[np.intp]:
     idx = np.nonzero(mask)[0]
     if len(idx) <= k:
@@ -203,6 +311,8 @@ def build_batch(
     exclude: set[str] | None = None,
     seed_ids: list[str] | None = None,
     rng: np.random.Generator | None = None,
+    ranker: Ranker | None = None,
+    ctx: Context | None = None,
 ) -> list[Pick]:
     """Кандидаты → фильтры → оценка → соответствие контексту → MMR с правилами разнообразия."""
     rng = rng or np.random.default_rng()
@@ -279,7 +389,7 @@ def build_batch(
     add(rng.permutation(np.nonzero(allowed)[0])[:40].tolist(), "explore")
 
     cand = np.array(sorted(sources))
-    # 3. Оценка v1.
+    # 3. Оценка v1 (она же — признак и запасной вариант для v2).
     score = (
         0.35 * sim_session[cand]
         + 0.25 * sim_taste[cand]
@@ -297,8 +407,10 @@ def build_batch(
         ab = profile.bandit.get(src)
         theta[src] = float(rng.beta(ab[0], ab[1])) if ab else 0.5
     score = score + BANDIT_WEIGHT * (np.array([theta[sources[int(i)]] for i in cand]) - 0.5)
+    v1 = score.copy()
 
     # 4. Соответствие контексту; если отсеялось слишком много — порог ослабляется.
+    f = np.ones(len(cand))
     if rules:
         f = fit({k: v[cand] for k, v in index.features.items()}, rules)
         # Нужен запас для разнообразия, но не больше 50 и не больше четверти пула.
@@ -307,11 +419,33 @@ def build_batch(
             keep = f >= threshold
             if keep.sum() >= enough or threshold == 0.0:
                 break
-        cand, score, f = cand[keep], score[keep], f[keep]
+        cand, score, f, v1 = cand[keep], score[keep], f[keep], v1[keep]
         score = score + 0.5 * f  # контекст важнее вкуса, когда выбран явно
     if len(cand) == 0:
         return []
 
+    # Признаки кандидатов: пишутся с показом (обучение v2) и подаются обученному ранжировщику.
+    ctx = ctx or Context()
+    f64 = np.float64
+    x = candidate_features(
+        index,
+        profile,
+        session,
+        cand,
+        sources,
+        np.asarray(sim_session, f64),
+        np.asarray(sim_taste, f64),
+        cf_score,
+        np.asarray(artist_aff, f64),
+        np.asarray(v1, f64),
+        np.asarray(f, f64),
+        ctx,
+    )
+    if ranker is not None:
+        # v2: порядок задаёт модель; в MMR нужна шкала 0..1 — берём ранг предсказания.
+        pred = ranker.predict(x)
+        score = np.argsort(np.argsort(pred)) / max(len(pred) - 1, 1)
+
     # 5. MMR + правила: 1 трек исполнителя на пачку, не больше 2 на последние 10; доля «нового».
     picks: list[Pick] = []
     chosen: list[int] = []
@@ -346,7 +480,14 @@ def build_batch(
         i = int(cand[best_j])
         chosen.append(i)
         order.remove(best_j)
-        picks.append(Pick(id=index.ids[i], source=sources[i], score=round(float(score[best_j]), 4)))
+        picks.append(
+            Pick(
+                id=index.ids[i],
+                source=sources[i],
+                score=round(float(score[best_j]), 4),
+                features=x[best_j].tolist(),
+            )
+        )
 
     # Исследование: «новых» (неслышанных) меньше доли exploration — заменяем ими слабейшие выборы.
     fresh_left = sorted(
@@ -361,5 +502,7 @@ def build_batch(
         weakest = min(others, key=lambda k: picks[k].score)
         if any(index.artist[index.pos[p.id]] == index.artist[i] for k, p in enumerate(picks) if k != weakest):
             continue
-        picks[weakest] = Pick(id=index.ids[i], source="fresh", score=round(float(score[j]), 4))
+        picks[weakest] = Pick(
+            id=index.ids[i], source="fresh", score=round(float(score[j]), 4), features=x[j].tolist()
+        )
     return picks

services/recommender/pipu_rec/wave/service.py

@@ -2,18 +2,22 @@
 
 from __future__ import annotations
 
+import logging
 from typing import Literal
 
 import numpy as np
 import psycopg
 from pydantic import BaseModel, Field
 
+from pipu_rec.train.ranker import RankerHolder
 from pipu_rec.wave.contexts import Contexts, Rule
-from pipu_rec.wave.engine import Pick, build_batch, model_version, taste_centroids, update_session
+from pipu_rec.wave.engine import Context, Pick, build_batch, model_version, taste_centroids, update_session
 from pipu_rec.wave.store import IndexHolder, SessionStore, load_profile, seed_tracks
 
 MAX_SERVED_MEMORY = 300
 
+log = logging.getLogger(__name__)
+
 Outcome = Literal["complete", "skip", "skip_early", "like", "dislike"]
 
 
@@ -55,6 +59,12 @@ class WaveService:
         self.index = IndexHolder(dsn)
         self.sessions = SessionStore(redis_url)
         self.contexts = contexts
+        self.ranker = RankerHolder(dsn)
+
+    def reload(self) -> None:
+        """После ночного обучения: перечитать каталог (векторы ALS) и активный ранжировщик."""
+        self.index.refresh()
+        self.ranker.reload()
 
     def rules(self, activity: str | None, mood: str | None) -> list[Rule]:
         """Занятие и настроение комбинируются — пересечение профилей; неизвестный ключ игнорируется."""
@@ -74,6 +84,7 @@ class WaveService:
         update_session(
             index, session, [(f.track_id, f.outcome) for f in req.feedback], taste_centroids(index, profile)
         )
+        ranker = self.ranker.get()
         picks: list[Pick] = build_batch(
             index,
             profile,
@@ -83,12 +94,31 @@ class WaveService:
             exclude=set(req.exclude_ids),
             seed_ids=seeds,
             rng=np.random.default_rng(),
+            ranker=ranker,
+            ctx=Context(has_activity=bool(req.activity), has_mood=bool(req.mood)),
         )
+        self.log_features(req.session_id, picks)
         session.served = (session.served + [p.id for p in picks])[-MAX_SERVED_MEMORY:]
         artists = [index.artist[index.pos[p.id]] for p in picks]
         session.recent_artists = (session.recent_artists + artists)[-10:]
         self.sessions.put(req.session_id, session)
         return BatchResponse(
             tracks=[BatchTrack(id=p.id, source=p.source, score=p.score) for p in picks],
-            model_ver=model_version(index, profile),
+            model_ver=model_version(index, profile, ranker),
         )
+
+    def log_features(self, session_id: str, picks: list[Pick]) -> None:
+        """Признаки показанных треков — обучающие данные ранжировщика v2. Сбой записи не ломает волну."""
+        rows = [(session_id, p.id, p.features) for p in picks if p.features is not None]
+        if not rows:
+            return
+        try:
+            with psycopg.connect(self.dsn, autocommit=True) as conn, conn.cursor() as cur:
+                cur.executemany(
+                    "INSERT INTO rec_impression_features (wave_session_id, track_id, features) "
+                    "VALUES (%s, %s, %s) "
+                    "ON CONFLICT DO NOTHING",
+                    rows,
+                )
+        except psycopg.Error as e:
+            log.warning("impression features not saved: %s", e)