mrleo1nidandClaude Opus 5.5
2285107d04
- newApp (extracted from main) is the exact production wiring; the `ci / db` job now also runs it on PostgreSQL, Redis and Meilisearch and walks the main user flows over HTTP: first admin sign-in and credential change, local source and scan, search, album / artist / top tracks, signed stream, events and history, likes, fallback wave, playback state, upload queue up to review. - ALS on a small library produced fewer than 64 factors and the insert into vector(64) failed; vectors are now zero-padded to the column size. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
232 lines
10 KiB
Python
232 lines
10 KiB
Python
"""Ночной пайплайн (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)
|
|
|
|
|
|
# Размерность колонок track_embeddings.cf / user_profiles_rec.taste_cf (vector(64)).
|
|
CF_DIM = 64
|
|
|
|
|
|
def pad(v: np.ndarray, dim: int = CF_DIM) -> np.ndarray:
|
|
"""Маленькая библиотека даёт меньше факторов, чем колонка: добиваем нулями (косинус не меняется)."""
|
|
if v.shape[1] >= dim:
|
|
return v[:, :dim]
|
|
return np.pad(v, ((0, 0), (0, dim - v.shape[1])))
|
|
|
|
|
|
def save_als(conn: psycopg.Connection[Any], model: ALSModel, version: str, metrics: dict[str, Any]) -> None:
|
|
items = pad(normalize(model.item_factors))
|
|
users = pad(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()
|