"""Ранжировщик волны 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