diff --git a/docs/recommendations.md b/docs/recommendations.md index f27de36..823021a 100644 --- a/docs/recommendations.md +++ b/docs/recommendations.md @@ -152,6 +152,14 @@ score = 0.35·cos(e, session_vec) + 0.25·max_c cos(e, centroid_c) + 0.15·artis Обучение — ночью; новая модель активируется, только если офлайн-NDCG@10 не хуже активной более чем на 2%. Пока размеченных показов меньше ~20k, используется v1 (переключение автоматическое по порогу). +Реализация (`pipu_rec/train/ranker.py`): при каждом показе волна пишет вектор признаков кандидата +в `rec_impression_features` (29 признаков, `FEATURE_NAMES` в `wave/engine.py`; порядок версионируется +`FEATURES_VERSION`). Ночью признаки соединяются с исходами `wave_impressions`, группа — сессия волны; +валидация — сессии последних 7 дней. Модель включается, если её NDCG@10 не ниже порядка по оценке v1 +и не хуже активной более чем на 2%. Модель хранится строкой в `model_versions.artifact` (kind `ranker`), +сервис загружает активную при старте и после ночного обучения; в выдаче её порядок заменяет оценку v1, +MMR-разнообразие и квоты остаются. Пока модели нет, `model_ver` показа — версия v1. + ### 6.4 Занятия и настроения Каждый контекст — профиль целевых значений аудиопризнаков (`services/recommender/config/contexts.yaml`, diff --git a/services/api/migrations/00010_ranker.sql b/services/api/migrations/00010_ranker.sql new file mode 100644 index 0000000..5ec9f62 --- /dev/null +++ b/services/api/migrations/00010_ranker.sql @@ -0,0 +1,17 @@ +-- +goose Up +-- Ранжировщик волны v2 (docs/recommendations.md#6.3): признаки кандидата в момент показа — для обучения +-- LightGBM на исходах показов (wave_impressions.outcome); артефакт модели хранится в model_versions. +CREATE TABLE rec_impression_features ( + wave_session_id uuid NOT NULL REFERENCES wave_sessions(id) ON DELETE CASCADE, + track_id uuid NOT NULL, + features real[] NOT NULL, + served_at timestamptz NOT NULL DEFAULT now(), + PRIMARY KEY (wave_session_id, track_id) +); +CREATE INDEX rec_impression_features_served_idx ON rec_impression_features (served_at); + +ALTER TABLE model_versions ADD COLUMN artifact text; + +-- +goose Down +ALTER TABLE model_versions DROP COLUMN artifact; +DROP TABLE rec_impression_features; diff --git a/services/recommender/Dockerfile b/services/recommender/Dockerfile index b57dbb7..8a6ba85 100644 --- a/services/recommender/Dockerfile +++ b/services/recommender/Dockerfile @@ -15,8 +15,8 @@ COPY services/recommender/ ./ RUN --mount=type=cache,target=/root/.cache/uv uv sync --frozen --no-dev FROM python:3.12-slim -# ffmpeg декодирует аудио для анализатора. -RUN apt-get update && apt-get install -y --no-install-recommends ffmpeg && rm -rf /var/lib/apt/lists/* && useradd --system --uid 10001 pipu +# ffmpeg декодирует аудио для анализатора; libgomp1 нужен LightGBM (ранжировщик волны). +RUN apt-get update && apt-get install -y --no-install-recommends ffmpeg libgomp1 && rm -rf /var/lib/apt/lists/* && useradd --system --uid 10001 pipu COPY --from=build /app /app ARG VERSION=0.0.0-dev ENV PATH=/app/.venv/bin:$PATH PIPU_VERSION=${VERSION} PYTHONUNBUFFERED=1 diff --git a/services/recommender/pipu_rec/cli.py b/services/recommender/pipu_rec/cli.py index fefb47f..07849db 100644 --- a/services/recommender/pipu_rec/cli.py +++ b/services/recommender/pipu_rec/cli.py @@ -30,7 +30,7 @@ def _serve(settings: Settings) -> None: wave = WaveService(settings.database_url, settings.redis_url, load_contexts(settings.contexts_path)) if settings.train_hour >= 0: # После активации новой модели каталог перечитывается сразу — горячая подмена векторов. - NightlyTrainer(settings.database_url, settings.train_hour, on_done=wave.index.refresh).start() + NightlyTrainer(settings.database_url, settings.train_hour, on_done=wave.reload).start() app = create_app(settings, {"postgres": check_postgres, "redis": check_redis}, wave) uvicorn.run(app, host=settings.http_host, port=settings.http_port, log_level=settings.log_level) diff --git a/services/recommender/pipu_rec/train/pipeline.py b/services/recommender/pipu_rec/train/pipeline.py index 38e76b0..e3f9d0a 100644 --- a/services/recommender/pipu_rec/train/pipeline.py +++ b/services/recommender/pipu_rec/train/pipeline.py @@ -17,6 +17,7 @@ import psycopg from pipu_rec.train.als import ALSModel, normalize, train_als from pipu_rec.train.bandit import update_bandits from pipu_rec.train.metrics import evaluate +from pipu_rec.train.ranker import train_and_save from pipu_rec.train.signals import DISLIKE, LIKE, Interaction, event_weight, implicit_scores log = logging.getLogger(__name__) @@ -160,9 +161,19 @@ def train_all(dsn: str, now: float | None = None) -> dict[str, Any]: try: # Бандит считается всегда — ему хватает и пары сессий волны, в отличие от ALS. bandit_users = update_bandits(conn, now) + # Ранжировщик v2 учится на показах волны и от числа событий ALS не зависит. + try: + ranker = train_and_save(conn, now) + except Exception as e: # обучение ранжировщика не должно мешать ALS + log.exception("ranker training failed") + ranker = {"ranker": {"error": str(e)}} interactions = load_interactions(conn) if len(interactions) < MIN_INTERACTIONS: - return {"skipped": f"only {len(interactions)} interactions", "bandit_users": bandit_users} + return { + "skipped": f"only {len(interactions)} interactions", + "bandit_users": bandit_users, + **ranker, + } started = time.monotonic() offline = offline_metrics(interactions, now) model = train_als(implicit_scores(interactions, now)) @@ -184,10 +195,10 @@ def train_all(dsn: str, now: float | None = None) -> dict[str, Any]: (version, json.dumps(metrics)), ) log.warning("als %s not activated: offline metrics regressed %s", version, offline) - return {"version": version, "activated": False, **metrics} + return {"version": version, "activated": False, **metrics, **ranker} save_als(conn, model, version, metrics) log.info("als %s activated: %s", version, metrics) - return {"version": version, "activated": True, **metrics} + return {"version": version, "activated": True, **metrics, **ranker} finally: conn.execute("SELECT pg_advisory_unlock(%s)", (ADVISORY_LOCK,)) @@ -218,7 +229,8 @@ class NightlyTrainer: try: result = train_all(self.dsn) log.info("training finished: %s", result) - if result.get("activated") and self.on_done: + ranker_on = bool((result.get("ranker") or {}).get("activated")) + if (result.get("activated") or ranker_on) and self.on_done: self.on_done() except Exception: log.exception("training failed") diff --git a/services/recommender/pipu_rec/train/ranker.py b/services/recommender/pipu_rec/train/ranker.py new file mode 100644 index 0000000..d958e8f --- /dev/null +++ b/services/recommender/pipu_rec/train/ranker.py @@ -0,0 +1,197 @@ +"""Ранжировщик волны v2 — LightGBM lambdarank на показах всех пользователей (docs/recommendations.md#6.3). + +Признаки кандидата пишутся в момент показа (rec_impression_features), метка — исход показа: +3 — лайк, 2 — доиграл, 0 — скип; ранний скип и дизлайк — 0 с весом ×2. Группа — сессия волны. +Новая модель включается, только если её NDCG@10 на отложенной неделе не хуже порядка по оценке v1 и не хуже +активной модели более чем на 2%. Пока размеченных показов меньше MIN_ROWS, работает v1. +""" + +from __future__ import annotations + +import datetime as dt +import json +import logging +import math +import threading +from collections import defaultdict +from typing import Any + +import lightgbm as lgb +import numpy as np +import psycopg +from numpy.typing import NDArray + +from pipu_rec.wave.engine import FEATURE_NAMES, FEATURES_VERSION + +log = logging.getLogger(__name__) + +MIN_ROWS = 20_000 +HOLDOUT_DAYS = 7 +MAX_REGRESSION = 0.02 +LABEL = {"like": 3, "complete": 2, "skip": 0, "skip_early": 0, "dislike": 0} +WEIGHT = {"skip_early": 2.0, "dislike": 2.0} +V1_COLUMN = FEATURE_NAMES.index("v1_score") + + +Row = tuple[str, list[float], str, float] # сессия, признаки, исход, время показа + + +def load_rows(conn: psycopg.Connection[Any]) -> list[Row]: + rows = conn.execute( + """ + SELECT f.wave_session_id::text, f.features, wi.outcome, extract(epoch FROM f.served_at)::float8 + FROM rec_impression_features f + JOIN wave_impressions wi ON wi.wave_session_id = f.wave_session_id AND wi.track_id = f.track_id + WHERE wi.outcome IS NOT NULL AND f.served_at > now() - interval '90 days' + ORDER BY f.wave_session_id, f.served_at + """ + ).fetchall() + return [(s, list(x), o, float(ts)) for s, x, o, ts in rows if o in LABEL and len(x) == len(FEATURE_NAMES)] + + +def _groups(rows: list[Row]) -> list[list[Row]]: + by: dict[str, list[Row]] = defaultdict(list) + for r in rows: + by[r[0]].append(r) + return [g for g in by.values() if len(g) > 1] + + +def _ndcg(scores: NDArray[np.float64], labels: NDArray[np.float64], k: int = 10) -> float | None: + if labels.max() <= 0: + return None # в группе нечего ранжировать — в среднем не участвует + order = np.argsort(-scores)[:k] + dcg = sum((2 ** labels[i] - 1) / math.log2(r + 2) for r, i in enumerate(order)) + ideal = sorted(labels, reverse=True)[:k] + idcg = sum((2**v - 1) / math.log2(r + 2) for r, v in enumerate(ideal)) + return float(dcg / idcg) if idcg > 0 else None + + +def mean_ndcg(groups: list[list[Row]], score: Any) -> float: + vals = [] + for g in groups: + x = np.array([r[1] for r in g], dtype=np.float64) + y = np.array([LABEL[r[2]] for r in g], dtype=np.float64) + v = _ndcg(np.asarray(score(x), dtype=np.float64), y) + if v is not None: + vals.append(v) + return round(float(np.mean(vals)), 4) if vals else 0.0 + + +def train(rows: list[Row], now: float, min_rows: int = MIN_ROWS) -> tuple[lgb.Booster | None, dict[str, Any]]: + """Обучает модель; None — данных мало (метрики объясняют почему).""" + metrics: dict[str, Any] = {"rows": len(rows), "features_version": FEATURES_VERSION} + if len(rows) < min_rows: + metrics["skipped"] = f"need {min_rows} labelled impressions" + return None, metrics + cutoff = now - HOLDOUT_DAYS * 86400 + groups = _groups(rows) + train_g = [g for g in groups if max(r[3] for r in g) < cutoff] + valid_g = [g for g in groups if max(r[3] for r in g) >= cutoff] + if not train_g or not valid_g: + metrics["skipped"] = "no time split" + return None, metrics + + def matrix( + gs: list[list[Row]], + ) -> tuple[NDArray[np.float64], NDArray[np.int64], NDArray[np.float64], list[int]]: + x = np.array([r[1] for g in gs for r in g], dtype=np.float64) + y = np.array([LABEL[r[2]] for g in gs for r in g], dtype=np.int64) + w = np.array([WEIGHT.get(r[2], 1.0) for g in gs for r in g], dtype=np.float64) + return x, y, w, [len(g) for g in gs] + + x, y, w, group = matrix(train_g) + params = { + "objective": "lambdarank", + "learning_rate": 0.05, + "num_leaves": 31, + "min_data_in_leaf": 20, + "verbose": -1, + } + booster = lgb.train(params, lgb.Dataset(x, y, weight=w, group=group), num_boost_round=200) + metrics.update( + { + "train_groups": len(train_g), + "valid_groups": len(valid_g), + "ndcg@10": mean_ndcg(valid_g, booster.predict), + "ndcg@10_v1": mean_ndcg(valid_g, lambda m: m[:, V1_COLUMN]), + } + ) + return booster, metrics + + +def should_activate(metrics: dict[str, Any], current: dict[str, Any] | None) -> bool: + new = float(metrics.get("ndcg@10", 0.0)) + if new < float(metrics.get("ndcg@10_v1", 0.0)): + return False + if current and current.get("ndcg@10"): + return new >= float(current["ndcg@10"]) * (1 - MAX_REGRESSION) + return True + + +def train_and_save(conn: psycopg.Connection[Any], now: float) -> dict[str, Any]: + """Шаг ночного пайплайна: обучение, проверка, запись в model_versions.""" + booster, metrics = train(load_rows(conn), now) + if booster is None: + return {"ranker": metrics} + version = "lgbm-" + dt.datetime.fromtimestamp(now, dt.UTC).strftime("%Y%m%d%H%M") + row = conn.execute( + "SELECT metrics FROM model_versions WHERE kind = 'ranker' AND is_active " + "ORDER BY created_at DESC LIMIT 1" + ).fetchone() + active = should_activate(metrics, row[0] if row else None) + with conn.transaction(): + if active: + conn.execute("UPDATE model_versions SET is_active = false WHERE kind = 'ranker' AND is_active") + conn.execute( + "INSERT INTO model_versions (id, kind, metrics, is_active, artifact) " + "VALUES (%s, 'ranker', %s, %s, %s) " + "ON CONFLICT (id) DO NOTHING", + (version, json.dumps(metrics), active, booster.model_to_string()), + ) + log.info("ranker %s %s: %s", version, "activated" if active else "kept inactive", metrics) + return {"ranker": {"version": version, "activated": active, **metrics}} + + +class LoadedRanker: + """Активная модель в памяти сервиса (Ranker для engine.build_batch).""" + + def __init__(self, version: str, model: str) -> None: + self.version = version + self.booster = lgb.Booster(model_str=model) + + def predict(self, x: NDArray[np.float64]) -> NDArray[np.float64]: + return np.asarray(self.booster.predict(x), dtype=np.float64) + + +class RankerHolder: + """Загружает активный ранжировщик из model_versions; reload — после ночного обучения.""" + + def __init__(self, dsn: str) -> None: + self.dsn = dsn + self._ranker: LoadedRanker | None = None + self._loaded = False + self._lock = threading.Lock() + + def get(self) -> LoadedRanker | None: + if not self._loaded: + self.reload() + return self._ranker + + def reload(self) -> None: + with self._lock: + self._loaded = True + try: + with psycopg.connect(self.dsn) as conn: + row = conn.execute( + "SELECT id, artifact, metrics FROM model_versions " + "WHERE kind = 'ranker' AND is_active " + "AND artifact IS NOT NULL ORDER BY created_at DESC LIMIT 1" + ).fetchone() + if row is None or (row[2] or {}).get("features_version") != FEATURES_VERSION: + self._ranker = None + return + self._ranker = LoadedRanker(row[0], row[1]) + log.info("ranker %s loaded", row[0]) + except Exception: + log.exception("ranker load failed — wave uses v1") + self._ranker = None diff --git a/services/recommender/pipu_rec/wave/engine.py b/services/recommender/pipu_rec/wave/engine.py index 5ab2193..10a3965 100644 --- a/services/recommender/pipu_rec/wave/engine.py +++ b/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 diff --git a/services/recommender/pipu_rec/wave/service.py b/services/recommender/pipu_rec/wave/service.py index a78b25f..987831b 100644 --- a/services/recommender/pipu_rec/wave/service.py +++ b/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) diff --git a/services/recommender/pyproject.toml b/services/recommender/pyproject.toml index 1c2be52..a287593 100644 --- a/services/recommender/pyproject.toml +++ b/services/recommender/pyproject.toml @@ -11,6 +11,7 @@ dependencies = [ "redis>=6.2,<7", "numpy>=2.2,<3", "pyyaml>=6.0,<7", + "lightgbm>=4.5,<5", ] [project.scripts] diff --git a/services/recommender/tests/test_db_integration.py b/services/recommender/tests/test_db_integration.py index fcb074b..ec26b8e 100644 --- a/services/recommender/tests/test_db_integration.py +++ b/services/recommender/tests/test_db_integration.py @@ -159,3 +159,25 @@ def test_onboarding_seeds_taste(world: dict[str, Any]) -> None: profile = load_profile(conn, user) assert len(profile.scores) >= 10, "chosen artist's tracks become the initial taste" assert not profile.played + + +def test_impression_features_roundtrip(world: dict[str, Any]) -> None: + from pipu_rec.train.ranker import load_rows, train_and_save + from pipu_rec.wave.engine import FEATURE_NAMES + + session, user, track = uid(), world["users"][0], world["tracks"][0] + with psycopg.connect(DSN, autocommit=True) as conn: + conn.execute("INSERT INTO wave_sessions (id, user_id) VALUES (%s, %s)", (session, user)) + conn.execute( + "INSERT INTO wave_impressions (wave_session_id, position, track_id, candidate_source, outcome) " + "VALUES (%s, 0, %s, 'audio_sim', 'complete')", + (session, track), + ) + conn.execute( + "INSERT INTO rec_impression_features (wave_session_id, track_id, features) VALUES (%s, %s, %s)", + (session, track, [0.5] * len(FEATURE_NAMES)), + ) + rows = load_rows(conn) + assert any(r[0] == session and r[2] == "complete" for r in rows) + result = train_and_save(conn, time.time()) + assert "skipped" in result["ranker"], "too few impressions — v1 stays" diff --git a/services/recommender/tests/test_train.py b/services/recommender/tests/test_train.py index 819a21d..767b06d 100644 --- a/services/recommender/tests/test_train.py +++ b/services/recommender/tests/test_train.py @@ -152,3 +152,60 @@ def test_bandit_favours_successful_source() -> None: ) picks = build_batch(index, good, Session(), [], n=5, rng=np.random.default_rng(1)) assert sum(p.source == "popular" for p in picks) >= 4 + + +def test_ranker_learns_what_v1_misses() -> None: + from pipu_rec.train.ranker import V1_COLUMN, should_activate, train + from pipu_rec.wave.engine import FEATURE_NAMES + + rng = np.random.default_rng(0) + now = 200 * DAY + rows = [] + col = FEATURE_NAMES.index("sim_session") + for g in range(400): + ts = now - (1 if g >= 320 else 30) * DAY # последние сессии — отложенная неделя + for _ in range(8): + x = rng.random(len(FEATURE_NAMES)).tolist() + # Исход определяется sim_session, а оценка v1 в признаках — случайный шум. + outcome = "like" if x[col] > 0.85 else "complete" if x[col] > 0.5 else "skip_early" + x[V1_COLUMN] = float(rng.random()) + rows.append((f"s{g}", x, outcome, ts)) + booster, m = train(rows, now, min_rows=100) + assert booster is not None, m + assert m["ndcg@10"] > m["ndcg@10_v1"] + 0.1, m + assert should_activate(m, None) + assert not should_activate({"ndcg@10": 0.5, "ndcg@10_v1": 0.6}, None) + assert not should_activate({"ndcg@10": 0.7, "ndcg@10_v1": 0.6}, {"ndcg@10": 0.8}) + _, few = train(rows[:50], now) + assert "skipped" in few + + +def test_engine_uses_ranker_and_logs_features() -> None: + from pipu_rec.wave.engine import FEATURE_NAMES + + n = 40 + ids = [f"t{i}" for i in range(n)] + index = CatalogIndex.build( + ids=ids, artist=[f"a{i}" for i in range(n)], raw_emb=[None] * n, features={}, plays=[0] * n + ) + profile = Profile( + scores={}, + disliked=set(), + disliked_artists=set(), + recent=set(), + played=set(), + liked_recent=set(), + rediscover=set(), + exploration=0.0, + ) + + class LastFirst: + version = "fake-ranker" + + def predict(self, x: np.ndarray) -> np.ndarray: + return -x[:, FEATURE_NAMES.index("popularity")] + np.arange(len(x)) * 1e-3 + + picks = build_batch(index, profile, Session(), [], n=5, rng=np.random.default_rng(0), ranker=LastFirst()) + assert len(picks) == 5 + assert all(p.features is not None and len(p.features) == len(FEATURE_NAMES) for p in picks) + assert model_version(index, profile, LastFirst()) == "fake-ranker" diff --git a/services/recommender/uv.lock b/services/recommender/uv.lock index a250848..dae221d 100644 --- a/services/recommender/uv.lock +++ b/services/recommender/uv.lock @@ -176,6 +176,24 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/ff/3d/9668a400c8dd81d162eba38b33fa49fa6205f1a64493570a13fbb815c3ee/librt-0.16.0-cp312-cp312-win_arm64.whl", hash = "sha256:f7be7cf555bc30ec12622e9447299cc4a9b8ff307548b634794353db0c2065dc", size = 116572, upload-time = "2026-09-29T00:53:04.815Z" }, ] +[[package]] +name = "lightgbm" +version = "4.7.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "narwhals" }, + { name = "numpy" }, + { name = "scipy" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/63/8e/4db5e29290d7e619c307fdb8dab0a0514090af2ce3ec483050e024ec6126/lightgbm-4.7.0.tar.gz", hash = "sha256:f8e20f682c9aabd000bcf4a7ed8aa6f473c1adfecccae34ec24e823d156f4af0", size = 1792896, upload-time = "2026-07-18T21:00:56.139Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/cd/05/7213965863cba1ed0150ad045bceed6276a1afaaaedbaeff4699ec4f0ccb/lightgbm-4.7.0-py3-none-macosx_10_15_x86_64.whl", hash = "sha256:dfc1cfe8e760387be1e7ba7a214688be21fdff96e4ed9749188f83e1877c2477", size = 1877851, upload-time = "2026-07-18T21:00:35.225Z" }, + { url = "https://files.pythonhosted.org/packages/b2/86/f4fe714f2e0bf3941705a20d7f6849dc476276d71236e82ea6b0d6539b86/lightgbm-4.7.0-py3-none-macosx_12_0_arm64.whl", hash = "sha256:129535462686f274df179133643118c5c5c5667167fe6c3a28d955f0b3c8e868", size = 1498914, upload-time = "2026-07-18T21:00:36.549Z" }, + { url = "https://files.pythonhosted.org/packages/c6/a3/b29580948b92e8c2f84dea70118ac702ff067dc52ec4ffb5d73c953536a5/lightgbm-4.7.0-py3-none-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:d4529acec5c6fefe4768302a529707d0ead90f6a6f42df694b856212e09695b8", size = 3349492, upload-time = "2026-07-18T21:00:37.943Z" }, + { url = "https://files.pythonhosted.org/packages/15/eb/837ea3b40cc36e22eeebb9785c01e42b2c255d033eea1d2d9ee8e2540e55/lightgbm-4.7.0-py3-none-manylinux_2_27_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:d23e922acd891e77212e4d0fbcee9ba973c96dee479491341d05ba595357ebb7", size = 3476028, upload-time = "2026-07-18T21:00:39.331Z" }, + { url = "https://files.pythonhosted.org/packages/d5/0b/c5c17d862b12ce292f24cd85d40f2f8f8981668fbdbd43fdc2625eccbc79/lightgbm-4.7.0-py3-none-win_amd64.whl", hash = "sha256:f42d1e5b32b6f170e606d7c689c6165671da98d7bf37f1addec2623efc8740c9", size = 1360833, upload-time = "2026-07-18T21:00:40.865Z" }, +] + [[package]] name = "mypy" version = "1.20.2" @@ -207,6 +225,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/79/7b/2c79738432f5c924bef5071f933bcc9efd0473bac3b4aa584a6f7c1c8df8/mypy_extensions-1.1.0-py3-none-any.whl", hash = "sha256:1be4cccdb0f2482337c4743e60421de3a356cd97508abadd57d47403e94f5505", size = 4963, upload-time = "2025-04-22T14:54:22.983Z" }, ] +[[package]] +name = "narwhals" +version = "2.27.1" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/bf/21/f64d6b2dbea7bf3f8c38cdc786dcc6ef012ca3d173ad208c782c9a7bedf6/narwhals-2.27.1.tar.gz", hash = "sha256:aed93076a3ea42d9c32c88e4eb5ea422a21937011cbe1f480f9572a523c82094", size = 735013, upload-time = "2026-10-10T06:52:18.113Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/d1/89/5d4c86da1130d9059681e5b6cd7645df5c10279a6a079c5c37dcb2cc6f3f/narwhals-2.27.1-py3-none-any.whl", hash = "sha256:d057df13f5852b8e157596e82eb5e955fad267425df5e420e0ee9863da483b31", size = 483211, upload-time = "2026-10-10T06:52:16.32Z" }, +] + [[package]] name = "numpy" version = "2.5.4" @@ -262,6 +289,7 @@ version = "0.1.0" source = { editable = "." } dependencies = [ { name = "fastapi" }, + { name = "lightgbm" }, { name = "numpy" }, { name = "psycopg", extra = ["binary"] }, { name = "pydantic-settings" }, @@ -282,6 +310,7 @@ dev = [ [package.metadata] requires-dist = [ { name = "fastapi", specifier = ">=0.115,<1" }, + { name = "lightgbm", specifier = ">=4.5,<5" }, { name = "numpy", specifier = ">=2.2,<3" }, { name = "psycopg", extras = ["binary"], specifier = ">=3.2,<4" }, { name = "pydantic-settings", specifier = ">=2.9,<3" }, @@ -490,6 +519,27 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/df/c9/0f2a4dea06a5e12c1191ee3d96b1e1233797d0d5e18f97a2b909a95a9836/ruff-0.17.1-py3-none-win_arm64.whl", hash = "sha256:42400b9a6ff8515ff27796f07f92e0195be011c86e81e381606f58ea5502fef4", size = 9853542, upload-time = "2026-10-15T15:23:16.05Z" }, ] +[[package]] +name = "scipy" +version = "1.18.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "numpy" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/7e/74/66de6258867beb2ef08f35f9f2ac017a52cacd5081714d239ff1a442d458/scipy-1.18.1.tar.gz", hash = "sha256:52c4b7422442aba924d03ad4019852b08a92e64ea187b933135687bfe2747307", size = 30781235, upload-time = "2026-08-21T23:28:50.599Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/18/f7/240c110c08693826b4513a52f5717d62ec7c7af72f2920821247c03b17b3/scipy-1.18.1-cp312-cp312-macosx_10_15_x86_64.whl", hash = "sha256:457fd7a2a8edeb044ab6ffbc0aa03ff6cd18491356e5e0c834d76ce621b916d1", size = 31111061, upload-time = "2026-08-21T23:23:44.522Z" }, + { url = "https://files.pythonhosted.org/packages/05/4a/78c6285577c375e7cf27277ea8ee6961224327f1e1a0c44af5f17f23635c/scipy-1.18.1-cp312-cp312-macosx_12_0_arm64.whl", hash = "sha256:e708533e8b2ae2497d65346538a7dcc92814410b25b81432eac66de0f2af8265", size = 28733332, upload-time = "2026-08-21T23:23:50.015Z" }, + { url = "https://files.pythonhosted.org/packages/a5/f6/a5b82f8abbe14d134691b8b903696f701d25a081353a29dc655c364d9e62/scipy-1.18.1-cp312-cp312-macosx_14_0_arm64.whl", hash = "sha256:7bbf207c4453ce1ad2e00b17313852b33310b83090c2311bdaf97f93c0380d12", size = 20475078, upload-time = "2026-08-21T23:23:54.138Z" }, + { url = "https://files.pythonhosted.org/packages/23/22/0858a0bbd6b3e825ceb8cd9baf9eaf3b2f2b1d77727eb6be40500bcdc92f/scipy-1.18.1-cp312-cp312-macosx_14_0_x86_64.whl", hash = "sha256:78c0665edead396b1abb4897c41a5c1d9bf090c8a637a4c20a61678e0a264e66", size = 23108904, upload-time = "2026-08-21T23:23:57.824Z" }, + { url = "https://files.pythonhosted.org/packages/75/9a/2e71719f31eaefe0e3a1706c4a1ded94e664bfd95ffca2b219a671faee01/scipy-1.18.1-cp312-cp312-manylinux_2_27_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:3c085faa2cfa879c5141df483f836f4d691045a078224a670fa570fa01612d89", size = 34025113, upload-time = "2026-08-21T23:24:02.209Z" }, + { url = "https://files.pythonhosted.org/packages/df/64/ff35eb9e54894cf471ff4716abd3c81eb0a0626869217ce3e6ba4ccf17d7/scipy-1.18.1-cp312-cp312-manylinux_2_27_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:f55fa87b6c612ecd6b058f167c53231b1d14e412efe361d3d6e38b3631c73218", size = 35344199, upload-time = "2026-08-21T23:24:07.844Z" }, + { url = "https://files.pythonhosted.org/packages/d3/af/c5538be1792f7034c12c7db6ee67cace58253c7b87b122d68253eaf5de89/scipy-1.18.1-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:c35d74ce0e193ff740c2f2be2ac913ddc232fe6c1ff40b26cfecb9c670c63314", size = 35639587, upload-time = "2026-08-21T23:24:13.05Z" }, + { url = "https://files.pythonhosted.org/packages/91/4c/075e4f66471bac101141ac739e9e135549be1bae584571bd03a530c056e1/scipy-1.18.1-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:d2924a03db38dc2e848bca2fe9f077dafb891480b91a00a0963a8cf86dfc31c1", size = 37480330, upload-time = "2026-08-21T23:24:19.608Z" }, + { url = "https://files.pythonhosted.org/packages/39/e7/979fd14e75008623df31ba70d6bb144700f68feadcea042021c06a05bf82/scipy-1.18.1-cp312-cp312-win_amd64.whl", hash = "sha256:5e4d44984abc0020154ea81b247adeddcc3ac5527b975ff798bd1ba0adc513c2", size = 36658278, upload-time = "2026-08-21T23:24:25.463Z" }, + { url = "https://files.pythonhosted.org/packages/c7/0b/e1525354ff9d7d5feb6d1b31af6d14072e5c91e9607b421fa1ec889660b3/scipy-1.18.1-cp312-cp312-win_arm64.whl", hash = "sha256:d65d448389b8436493abcf629cc94ad0cf32aecaf06e1acca1de53cc795f2f12", size = 24400588, upload-time = "2026-08-21T23:24:30.579Z" }, +] + [[package]] name = "starlette" version = "1.9.0"