"""Сборка пачки «Моей волны» (v1: эвристика). Чистые функции над массивами — без БД и сети.""" 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 from pipu_rec.wave.contexts import FEATURES, Rule, fit MODEL_VER = "heuristic-v1" # С коллаборативным сигналом (ALS) — суффикс к версии модели в показах. CF_SUFFIX = "+cf" Vec = NDArray[np.float32] @dataclass class CatalogIndex: """Каталог в памяти: строка i — трек ids[i].""" ids: list[str] artist: list[str] # основной исполнитель emb: Vec # (n, d), стандартизирован и нормирован; нулевой — нет анализа 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) @staticmethod def build( ids: list[str], artist: list[str], raw_emb: list[Vec | None], 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) has = np.zeros(n, dtype=bool) for i, e in enumerate(raw_emb): if e is not None: emb[i], has[i] = e, True if has.sum() >= 2: # Стандартизация по библиотеке: косинус сравнивает отличия треков, а не общий «звук библиотеки». mu = emb[has].mean(axis=0) sd = emb[has].std(axis=0) + 1e-6 emb[has] = (emb[has] - mu) / sd emb[has] /= np.linalg.norm(emb[has], axis=1, keepdims=True) + 1e-9 feats = { k: np.array([np.nan if v is None else v for v in features.get(k, [None] * n)], dtype=np.float64) for k in FEATURES } p = np.log1p(np.array(plays, dtype=np.float64)) pop = p / p.max() if n and p.max() > 0 else np.zeros(n) 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 class Profile: """Вкус пользователя: оценки треков и производные.""" scores: dict[str, float] # implicit_score > 0 — нравится disliked: set[str] disliked_artists: set[str] recent: set[str] # сыграны за последние 3 часа played: set[str] # слышал когда-либо liked_recent: set[str] # лайкнутые, игравшие за 24 ч — штраф rediscover: set[str] # любимые, не игравшие > 60 дней exploration: float = 0.15 taste_cf: list[float] | None = None # фактор пользователя ALS bandit: dict[str, list[float]] = field( default_factory=dict ) # источник → [α, β] (docs/recommendations.md#6.6) recent_likes: list[str] = field(default_factory=list) # последние лайки, новые первыми (для item_cf) @dataclass class Session: """Состояние сессии волны (Redis).""" vec: list[float] | None = None served: list[str] = field(default_factory=list) recent_artists: list[str] = field(default_factory=list) skips_in_row: int = 0 @dataclass(frozen=True) 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]). BANDIT_WEIGHT = 0.2 # Вклад реакции в сессионный вектор: s ← normalize(s + η·r·e). REACTION_WEIGHT = {"complete": 1.0, "like": 2.0, "skip_early": -1.0, "dislike": -2.0, "skip": -0.5} ETA = 0.3 def taste_centroids(index: CatalogIndex, profile: Profile, max_k: int = 6, seed: int = 0) -> Vec: """Центроиды вкуса: k-means по эмбеддингам треков с положительной оценкой (веса = оценка). k растёт с историей: 1 центроид на ~8 любимых треков, не больше max_k.""" rows = [(index.pos[t], s) for t, s in profile.scores.items() if s > 0 and t in index.pos] rows = [(i, s) for i, s in rows if index.has_emb[i]] if not rows: return np.zeros((0, index.emb.shape[1]), dtype=np.float32) x = index.emb[[i for i, _ in rows]] w = np.array([s for _, s in rows]) k = int(min(max_k, max(1, len(rows) // 8))) rng = np.random.default_rng(seed) centers = x[rng.choice(len(x), size=k, replace=False)] for _ in range(10): assign = np.argmax(x @ centers.T, axis=1) for c in range(k): m = assign == c if m.any(): v = (x[m] * w[m, None]).sum(axis=0) centers[c] = v / (np.linalg.norm(v) + 1e-9) return centers.astype(np.float32) def update_session( index: CatalogIndex, session: Session, feedback: list[tuple[str, str]], centroids: Vec ) -> None: """Обновляет сессионный вектор и счётчик скипов по реакциям на выданные треки. Три скипа подряд — «смена курса»: вектор переводится к наименее похожему центроиду вкуса.""" d = index.emb.shape[1] vec = np.array(session.vec, dtype=np.float32) if session.vec else np.zeros(d, dtype=np.float32) for track_id, outcome in feedback: i = index.pos.get(track_id) r = REACTION_WEIGHT.get(outcome) if r is None: continue session.skips_in_row = session.skips_in_row + 1 if outcome in ("skip", "skip_early", "dislike") else 0 if i is None or not index.has_emb[i]: continue vec = vec + ETA * r * index.emb[i] norm = float(np.linalg.norm(vec)) vec = vec / norm if norm > 0 else vec if session.skips_in_row >= 3 and len(centroids): sims = centroids @ vec if np.any(vec) else np.zeros(len(centroids)) vec = centroids[int(np.argmin(sims))].copy() session.skips_in_row = 0 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, 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: return idx part = idx[np.argpartition(-scores[idx], k)[:k]] return part def build_batch( index: CatalogIndex, profile: Profile, session: Session, rules: list[Rule], n: int = 5, 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, friend_ids: list[str] | None = None, ) -> list[Pick]: """Кандидаты → фильтры → оценка → соответствие контексту → MMR с правилами разнообразия.""" rng = rng or np.random.default_rng() size = len(index) if size == 0: return [] centroids = taste_centroids(index, profile) svec = np.array(session.vec, dtype=np.float32) if session.vec else None # 2. Фильтры. excluded = profile.disliked | profile.recent | set(session.served) | (exclude or set()) allowed = np.array( [ t not in excluded and a not in profile.disliked_artists for t, a in zip(index.ids, index.artist, strict=True) ] ) if not allowed.any(): allowed = np.array([t not in profile.disliked for t in index.ids]) if not allowed.any(): return [] # Сходства со вкусом и сессией. sim_taste = (index.emb @ centroids.T).max(axis=1) if len(centroids) else np.zeros(size) 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]] = {} for t, s in profile.scores.items(): i = index.pos.get(t) if i is not None: totals.setdefault(index.artist[i], []).append(s) affinity = {a: sum(v) / (len(v) + 2) for a, v in totals.items()} artist_aff = np.array([np.tanh(affinity.get(a, 0.0) / 3) for a in index.artist]) # 1. Источники кандидатов. sources: dict[int, str] = {} def add(idxs: NDArray[np.intp] | list[int], source: str) -> None: for i in idxs: sources.setdefault(int(i), source) if seed_ids: seeds = [index.pos[s] for s in seed_ids if s in index.pos and index.has_emb[index.pos[s]]] if seeds: q = index.emb[seeds].mean(axis=0) add(_top((index.emb @ q).astype(np.float64), allowed & index.has_emb, 100), "seed") friends = [index.pos[t] for t in (friend_ids or []) if t in index.pos and allowed[index.pos[t]]][:10] add(friends, "friends") query = np.maximum(sim_session, sim_taste) if svec is not None or len(centroids) else None if query is not None: # «Новое» помечаем раньше audio_sim: неслышанный трек рядом со вкусом считается исследованием. 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] add(rng.permutation(by_artist)[:60].tolist() if by_artist else [], "artist") redis_ = [index.pos[t] for t in profile.rediscover if t in index.pos and allowed[index.pos[t]]] add(rng.permutation(redis_)[:30].tolist() if redis_ else [], "rediscover") add(_top(index.popularity + rng.random(size) * 0.3, allowed, 30), "popular") # Холодный старт и исследование: случайные треки из разных частей библиотеки. add(rng.permutation(np.nonzero(allowed)[0])[:40].tolist(), "explore") cand = np.array(sorted(sources)) # 3. Оценка v1 (она же — признак и запасной вариант для v2). score = ( 0.35 * sim_session[cand] + 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]) score = score - 0.2 * liked_recent # Бандит: семплирование Томпсона по источникам — удачные для пользователя источники получают бонус. # Источник без наблюдений — нейтральные 0.5 (без шума, иначе холодный старт ранжировался бы случайно). theta: dict[str, float] = {} for src in set(sources.values()): 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 и не больше четверти пула. enough = min(50, max(3 * n, len(cand) // 4)) for threshold in (0.35, 0.15, 0.0): keep = f >= threshold if keep.sum() >= enough or threshold == 0.0: break 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) # Social influence is small, context-sensitive, and survives the learned ranker. score = score + 0.06 * np.isin(cand, friends) * f # 5. MMR + правила: 1 трек исполнителя на пачку, не больше 2 на последние 10; доля «нового». picks: list[Pick] = [] chosen: list[int] = [] recent_artists = session.recent_artists[-10:] want_fresh = max(0, round(n * profile.exploration)) order = list(range(len(cand))) lam = 0.75 while len(picks) < n and order: best_j, best_v = -1, -np.inf for j in order: i = int(cand[j]) if sources[i] == "friends" and any(p.source == "friends" for p in picks): continue a = index.artist[i] if any(index.artist[c] == a for c in chosen) or recent_artists.count(a) >= 2: continue redundancy = max((float(index.emb[i] @ index.emb[c]) for c in chosen), default=0.0) v = lam * float(score[j]) - (1 - lam) * redundancy if v > best_v: best_j, best_v = j, v if best_j < 0: # Правила разнообразия не дают набрать пачку — ослабляем их, но не повторяем треки. if recent_artists: recent_artists = [] continue if chosen: chosen_artists = {index.artist[c] for c in chosen} social_chosen = any(p.source == "friends" for p in picks) rest = [j for j in order if index.artist[int(cand[j])] in chosen_artists and not (sources[int(cand[j])] == "friends" and social_chosen)] if not rest: break best_j = max(rest, key=lambda j: score[j]) else: break 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), features=x[best_j].tolist(), ) ) # Исследование: «новых» (неслышанных) меньше доли exploration — заменяем ими слабейшие выборы. fresh_left = sorted( (j for j in order if sources[int(cand[j])] == "fresh"), key=lambda j: score[j], reverse=True ) while sum(p.source == "fresh" for p in picks) < want_fresh and fresh_left: j = fresh_left.pop(0) i = int(cand[j]) others = [k for k, p in enumerate(picks) if p.source != "fresh"] if not others: break 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), features=x[j].tolist() ) return picks