mrleo1nidandClaude Opus 5.5
ff0aafdb1e
Nightly Beta(α, β) per user and source from wave impressions (reward: complete or like, ×0.99 daily decay) stored in user_profiles_rec.bandit; the wave adds 0.2·(θ − 0.5) per candidate source, sources without data stay neutral. The user's exploration share now follows how well `fresh` does against the other sources. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
69 lines
3.2 KiB
Python
69 lines
3.2 KiB
Python
"""Бандит по источникам кандидатов (docs/recommendations.md#6.6): Beta(α, β) на пару (пользователь, источник).
|
||
|
||
Награда — трек из источника доигран или лайкнут; скип и дизлайк — неудача. Наблюдения затухают ×0.99 в день,
|
||
чтобы бандит подстраивался под смену вкуса. Доля «нового» (exploration) следует за успехом источника `fresh`.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
from collections import defaultdict
|
||
from typing import Any
|
||
|
||
import psycopg
|
||
|
||
DAILY_DECAY = 0.99
|
||
REWARD = {"complete": 1.0, "like": 1.0, "skip": 0.0, "skip_early": 0.0, "dislike": 0.0}
|
||
EXPLORATION_MIN, EXPLORATION_MAX = 0.05, 0.4
|
||
|
||
Params = dict[str, list[float]] # источник → [α, β]
|
||
|
||
|
||
def compute(rows: list[tuple[str, str, str, float]], now: float) -> dict[str, Params]:
|
||
"""rows: (user, source, outcome, served_at) → user → source → [α, β] с приором Beta(1, 1)."""
|
||
out: dict[str, Params] = defaultdict(dict)
|
||
for user, source, outcome, ts in rows:
|
||
r = REWARD.get(outcome)
|
||
if r is None:
|
||
continue
|
||
w = DAILY_DECAY ** max((now - ts) / 86400.0, 0.0)
|
||
ab = out[user].setdefault(source, [1.0, 1.0])
|
||
ab[0] += w * r
|
||
ab[1] += w * (1.0 - r)
|
||
return {u: {s: [round(a, 4), round(b, 4)] for s, (a, b) in p.items()} for u, p in out.items()}
|
||
|
||
|
||
def exploration(params: Params, default: float = 0.15) -> float:
|
||
"""Доля «нового»: успешнее `fresh` относительно остальных источников — больше исследования."""
|
||
fresh = params.get("fresh")
|
||
others = [a / (a + b) for s, (a, b) in params.items() if s != "fresh"]
|
||
if fresh is None or not others:
|
||
return default
|
||
rate = fresh[0] / (fresh[0] + fresh[1])
|
||
base = sum(others) / len(others)
|
||
value = default * (rate / base if base > 0 else 1.0)
|
||
return round(min(max(value, EXPLORATION_MIN), EXPLORATION_MAX), 3)
|
||
|
||
|
||
def update_bandits(conn: psycopg.Connection[Any], now: float) -> int:
|
||
"""Пересчитывает параметры по показам волны за 90 дней. Возвращает число пользователей."""
|
||
rows = conn.execute(
|
||
"""
|
||
SELECT ws.user_id::text, wi.candidate_source, wi.outcome, extract(epoch FROM wi.served_at)::float8
|
||
FROM wave_impressions wi JOIN wave_sessions ws ON ws.id = wi.wave_session_id
|
||
WHERE wi.outcome IS NOT NULL AND wi.served_at > now() - interval '90 days'
|
||
"""
|
||
).fetchall()
|
||
params = compute([(u, s, o, float(ts)) for u, s, o, ts in rows], now)
|
||
with conn.transaction(), conn.cursor() as cur:
|
||
cur.executemany(
|
||
"""
|
||
INSERT INTO user_profiles_rec (user_id, bandit, exploration, updated_at)
|
||
VALUES (%s, %s, %s, now())
|
||
ON CONFLICT (user_id) DO UPDATE SET bandit = EXCLUDED.bandit, exploration = EXCLUDED.exploration,
|
||
updated_at = now()
|
||
""",
|
||
[(u, json.dumps(p), exploration(p)) for u, p in params.items()],
|
||
)
|
||
return len(params)
|