mrleo1nidandClaude Opus 5.5
58bd14ef38
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>
198 lines
8.1 KiB
Python
198 lines
8.1 KiB
Python
"""Ранжировщик волны 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
|