diff --git a/docs/recommendations.md b/docs/recommendations.md index b189c37..3441345 100644 --- a/docs/recommendations.md +++ b/docs/recommendations.md @@ -210,8 +210,12 @@ score = 0.35·cos(e, session_vec) + 0.25·max_c cos(e, centroid_c) + 0.15·artis Ночной пайплайн (03:00, APScheduler в `recommender`): 1. Пересчёт `user_track_stats.implicit_score` (затухание), artist / genre affinity. -2. **ALS** (`implicit`, 64 фактора, confidence `c = 1 + 20·max(score, 0)`, отрицательные — через исключение - и штраф в ранжировании) → `track_embeddings.cf`, `user_profiles_rec.taste_cf`. +2. **ALS** (свой на numpy — `pipu_rec/train/als.py`, 64 фактора, confidence `c = 1 + 20·score` для положительных; + отрицательные — предпочтение 0 с уверенностью `1 + 10·|score|`) → `track_embeddings.cf`, `user_profiles_rec.taste_cf`. + Проверка: последние 7 дней — тест (впервые понравившиеся треки), NDCG@10 / Recall@50 против популярности; + новая модель активируется, если NDCG@10 не хуже активной более чем на 2%. В волне ALS даёт источники `cf` + и `item_cf` и слагаемое `0.10·cf_score`; версия модели в показах — `heuristic-v1+cf`. + Запуск: `TRAIN_HOUR` (по умолчанию 3, `-1` — выключено), вручную — `pipu-rec train`. 3. Кластеризация вкусов (центроиды) для каждого пользователя. 4. Обучение LightGBM-ранжировщика на показах за 90 дней (time-split: последние 7 дней — валидация). 5. Проверка метрик, запись `model_versions`, горячая подмена моделей в памяти сервиса. diff --git a/services/recommender/pipu_rec/cli.py b/services/recommender/pipu_rec/cli.py index 9e5a2b6..fefb47f 100644 --- a/services/recommender/pipu_rec/cli.py +++ b/services/recommender/pipu_rec/cli.py @@ -10,6 +10,7 @@ import uvicorn from pipu_rec.analysis.worker import Worker from pipu_rec.api.app import create_app from pipu_rec.settings import Settings +from pipu_rec.train.pipeline import NightlyTrainer, train_all from pipu_rec.wave.contexts import load_contexts from pipu_rec.wave.service import WaveService @@ -27,6 +28,9 @@ def _serve(settings: Settings) -> None: await rdb.ping() 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() 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) @@ -47,12 +51,15 @@ def main() -> None: sub.add_parser("serve", help="внутренний API рекомендаций") analyze = sub.add_parser("analyze", help="воркер аудиоанализа") analyze.add_argument("--workers", type=int, default=4) + sub.add_parser("train", help="обучить модели сейчас (то же, что ночной прогон)") args = parser.parse_args() settings = Settings() logging.basicConfig(level=settings.log_level.upper()) if args.cmd == "serve": _serve(settings) + elif args.cmd == "train": + log.info("training: %s", train_all(settings.database_url)) else: _analyze(settings, args.workers) diff --git a/services/recommender/pipu_rec/settings.py b/services/recommender/pipu_rec/settings.py index 3139ffe..8e45fc4 100644 --- a/services/recommender/pipu_rec/settings.py +++ b/services/recommender/pipu_rec/settings.py @@ -17,4 +17,5 @@ class Settings(BaseSettings): library_host_root: str = "/library-host" ffmpeg_path: str = "ffmpeg" contexts_path: str = Field(default="", description="contexts.yaml; пусто — встроенный") + train_hour: int = Field(default=3, ge=-1, le=23, description="час ночного обучения; -1 — выключено") version: str = Field(default="0.0.0-dev", alias="PIPU_VERSION") diff --git a/services/recommender/pipu_rec/train/__init__.py b/services/recommender/pipu_rec/train/__init__.py new file mode 100644 index 0000000..caa394f --- /dev/null +++ b/services/recommender/pipu_rec/train/__init__.py @@ -0,0 +1 @@ +"""Ночное обучение: неявные оценки, ALS, офлайн-метрики (docs/recommendations.md#8).""" diff --git a/services/recommender/pipu_rec/train/als.py b/services/recommender/pipu_rec/train/als.py new file mode 100644 index 0000000..70b1dbc --- /dev/null +++ b/services/recommender/pipu_rec/train/als.py @@ -0,0 +1,111 @@ +"""ALS для неявной обратной связи (Hu, Koren, Volinsky, 2008) на numpy. + +На масштабе проекта (~100 пользователей × десятки тысяч треков) решение каждой строки — система 64×64: +весь проход — секунды, без компилируемых зависимостей. Положительная оценка — предпочтение 1 с уверенностью +c = 1 + α·score; отрицательная — предпочтение 0 с повышенной уверенностью (модель учит «не это»). +""" + +from __future__ import annotations + +from dataclasses import dataclass + +import numpy as np +from numpy.typing import NDArray + +F32 = NDArray[np.float32] + + +@dataclass +class ALSModel: + users: list[str] + items: list[str] + user_factors: F32 # (n_users, k) + item_factors: F32 # (n_items, k) + + def user_vec(self, user: str) -> F32 | None: + try: + i = self.users.index(user) + except ValueError: + return None + vec: F32 = self.user_factors[i] + return vec + + def scores(self, user: str) -> NDArray[np.float64]: + u = self.user_vec(user) + if u is None: + return np.zeros(len(self.items)) + return (self.item_factors @ u).astype(np.float64) + + +def _solve_side( + fixed: F32, + rows: list[tuple[NDArray[np.intp], NDArray[np.float64], NDArray[np.float64]]], + reg: float, +) -> F32: + """Пересчёт одной стороны: для каждой строки (индексы, уверенность, предпочтение) — закрытое решение + (YᵀY + Yᵀ(C−I)Y + λI) x = YᵀCp, где YᵀY общий для всех строк.""" + k = fixed.shape[1] + yty = fixed.T @ fixed + out = np.zeros((len(rows), k), dtype=np.float32) + eye = reg * np.eye(k) + for r, (idx, conf, pref) in enumerate(rows): + if len(idx) == 0: + continue + y = fixed[idx].astype(np.float64) + a = yty + (y.T * (conf - 1.0)) @ y + eye + b = (y.T * conf) @ pref + out[r] = np.linalg.solve(a, b) + return out + + +def train_als( + scores: dict[str, dict[str, float]], + factors: int = 64, + reg: float = 0.1, + alpha: float = 20.0, + iterations: int = 12, + seed: int = 0, +) -> ALSModel | None: + """Обучает ALS по implicit_score. None — данных нет.""" + users = sorted(u for u, row in scores.items() if row) + items = sorted({t for row in scores.values() for t in row}) + if not users or not items: + return None + upos = {u: i for i, u in enumerate(users)} + ipos = {t: i for i, t in enumerate(items)} + by_user: list[list[tuple[int, float]]] = [[] for _ in users] + by_item: list[list[tuple[int, float]]] = [[] for _ in items] + for u, row in scores.items(): + if u not in upos: + continue + for t, s in row.items(): + by_user[upos[u]].append((ipos[t], s)) + by_item[ipos[t]].append((upos[u], s)) + + def prep( + lists: list[list[tuple[int, float]]], + ) -> list[tuple[NDArray[np.intp], NDArray[np.float64], NDArray[np.float64]]]: + out = [] + for lst in lists: + idx = np.array([i for i, _ in lst], dtype=np.intp) + s = np.array([v for _, v in lst], dtype=np.float64) + conf = 1.0 + alpha * np.abs(s) * np.where(s > 0, 1.0, 0.5) + pref = (s > 0).astype(np.float64) + out.append((idx, conf, pref)) + return out + + u_rows, i_rows = prep(by_user), prep(by_item) + k = min(factors, max(2, len(items) - 1)) + rng = np.random.default_rng(seed) + x = (rng.standard_normal((len(users), k)) * 0.01).astype(np.float32) + y = (rng.standard_normal((len(items), k)) * 0.01).astype(np.float32) + for _ in range(iterations): + x = _solve_side(y, u_rows, reg) + y = _solve_side(x, i_rows, reg) + return ALSModel(users=users, items=items, user_factors=x, item_factors=y) + + +def normalize(v: F32) -> F32: + """L2-нормировка строк (косинусное сходство в волне).""" + n = np.linalg.norm(v, axis=-1, keepdims=True) + return (v / np.where(n > 0, n, 1.0)).astype(np.float32) diff --git a/services/recommender/pipu_rec/train/metrics.py b/services/recommender/pipu_rec/train/metrics.py new file mode 100644 index 0000000..a303e27 --- /dev/null +++ b/services/recommender/pipu_rec/train/metrics.py @@ -0,0 +1,39 @@ +"""Офлайн-метрики на отложенной по времени выборке: NDCG@10, Recall@50 (docs/recommendations.md#1).""" + +from __future__ import annotations + +import math +from collections.abc import Callable, Iterable + + +def ndcg_at_k(ranked: list[str], relevant: set[str], k: int = 10) -> float: + dcg = sum(1.0 / math.log2(i + 2) for i, t in enumerate(ranked[:k]) if t in relevant) + ideal = sum(1.0 / math.log2(i + 2) for i in range(min(k, len(relevant)))) + return dcg / ideal if ideal > 0 else 0.0 + + +def recall_at_k(ranked: list[str], relevant: set[str], k: int = 50) -> float: + if not relevant: + return 0.0 + return len(set(ranked[:k]) & relevant) / len(relevant) + + +def evaluate( + users: Iterable[str], + relevant: dict[str, set[str]], + rank: Callable[[str], list[str]], +) -> dict[str, float]: + """Средние метрики по пользователям, у которых в тесте есть хотя бы один релевантный трек.""" + n = 0 + ndcg = rec = 0.0 + for u in users: + rel = relevant.get(u) + if not rel: + continue + ranked = rank(u) + ndcg += ndcg_at_k(ranked, rel, 10) + rec += recall_at_k(ranked, rel, 50) + n += 1 + if n == 0: + return {"users": 0, "ndcg@10": 0.0, "recall@50": 0.0} + return {"users": n, "ndcg@10": round(ndcg / n, 4), "recall@50": round(rec / n, 4)} diff --git a/services/recommender/pipu_rec/train/pipeline.py b/services/recommender/pipu_rec/train/pipeline.py new file mode 100644 index 0000000..e52bc4e --- /dev/null +++ b/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() diff --git a/services/recommender/pipu_rec/train/signals.py b/services/recommender/pipu_rec/train/signals.py new file mode 100644 index 0000000..d7e5008 --- /dev/null +++ b/services/recommender/pipu_rec/train/signals.py @@ -0,0 +1,64 @@ +"""Неявная оценка пары (пользователь, трек) из событий — таблица docs/recommendations.md#3.""" + +from __future__ import annotations + +import math +from collections import defaultdict +from dataclasses import dataclass + +HALF_LIFE_DAYS = 90.0 +DAY_S = 86400.0 + +LIKE = 3.0 +DISLIKE = -5.0 + + +@dataclass(frozen=True) +class Interaction: + user: str + track: str + ts: float # unix-время события + weight: float # вклад до затухания + + +def event_weight(kind: str, played_ms: int | None, duration_ms: int | None, in_wave: bool) -> float | None: + """Вклад одного прослушивания. Скип вне волны слабее (×0.5): человек мог просто листать альбом.""" + played = played_ms or 0 + if kind == "play_end": + return 1.0 + if kind != "skip": + return None + if duration_ms and played >= 0.5 * duration_ms: + return 0.7 # засчитанное прослушивание без доигрывания + if played < 10_000: + w = -1.0 + elif played < 30_000: + w = -0.7 + else: + w = -0.2 + return w if in_wave else w * 0.5 + + +def decay(ts: float, now: float) -> float: + """Экспоненциальное затухание с полупериодом 90 дней.""" + return math.exp(-math.log(2) * max(now - ts, 0.0) / DAY_S / HALF_LIFE_DAYS) + + +def implicit_scores( + interactions: list[Interaction], now: float, before: float | None = None +) -> dict[str, dict[str, float]]: + """user → track → implicit_score; before — учитывать только события раньше этого момента. + Дизлайк перекрывает всё: оценка трека становится DISLIKE (исключение из рекомендаций).""" + out: dict[str, dict[str, float]] = defaultdict(dict) + disliked: set[tuple[str, str]] = set() + for it in interactions: + if before is not None and it.ts >= before: + continue + if it.weight <= DISLIKE: + disliked.add((it.user, it.track)) + continue + row = out[it.user] + row[it.track] = row.get(it.track, 0.0) + it.weight * decay(it.ts, now) + for user, track in disliked: + out[user][track] = DISLIKE + return dict(out) diff --git a/services/recommender/pipu_rec/wave/engine.py b/services/recommender/pipu_rec/wave/engine.py index 56c325b..d1c9130 100644 --- a/services/recommender/pipu_rec/wave/engine.py +++ b/services/recommender/pipu_rec/wave/engine.py @@ -10,6 +10,8 @@ from numpy.typing import NDArray from pipu_rec.wave.contexts import FEATURES, Rule, fit MODEL_VER = "heuristic-v1" +# С коллаборативным сигналом (ALS) — суффикс к версии модели в показах. +CF_SUFFIX = "+cf" Vec = NDArray[np.float32] @@ -24,10 +26,15 @@ class CatalogIndex: has_emb: NDArray[np.bool_] features: dict[str, NDArray[np.float64]] # NaN — нет признака popularity: NDArray[np.float64] # 0..1 + cf: Vec = field(default_factory=lambda: np.zeros((0, 0), dtype=np.float32)) # (n, k) ALS, нормирован + has_cf: NDArray[np.bool_] = field(default_factory=lambda: np.zeros(0, dtype=bool)) pos: dict[str, int] = field(default_factory=dict) def __post_init__(self) -> None: self.pos = {t: i for i, t in enumerate(self.ids)} + if self.has_cf.shape[0] != len(self.ids): + self.cf = np.zeros((len(self.ids), 1), dtype=np.float32) + self.has_cf = np.zeros(len(self.ids), dtype=bool) def __len__(self) -> int: return len(self.ids) @@ -40,6 +47,7 @@ class CatalogIndex: features: dict[str, list[float | None]], plays: list[int], dim: int = 256, + raw_cf: list[Vec | None] | None = None, ) -> CatalogIndex: n = len(ids) emb = np.zeros((n, dim), dtype=np.float32) @@ -59,7 +67,17 @@ class CatalogIndex: } p = np.log1p(np.array(plays, dtype=np.float64)) pop = p / p.max() if n and p.max() > 0 else np.zeros(n) - return CatalogIndex(ids=ids, artist=artist, emb=emb, has_emb=has, features=feats, popularity=pop) + cf_dim = next((len(v) for v in raw_cf or [] if v is not None), 0) + cf = np.zeros((n, max(cf_dim, 1)), dtype=np.float32) + has_cf = np.zeros(n, dtype=bool) + for i, v in enumerate(raw_cf or []): + if v is not None and len(v) == cf_dim: + norm = float(np.linalg.norm(v)) + if norm > 0: + cf[i], has_cf[i] = v / norm, True + return CatalogIndex( + ids=ids, artist=artist, emb=emb, has_emb=has, features=feats, popularity=pop, cf=cf, has_cf=has_cf + ) @dataclass @@ -74,6 +92,8 @@ class Profile: liked_recent: set[str] # лайкнутые, игравшие за 24 ч — штраф rediscover: set[str] # любимые, не игравшие > 60 дней exploration: float = 0.15 + taste_cf: list[float] | None = None # фактор пользователя ALS + recent_likes: list[str] = field(default_factory=list) # последние лайки, новые первыми (для item_cf) @dataclass @@ -145,6 +165,21 @@ def update_session( session.vec = vec.tolist() if np.any(vec) else None +def cf_scores(index: CatalogIndex, profile: Profile) -> NDArray[np.float64] | None: + """Косинус треков к фактору пользователя ALS; None — нет модели или пользователь в неё не попал.""" + if profile.taste_cf is None or not index.has_cf.any(): + return None + u = np.array(profile.taste_cf, dtype=np.float32) + if u.shape[0] != index.cf.shape[1]: + return None + u = u / (float(np.linalg.norm(u)) or 1.0) + return np.where(index.has_cf, index.cf @ u, 0.0).astype(np.float64) + + +def model_version(index: CatalogIndex, profile: Profile) -> str: + return MODEL_VER + (CF_SUFFIX if cf_scores(index, profile) is not None else "") + + def _top(scores: NDArray[np.float64], mask: NDArray[np.bool_], k: int) -> NDArray[np.intp]: idx = np.nonzero(mask)[0] if len(idx) <= k: @@ -189,6 +224,7 @@ def build_batch( sim_session = index.emb @ svec if svec is not None else np.zeros(size) sim_taste = np.where(index.has_emb, sim_taste, 0.0) sim_session = np.where(index.has_emb, sim_session, 0.0) + cf_score = cf_scores(index, profile) # Привязанность к исполнителю — сглаженное среднее оценок его треков. totals: dict[str, list[float]] = {} @@ -217,6 +253,15 @@ def build_batch( never = np.array([t not in profile.played for t in index.ids]) add(_top(sim_taste + 0.3 * sim_session, allowed & index.has_emb & never, 50), "fresh") add(_top(query, allowed & index.has_emb, 150), "audio_sim") + # Коллективный опыт (ALS): похожие по вкусу других слушателей и «с этим слушают» к последним лайкам. + if cf_score is not None: + add(_top(cf_score, allowed & index.has_cf, 100), "cf") + likes = [ + index.pos[t] for t in profile.recent_likes[:10] if t in index.pos and index.has_cf[index.pos[t]] + ] + if likes: + item_sim = (index.cf @ index.cf[likes].T).max(axis=1).astype(np.float64) + add(_top(np.where(index.has_cf, item_sim, -1.0), allowed & index.has_cf, 60), "item_cf") liked_artists = {a for a, v in affinity.items() if v > 0.3} if liked_artists: by_artist = [int(i) for i in np.nonzero(allowed)[0] if index.artist[i] in liked_artists] @@ -234,6 +279,7 @@ def build_batch( + 0.25 * sim_taste[cand] + 0.15 * artist_aff[cand] + 0.05 * index.popularity[cand] + + (0.10 * cf_score[cand] if cf_score is not None else 0.0) + 0.02 * rng.random(len(cand)) # разбивает ничьи, особенно при холодном старте ) liked_recent = np.array([index.ids[i] in profile.liked_recent for i in cand]) diff --git a/services/recommender/pipu_rec/wave/service.py b/services/recommender/pipu_rec/wave/service.py index 5e353c7..a78b25f 100644 --- a/services/recommender/pipu_rec/wave/service.py +++ b/services/recommender/pipu_rec/wave/service.py @@ -9,7 +9,7 @@ import psycopg from pydantic import BaseModel, Field from pipu_rec.wave.contexts import Contexts, Rule -from pipu_rec.wave.engine import MODEL_VER, Pick, build_batch, taste_centroids, update_session +from pipu_rec.wave.engine import 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 @@ -89,5 +89,6 @@ class WaveService: 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_VER + tracks=[BatchTrack(id=p.id, source=p.source, score=p.score) for p in picks], + model_ver=model_version(index, profile), ) diff --git a/services/recommender/pipu_rec/wave/store.py b/services/recommender/pipu_rec/wave/store.py index b180c5d..a699266 100644 --- a/services/recommender/pipu_rec/wave/store.py +++ b/services/recommender/pipu_rec/wave/store.py @@ -39,7 +39,8 @@ def load_index(conn: psycopg.Connection[Any]) -> CatalogIndex: WHERE ta.track_id = t.id AND ta.role = 'main' ORDER BY ta.position LIMIT 1), al.album_artist_id::text), e.audio::text, f.bpm, f.energy, f.danceability, f.valence, f.acousticness, - COALESCE((SELECT sum(play_count) FROM user_track_stats s WHERE s.track_id = t.id), 0) + COALESCE((SELECT sum(play_count) FROM user_track_stats s WHERE s.track_id = t.id), 0), + e.cf::text FROM tracks t JOIN albums al ON al.id = t.album_id JOIN media_files mf ON mf.id = t.primary_file_id AND mf.missing_since IS NULL @@ -60,6 +61,7 @@ def load_index(conn: psycopg.Connection[Any]) -> CatalogIndex: raw_emb=[parse_vector(r[2]) for r in rows], features=feats, plays=[int(r[8]) for r in rows], + raw_cf=[parse_vector(r[9]) for r in rows], ) @@ -123,11 +125,15 @@ def load_profile(conn: psycopg.Connection[Any], user_id: str) -> Profile: scores[tid] = raw * math.exp(-math.log(2) * (now - ts) / 86400 / HALF_LIFE_DAYS) disliked: set[str] = set() liked: set[str] = set() + recent_likes: list[str] = [] for tid, reaction in conn.execute( - "SELECT track_id::text, reaction::text FROM track_reactions WHERE user_id = %s", (user_id,) + "SELECT track_id::text, reaction::text FROM track_reactions WHERE user_id = %s " + "ORDER BY created_at DESC", + (user_id,), ): if reaction == "like": liked.add(tid) + recent_likes.append(tid) scores[tid] = scores.get(tid, 0.0) + 3.0 else: disliked.add(tid) @@ -149,8 +155,11 @@ def load_profile(conn: psycopg.Connection[Any], user_id: str) -> Profile: (user_id,), ) } - row = conn.execute("SELECT exploration FROM user_profiles_rec WHERE user_id = %s", (user_id,)).fetchone() + row = conn.execute( + "SELECT exploration, taste_cf::text FROM user_profiles_rec WHERE user_id = %s", (user_id,) + ).fetchone() exploration = float(row[0]) if row else 0.15 + taste_cf = parse_vector(row[1]) if row else None if len(played) < 30: exploration = max(exploration, 0.4) # холодный старт: больше исследования на первых треках day = 86400 @@ -163,6 +172,8 @@ def load_profile(conn: psycopg.Connection[Any], user_id: str) -> Profile: liked_recent={t for t in liked if now - last.get(t, 0) < day}, rediscover={t for t, s in scores.items() if s > 1.5 and now - last.get(t, now) > 60 * day}, exploration=exploration, + taste_cf=taste_cf.tolist() if taste_cf is not None else None, + recent_likes=recent_likes, ) diff --git a/services/recommender/tests/test_train.py b/services/recommender/tests/test_train.py new file mode 100644 index 0000000..232fde3 --- /dev/null +++ b/services/recommender/tests/test_train.py @@ -0,0 +1,110 @@ +"""ALS, неявные оценки, офлайн-метрики и правило активации модели.""" + +import numpy as np + +from pipu_rec.train.als import train_als +from pipu_rec.train.metrics import evaluate, ndcg_at_k, recall_at_k +from pipu_rec.train.pipeline import holdout, should_activate +from pipu_rec.train.signals import DISLIKE, Interaction, event_weight, implicit_scores +from pipu_rec.wave.engine import CatalogIndex, Profile, Session, build_batch, model_version + +DAY = 86400.0 + + +def test_event_weights_follow_spec() -> None: + assert event_weight("play_end", 200_000, 200_000, True) == 1.0 + assert event_weight("skip", 120_000, 200_000, True) == 0.7 # ≥ 50% — засчитано + assert event_weight("skip", 5_000, 200_000, True) == -1.0 + assert event_weight("skip", 20_000, 200_000, True) == -0.7 + assert event_weight("skip", 5_000, 200_000, False) == -0.5 # вне волны — слабее + assert event_weight("play_start", 0, 200_000, True) is None + + +def test_implicit_scores_decay_and_dislike() -> None: + now = 1000 * DAY + inter = [ + Interaction("u", "a", now, 1.0), + Interaction("u", "b", now - 90 * DAY, 1.0), # полупериод — половина веса + Interaction("u", "c", now, 3.0), + Interaction("u", "c", now, DISLIKE), # дизлайк перекрывает + ] + s = implicit_scores(inter, now)["u"] + assert s["a"] == 1.0 + assert abs(s["b"] - 0.5) < 1e-9 + assert s["c"] == DISLIKE + + +def two_taste_world() -> dict[str, dict[str, float]]: + """Две группы слушателей с непересекающимися вкусами: rock_* и jazz_*.""" + rng = np.random.default_rng(1) + scores: dict[str, dict[str, float]] = {} + for g in ("rock", "jazz"): + for u in range(15): + liked = rng.choice(30, size=12, replace=False) + scores[f"{g}_user{u}"] = {f"{g}_{i}": 1.0 + float(rng.random()) for i in liked} + return scores + + +def test_als_learns_taste_groups() -> None: + model = train_als(two_taste_world(), factors=8, iterations=10) + assert model is not None + s = model.scores("rock_user0") + rock = np.mean([s[i] for i, t in enumerate(model.items) if t.startswith("rock")]) + jazz = np.mean([s[i] for i, t in enumerate(model.items) if t.startswith("jazz")]) + assert rock > jazz + 0.3 + + +def test_metrics() -> None: + assert ndcg_at_k(["a", "b"], {"a"}) == 1.0 + assert 0 < ndcg_at_k(["x", "a"], {"a"}) < 1 + assert recall_at_k(["a", "x"], {"a", "b"}) == 0.5 + res = evaluate(["u", "v"], {"u": {"a"}}, lambda u: ["a"]) + assert res == {"users": 1, "ndcg@10": 1.0, "recall@50": 1.0} + + +def test_holdout_keeps_only_new_positives() -> None: + now = 100 * DAY + inter = [ + Interaction("u", "old", now - 30 * DAY, 1.0), + Interaction("u", "old", now - DAY, 1.0), # уже был в обучении — не «новый» + Interaction("u", "new", now - DAY, 1.0), + Interaction("u", "skipped", now - DAY, -1.0), + ] + train, relevant = holdout(inter, now) + assert set(train["u"]) == {"old"} + assert relevant == {"u": {"new"}} + + +def test_activation_gate() -> None: + cur = {"offline": {"als": {"ndcg@10": 0.30}}} + assert should_activate({"als": {"ndcg@10": 0.295}}, cur) # −1.7% — допустимо + assert not should_activate({"als": {"ndcg@10": 0.28}}, cur) + assert should_activate({"users": 0}, cur) # не на чем проверить + assert should_activate({"als": {"ndcg@10": 0.1}}, None) + + +def test_engine_uses_cf_source() -> None: + n = 40 + ids = [f"t{i}" for i in range(n)] + cf = [ + np.array([1.0, 0.0], dtype=np.float32) if i < 20 else np.array([0.0, 1.0], dtype=np.float32) + 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, raw_cf=cf + ) + profile = Profile( + scores={}, + disliked=set(), + disliked_artists=set(), + recent=set(), + played=set(), + liked_recent=set(), + rediscover=set(), + exploration=0.0, + taste_cf=[1.0, 0.0], + ) + picks = build_batch(index, profile, Session(), [], n=5, rng=np.random.default_rng(0)) + assert len(picks) == 5 + assert sum(int(p.id[1:]) < 20 for p in picks) >= 4 # почти всё — из «своей» половины по ALS + assert model_version(index, profile).endswith("+cf")