From 553c9b9f4fb79b11ffcc7d092cb693e5e9b3261e Mon Sep 17 00:00:00 2001 From: Daniil Okhlopkov <5613295+ohld@users.noreply.github.com> Date: Sat, 31 Oct 2026 13:04:43 -0300 Subject: [PATCH 1/2] chore(reco): delete closed ranking experiments and dormant flags Remove recently_liked_blender_v2 / text_light_blender_v1 / viral_shares blender code, cold_start_fresh_viral_v1, source-diversity + shadow-score + full-payload/Sentry-span diagnostics, the always-off disliked-source block filter and the disabled channel-viral broadcast pick. explore_budget_v1, the text_light_lr_smoothed engine, the demote policy and the compact 'recommendation batch diagnostics' log line stay. Feed output unchanged. --- .env.example | 2 - specs/recommendations.md | 2 +- src/config.py | 18 - src/recommendations/blender_experiments.py | 445 +----------------- src/recommendations/broadcast_pick.py | 86 +--- src/recommendations/candidates.py | 124 +---- src/recommendations/cold_start_experiments.py | 96 ---- src/recommendations/meme_queue.py | 65 +-- src/recommendations/pipeline.py | 440 ++--------------- src/recommendations/utils.py | 46 -- .../test_blender_experiments.py | 338 ------------- .../test_block_disliked_sources.py | 140 ------ tests/recommendations/test_broadcast_pick.py | 72 --- .../recommendations/test_cold_start_fresh.py | 176 ------- .../recommendations/test_engine_contracts.py | 1 - tests/recommendations/test_meme_queue.py | 184 +------- tests/recommendations/test_pipeline.py | 154 +----- 17 files changed, 70 insertions(+), 2319 deletions(-) delete mode 100644 src/recommendations/cold_start_experiments.py delete mode 100644 tests/recommendations/test_block_disliked_sources.py delete mode 100644 tests/recommendations/test_cold_start_fresh.py diff --git a/.env.example b/.env.example index c34ebc1d..e6931232 100644 --- a/.env.example +++ b/.env.example @@ -49,8 +49,6 @@ OPENROUTER_API_KEY_SECONDARY= # Friend challenge (H12): apply migration first; 90/10 treatment/holdout. FRIEND_CHALLENGE_ENABLED=false -# Retention broadcasts: HQ pick by default (channel-viral disabled 2026-10-30, H3c). -BROADCAST_CHANNEL_VIRAL_PICK_ENABLED=false # Telethon (for channel stats collection) TELEGRAM_API_ID= diff --git a/specs/recommendations.md b/specs/recommendations.md index e2d33409..1ba31332 100644 --- a/specs/recommendations.md +++ b/specs/recommendations.md @@ -42,7 +42,7 @@ Removed engines: `fast_dopamine`, `classic`, `multiply_all_scores`, `selected_so | Mature | 100+ | Recently-liked blender v2; the text-light overlay is disabled unless explicitly enabled by a registered experiment | [`pipeline.py`](../src/recommendations/pipeline.py) | | Moderator/Admin | user_type check | 75% low_sent_pool + 25% regular (by maturity) | [`pipeline.py`](../src/recommendations/pipeline.py) | -`fixed_pos={0: "lr_smoothed"}` forces first position to `lr_smoothed` in blended mode. If the `text_light_blender_v1` overlay is explicitly enabled, its treatment slot becomes `text_light_lr_smoothed`. +`fixed_pos={0: "lr_smoothed"}` forces first position to `lr_smoothed` in blended mode. ## Diagnostics Refactor diff --git a/src/config.py b/src/config.py index 916f8fce..6a4241ff 100644 --- a/src/config.py +++ b/src/config.py @@ -57,36 +57,18 @@ class Config(BaseSettings): # FFM-1882: narrow true-new cold-start positions 2-10 experiment. # Roll back by setting this to false in production env. COLD_START_CANDIDATE_GUARDRAILS_ENABLED: bool = True - # H10 cold_start_fresh_viral_v1: KILLED 2026-10-30 (first tap and 1h block - # rate both worse in treatment). Off = no new assignment, everyone is control. - COLD_START_FRESH_VIRAL_EXPERIMENT_ENABLED: bool = False # explore_budget_v1: 50/50 non-cold-start users get the labeling budget, # fresh-viral blend slot and idle-return first slot. Kill switch = all control. EXPLORE_BUDGET_V1_ENABLED: bool = True - # Recommendation batch diagnostics are realtime operational data, not - # durable product facts. Keep compact logs/spans always on and sample full - # payloads so future Prometheus/Grafana adapters can reuse the same seam. - RECOMMENDATION_DIAGNOSTICS_SAMPLE_RATE: float = 0.01 - RECOMMENDATION_SOURCE_DIVERSITY_ENABLED: bool = False - RECOMMENDATION_SHADOW_SCORING_ENABLED: bool = True # Source affinity policies (skip/dislike ≈ "next", not "ban channel"). # Soft demote majority-dislike sources in ranking (default ON). RECOMMENDATION_DEMOTE_DISLIKED_SOURCES: bool = True RECOMMENDATION_DEMOTE_DISLIKED_MIN_REACTIONS: int = 5 RECOMMENDATION_DEMOTE_DISLIKED_MULTIPLIER: float = 0.15 - # Hard exclude only strong hate (default OFF — empty-queue risk). - RECOMMENDATION_BLOCK_DISLIKED_SOURCES: bool = False - RECOMMENDATION_BLOCK_DISLIKED_MIN_REACTIONS: int = 15 - RECOMMENDATION_BLOCK_DISLIKED_RATIO: float = 3.0 # ndislikes >= 3 * nlikes # Retention broadcasts: pick high-confidence meme (affinity + LR) instead of # blind Redis queue pop. Kill switch rolls back to queue path. BROADCAST_HIGH_QUALITY_PICK_ENABLED: bool = True - # Prefer last-7-day channel posts ranked by forwards (community-verified). - # Falls back to HQ/queue. Skips known channel members. - # Off since 2026-10-30 (H3c disable rule): viral LR 5-20 pp below HQ with - # equal/worse reactions; HQ is the default push pick. - BROADCAST_CHANNEL_VIRAL_PICK_ENABLED: bool = False # broadcast_cadence_holdout_v1: no-push holdout per cadence step (15m/24h # 15%, 1w/2w/4w 20%). See experiments/active/2026-10-30-*. BROADCAST_CADENCE_HOLDOUT_ENABLED: bool = True diff --git a/src/recommendations/blender_experiments.py b/src/recommendations/blender_experiments.py index 887eb282..61394139 100644 --- a/src/recommendations/blender_experiments.py +++ b/src/recommendations/blender_experiments.py @@ -1,453 +1,10 @@ import hashlib import logging -import uuid -from asyncio import sleep -from typing import Any -import orjson -from sqlalchemy import text - -from src.database import fetch_one -from src.feed_turn.planner import MATURE_BLEND_WEIGHTS -from src.redis import redis_client -from src.tgbot.service import ( - assign_experiment, - get_experiment_assignment, - get_experiment_variant, -) +from src.tgbot.service import assign_experiment, get_experiment_assignment logger = logging.getLogger(__name__) -RECENTLY_LIKED_BLENDER_V2_EXPERIMENT_ID = "recently_liked_blender_v2" -RECENTLY_LIKED_BLENDER_V2_CONTROL = "control" -RECENTLY_LIKED_BLENDER_V2_TREATMENT = "treatment" -RECENTLY_LIKED_BLENDER_V2_EXCLUDED = "excluded_high_volume_skipper" -RECENTLY_LIKED_BLENDER_V2_ENROLLMENT_FROZEN = True -RECENTLY_LIKED_BLENDER_V2_ENROLLMENT_FROZEN_AT = "2026-07-23T18:05:54Z" -RECENTLY_LIKED_BLENDER_V2_SAMPLE_GATE_PER_VARIANT = 1000 -RECENTLY_LIKED_BLENDER_V2_MATURE_MIN_MEMES_SENT = 100 -RECENTLY_LIKED_BLENDER_V2_SKIPPER_MIN_REACTIONS_7D = 50 -RECENTLY_LIKED_BLENDER_V2_SKIPPER_MAX_LR_7D = 0.20 -RECENTLY_LIKED_BLENDER_V2_QUARTILE_BOUNDARIES_CACHE_KEY = ( - "recently_liked_blender_v2:lr_quartile_boundaries" -) -RECENTLY_LIKED_BLENDER_V2_QUARTILE_BOUNDARIES_LOCK_KEY = ( - "recently_liked_blender_v2:lr_quartile_boundaries:lock" -) -RECENTLY_LIKED_BLENDER_V2_QUARTILE_BOUNDARIES_TTL_SECONDS = 15 * 60 -RECENTLY_LIKED_BLENDER_V2_QUARTILE_BOUNDARIES_LOCK_SECONDS = 30 -TEXT_LIGHT_BLENDER_V1_EXPERIMENT_ID = "text_light_blender_v1" -TEXT_LIGHT_BLENDER_V1_CONTROL = "control" -TEXT_LIGHT_BLENDER_V1_TREATMENT = "treatment_text_light_lr_smoothed" -TEXT_LIGHT_BLENDER_V1_MAX_OCR_WORDS = 30 -TEXT_LIGHT_BLENDER_V1_SAMPLE_GATE_PER_VARIANT = 1000 - -# Control = default mature plan weights (SSOT in feed_turn.planner). -# After recently_liked_blender_v2 closeout (2026-08-09), the shipped default IS -# the former treatment map. Historical treatment alias kept for old tests/docs. -MATURE_BLENDER_CONTROL_WEIGHTS = dict(MATURE_BLEND_WEIGHTS) -MATURE_BLENDER_TREATMENT_WEIGHTS = dict(MATURE_BLEND_WEIGHTS) -# Pre-closeout control (for archaeology / partial-read comparisons only). -MATURE_BLENDER_LEGACY_CONTROL_WEIGHTS = { - "best_uploaded_memes": 0.3, - "like_spread_and_recent_memes": 0.3, - "lr_smoothed": 0.4, - "recently_liked": 0.2, - "goat": 0.1, - "es_ranked": 0.1, -} - - -def _variant_for_quartile(user_id: int, lr_quartile: int) -> str: - key = f"{RECENTLY_LIKED_BLENDER_V2_EXPERIMENT_ID}:{lr_quartile}:{user_id}" - bucket = int.from_bytes(hashlib.sha256(key.encode("utf-8")).digest()[:8], "big") % 2 - if bucket == 0: - return RECENTLY_LIKED_BLENDER_V2_CONTROL - return RECENTLY_LIKED_BLENDER_V2_TREATMENT - - -def _weights_for_variant(variant: str) -> dict[str, float]: - if variant == RECENTLY_LIKED_BLENDER_V2_TREATMENT: - return dict(MATURE_BLENDER_TREATMENT_WEIGHTS) - return dict(MATURE_BLENDER_CONTROL_WEIGHTS) - - -def _text_light_weights(base_weights: dict[str, float]) -> dict[str, float]: - weights = dict(base_weights) - lr_weight = weights.pop("lr_smoothed", 0.0) - weights["text_light_lr_smoothed"] = lr_weight - return weights - - -def _text_light_variant_for_user(user_id: int) -> str: - key = f"{TEXT_LIGHT_BLENDER_V1_EXPERIMENT_ID}:{user_id}" - bucket = int.from_bytes(hashlib.sha256(key.encode("utf-8")).digest()[:8], "big") % 2 - if bucket == 0: - return TEXT_LIGHT_BLENDER_V1_CONTROL - return TEXT_LIGHT_BLENDER_V1_TREATMENT - - -def _coerce_lr_quartile_boundaries(raw_boundaries: Any) -> tuple[float, float, float] | None: - if raw_boundaries is None: - return None - - if isinstance(raw_boundaries, str): - raw_boundaries = raw_boundaries.strip("{}").split(",") - - try: - boundaries = tuple(float(value) for value in raw_boundaries) - except (TypeError, ValueError): - logger.warning( - "invalid recently_liked blender v2 LR quartile boundaries: %r", - raw_boundaries, - ) - return None - - if len(boundaries) != 3 or any(value != value for value in boundaries): - logger.warning( - "invalid recently_liked blender v2 LR quartile boundaries: %r", - raw_boundaries, - ) - return None - - return tuple(max(0.0, min(1.0, value)) for value in boundaries) - - -def _lr_quartile_from_boundaries(lr_7d: float, boundaries: tuple[float, float, float]) -> int: - if lr_7d <= boundaries[0]: - return 1 - if lr_7d <= boundaries[1]: - return 2 - if lr_7d <= boundaries[2]: - return 3 - return 4 - - -async def _get_cached_lr_quartile_boundaries() -> tuple[float, float, float] | None: - try: - cached = await redis_client.get(RECENTLY_LIKED_BLENDER_V2_QUARTILE_BOUNDARIES_CACHE_KEY) - except Exception: - logger.warning( - "recently_liked blender v2 LR quartile boundary cache read failed", - exc_info=True, - ) - return None - - if not cached: - return None - - try: - return _coerce_lr_quartile_boundaries(orjson.loads(cached)) - except orjson.JSONDecodeError: - logger.warning( - "recently_liked blender v2 LR quartile boundary cache payload is invalid", - exc_info=True, - ) - return None - - -async def _cache_lr_quartile_boundaries(boundaries: tuple[float, float, float]) -> None: - try: - await redis_client.set( - RECENTLY_LIKED_BLENDER_V2_QUARTILE_BOUNDARIES_CACHE_KEY, - orjson.dumps(boundaries), - ex=RECENTLY_LIKED_BLENDER_V2_QUARTILE_BOUNDARIES_TTL_SECONDS, - ) - except Exception: - logger.warning( - "recently_liked blender v2 LR quartile boundary cache write failed", - exc_info=True, - ) - - -async def _release_lr_quartile_boundaries_lock(token: str) -> None: - try: - current = await redis_client.get(RECENTLY_LIKED_BLENDER_V2_QUARTILE_BOUNDARIES_LOCK_KEY) - if current == token: - await redis_client.delete(RECENTLY_LIKED_BLENDER_V2_QUARTILE_BOUNDARIES_LOCK_KEY) - except Exception: - logger.warning( - "recently_liked blender v2 LR quartile boundary lock release failed", - exc_info=True, - ) - - -async def _calculate_recent_7d_lr_quartile_boundaries() -> tuple[float, float, float] | None: - query = text( - """ - WITH recent_user_lr AS ( - SELECT - COUNT(*) FILTER (WHERE umr.reaction_id = 1)::float - / NULLIF(COUNT(*), 0) AS lr_7d - FROM user_meme_reaction umr - INNER JOIN user_stats us ON us.user_id = umr.user_id - WHERE umr.reacted_at >= NOW() - INTERVAL '7 days' - AND umr.reaction_id IN (1, 2) - AND us.nmemes_sent >= :mature_min_memes_sent - GROUP BY umr.user_id - HAVING COUNT(*) > 0 - ) - SELECT - PERCENTILE_CONT(ARRAY[0.25, 0.5, 0.75]) - WITHIN GROUP (ORDER BY lr_7d) AS lr_boundaries - FROM recent_user_lr - """ - ) - row = await fetch_one( - query, - {"mature_min_memes_sent": RECENTLY_LIKED_BLENDER_V2_MATURE_MIN_MEMES_SENT}, - ) - return _coerce_lr_quartile_boundaries(row["lr_boundaries"] if row else None) - - -async def get_recent_7d_lr_quartile_boundaries() -> tuple[float, float, float] | None: - """Return cached mature-user 7d LR quartile cut points. - - Assignment is on the user-facing queue path, so only the lock holder performs - the global mature-user aggregate. Other workers wait briefly for the cache, - then skip enrollment until real cut points are cached rather than - stampeding the DB or assigning users against fallback boundaries. - """ - cached = await _get_cached_lr_quartile_boundaries() - if cached is not None: - return cached - - token = str(uuid.uuid4()) - try: - acquired = await redis_client.set( - RECENTLY_LIKED_BLENDER_V2_QUARTILE_BOUNDARIES_LOCK_KEY, - token, - nx=True, - ex=RECENTLY_LIKED_BLENDER_V2_QUARTILE_BOUNDARIES_LOCK_SECONDS, - ) - except Exception: - logger.warning( - "recently_liked blender v2 LR quartile boundary lock failed", - exc_info=True, - ) - return None - - if acquired: - try: - boundaries = await _calculate_recent_7d_lr_quartile_boundaries() - if boundaries is not None: - await _cache_lr_quartile_boundaries(boundaries) - return boundaries - finally: - await _release_lr_quartile_boundaries_lock(token) - - for _ in range(10): - await sleep(0.1) - cached = await _get_cached_lr_quartile_boundaries() - if cached is not None: - return cached - - logger.warning("recently_liked blender v2 LR quartile boundaries still uncached after waiting") - return None - - -async def get_recent_7d_lr_assignment_metrics(user_id: int) -> dict[str, Any] | None: - """Return enrollment-time 7d LR metrics and mature-user LR quartile.""" - boundaries = await get_recent_7d_lr_quartile_boundaries() - if boundaries is None: - return None - - query = text( - """ - SELECT - COUNT(*) FILTER (WHERE umr.reaction_id = 1) AS likes_7d, - COUNT(*) AS reactions_7d, - COALESCE( - COUNT(*) FILTER (WHERE umr.reaction_id = 1)::float / NULLIF(COUNT(*), 0), - 0.0 - ) AS lr_7d - FROM user_meme_reaction umr - WHERE umr.user_id = :user_id - AND umr.reacted_at >= NOW() - INTERVAL '7 days' - AND umr.reaction_id IN (1, 2) - """ - ) - row = await fetch_one(query, {"user_id": user_id}) - if row is None: - return { - "likes_7d": 0, - "reactions_7d": 0, - "lr_7d": 0.0, - "lr_quartile": 1, - "lr_quartile_boundaries": list(boundaries), - } - lr_7d = float(row["lr_7d"] or 0.0) - return { - "likes_7d": int(row["likes_7d"] or 0), - "reactions_7d": int(row["reactions_7d"] or 0), - "lr_7d": lr_7d, - "lr_quartile": _lr_quartile_from_boundaries(lr_7d, boundaries), - "lr_quartile_boundaries": list(boundaries), - } - - -def build_recently_liked_blender_v2_assignment( - user_id: int, - metrics: dict[str, Any], -) -> tuple[str, dict[str, Any]]: - lr_7d = float(metrics["lr_7d"]) - reactions_7d = int(metrics["reactions_7d"]) - lr_quartile = int(metrics["lr_quartile"]) - high_volume_skipper = ( - reactions_7d > RECENTLY_LIKED_BLENDER_V2_SKIPPER_MIN_REACTIONS_7D - and lr_7d < RECENTLY_LIKED_BLENDER_V2_SKIPPER_MAX_LR_7D - ) - - if high_volume_skipper: - variant = RECENTLY_LIKED_BLENDER_V2_EXCLUDED - excluded_reason = "lr_7d_below_20pct_and_reactions_7d_above_50" - else: - variant = _variant_for_quartile(user_id, lr_quartile) - excluded_reason = None - - assignment_metadata = { - "assignment_strategy": "sha256(experiment_id:lr_quartile:user_id)%2", - "lr_quartile": lr_quartile, - "likes_7d": int(metrics["likes_7d"]), - "reactions_7d": reactions_7d, - "lr_7d": round(lr_7d, 6), - "high_volume_skipper": high_volume_skipper, - "excluded_reason": excluded_reason, - "sample_gate_per_variant": RECENTLY_LIKED_BLENDER_V2_SAMPLE_GATE_PER_VARIANT, - "day3_guardrail": { - "checkpoint": "rollout_time_plus_3_days", - "failure_lr_delta_pp": -4, - "primary_read_rule": "sample_gate_per_variant", - }, - "assigned_weights": _weights_for_variant(variant), - } - if "lr_quartile_boundaries" in metrics: - assignment_metadata["lr_quartile_boundaries"] = [ - round(float(boundary), 6) for boundary in metrics["lr_quartile_boundaries"] - ] - return variant, assignment_metadata - - -async def get_or_assign_recently_liked_blender_v2_variant(user_id: int) -> str: - assignment = await get_experiment_assignment( - user_id, - RECENTLY_LIKED_BLENDER_V2_EXPERIMENT_ID, - ) - if assignment is not None: - return assignment["variant"] - - if RECENTLY_LIKED_BLENDER_V2_ENROLLMENT_FROZEN: - return RECENTLY_LIKED_BLENDER_V2_CONTROL - - metrics = await get_recent_7d_lr_assignment_metrics(user_id) - if metrics is None: - return RECENTLY_LIKED_BLENDER_V2_CONTROL - - proposed_variant, assignment_metadata = build_recently_liked_blender_v2_assignment( - user_id, - metrics, - ) - inserted = await assign_experiment( - user_id, - RECENTLY_LIKED_BLENDER_V2_EXPERIMENT_ID, - proposed_variant, - assignment_metadata, - ) - if inserted: - return proposed_variant - - return ( - await get_experiment_variant(user_id, RECENTLY_LIKED_BLENDER_V2_EXPERIMENT_ID) - or RECENTLY_LIKED_BLENDER_V2_CONTROL - ) - - -async def get_recently_liked_blender_v2_weights(user_id: int) -> dict[str, float]: - """Mature blend weights after recently_liked_blender_v2 closeout. - - Treatment won on session depth and user-level LR with no LR regression - (see docs/analyst/recently-liked-blender-v2-closeout.md). All mature users - now receive the shipped default; no new assignment / dual-path. - """ - del user_id # API kept for pipeline mature_weights_func signature - return dict(MATURE_BLEND_WEIGHTS) - - -def build_text_light_blender_v1_assignment( - user_id: int, - base_weights: dict[str, float], -) -> tuple[str, dict[str, Any]]: - variant = _text_light_variant_for_user(user_id) - treatment_weights = _text_light_weights(base_weights) - assignment_metadata = { - "assignment_strategy": "sha256(experiment_id:user_id)%2", - "max_ocr_words": TEXT_LIGHT_BLENDER_V1_MAX_OCR_WORDS, - "filtered_engine": "text_light_lr_smoothed", - "control_engine": "lr_smoothed", - "sample_gate_per_variant": TEXT_LIGHT_BLENDER_V1_SAMPLE_GATE_PER_VARIANT, - "primary_read": "compare post-assignment like_rate, session depth, fast-skip rate", - "base_weights": dict(base_weights), - "assigned_weights": ( - treatment_weights if variant == TEXT_LIGHT_BLENDER_V1_TREATMENT else dict(base_weights) - ), - } - return variant, assignment_metadata - - -async def get_or_assign_text_light_blender_v1_variant( - user_id: int, - base_weights: dict[str, float], -) -> str: - assignment = await get_experiment_assignment( - user_id, - TEXT_LIGHT_BLENDER_V1_EXPERIMENT_ID, - ) - if assignment is not None: - return assignment["variant"] - - proposed_variant, assignment_metadata = build_text_light_blender_v1_assignment( - user_id, - base_weights, - ) - inserted = await assign_experiment( - user_id, - TEXT_LIGHT_BLENDER_V1_EXPERIMENT_ID, - proposed_variant, - assignment_metadata, - ) - if inserted: - return proposed_variant - - return ( - await get_experiment_variant(user_id, TEXT_LIGHT_BLENDER_V1_EXPERIMENT_ID) - or TEXT_LIGHT_BLENDER_V1_CONTROL - ) - - -async def get_text_light_blender_v1_weights( - user_id: int, - base_weights: dict[str, float], -) -> dict[str, float]: - try: - variant = await get_or_assign_text_light_blender_v1_variant(user_id, base_weights) - except Exception: - logger.warning( - "text-light blender v1 assignment failed for user %d", - user_id, - exc_info=True, - ) - return dict(base_weights) - - if variant == TEXT_LIGHT_BLENDER_V1_TREATMENT: - return _text_light_weights(base_weights) - return dict(base_weights) - - -# viral_shares_blender_v1 closed 2026-10-30 (KILL): enrollment stopped, every -# mature user gets the shipped default. Existing assignment rows stay for readouts. -async def get_mature_blend_weights_with_experiments(user_id: int) -> dict[str, float]: - return await get_recently_liked_blender_v2_weights(user_id) - # --- explore_budget_v1: labeling budget + fresh-viral + idle-return first slot --- # See experiments/active/2026-10-30-explore-budget.md. diff --git a/src/recommendations/broadcast_pick.py b/src/recommendations/broadcast_pick.py index 02114f6c..5a06c1f0 100644 --- a/src/recommendations/broadcast_pick.py +++ b/src/recommendations/broadcast_pick.py @@ -2,8 +2,7 @@ Feed queue pop is optimized for continuous scrolling. Reengagement pushes need a single strong meme that earns a *fast* reaction when the user returns. -Prefer last-week channel-viral posts (community-verified forwards), then liked -sources + proven like rate. Never majority-dislike sources. +Prefer liked sources + proven like rate; majority-dislike sources are demoted. """ from __future__ import annotations @@ -15,19 +14,14 @@ from src.config import settings from src.database import fetch_one from src.recommendations.meme_queue import check_queue, get_next_meme_for_user -from src.recommendations.utils import ( - block_disliked_sources_sql_filter, - disliked_source_demote_sql, -) +from src.recommendations.utils import disliked_source_demote_sql from src.storage.schemas import MemeData -from src.tgbot.constants import TELEGRAM_CHANNEL_EN_CHAT_ID, TELEGRAM_CHANNEL_RU_CHAT_ID logger = logging.getLogger(__name__) # Delivery-path labels for analytics (dwell / reactivation). BROADCAST_RECOMMENDED_BY = "broadcast_reengagement" BROADCAST_HQ_RECOMMENDED_BY = "broadcast_reengagement_hq" -BROADCAST_CHANNEL_VIRAL_RECOMMENDED_BY = "broadcast_channel_viral" # Quality floors for the HQ pick (explicit reactions, not lr_smoothed alone). _HQ_MIN_EXPLICIT_REACTIONS = 15 @@ -37,16 +31,9 @@ async def pick_reengagement_meme(user_id: int) -> tuple[MemeData | None, str]: """Return (meme, recommended_by_label) for a retention push. - Tries last-7-day channel-viral posts first (label ``broadcast_channel_viral``), - then the affinity HQ pick (``broadcast_reengagement_hq``), then the feed - queue (``broadcast_reengagement``). Channel-hit experiment users skip the - viral path so the feed test stays the only channel-viral treatment. + Tries the affinity HQ pick (``broadcast_reengagement_hq``), then the feed + queue (``broadcast_reengagement``). """ - if settings.BROADCAST_CHANNEL_VIRAL_PICK_ENABLED: - meme = await _fetch_channel_viral_reengagement_meme(user_id) - if meme is not None: - return meme, BROADCAST_CHANNEL_VIRAL_RECOMMENDED_BY - if settings.BROADCAST_HIGH_QUALITY_PICK_ENABLED: meme = await _fetch_high_quality_reengagement_meme(user_id) if meme is not None: @@ -91,7 +78,6 @@ async def _fetch_high_quality_reengagement_meme(user_id: int) -> MemeData | None AND (MS.nlikes + MS.ndislikes) >= :min_reactions AND (MS.nlikes::float / NULLIF(MS.nlikes + MS.ndislikes, 0)) >= :min_raw_like_rate - {block_disliked_sources_sql_filter()} ORDER BY -1 * COALESCE( (UMSS.nlikes + 1.) / (UMSS.nlikes + UMSS.ndislikes + 1.), @@ -114,70 +100,6 @@ async def _fetch_high_quality_reengagement_meme(user_id: int) -> MemeData | None return _meme_from_row(row, BROADCAST_HQ_RECOMMENDED_BY) -async def _fetch_channel_viral_reengagement_meme(user_id: int) -> MemeData | None: - """Unseen image from our channels in the last 7 days, ranked by forwards. - - Skips known subscribers (they already got the channel push). Empty pool is - normal — HQ/queue still fire. - """ - query = """ - SELECT - M.id - , M.type - , M.telegram_file_id - , M.caption - , COALESCE(MS.nlikes, 0) AS nlikes - FROM crossposting CP - INNER JOIN meme M - ON M.id = CP.meme_id - LEFT JOIN meme_stats MS - ON MS.meme_id = M.id - INNER JOIN user_language L - ON L.language_code = M.language_code - AND L.user_id = :user_id - INNER JOIN LATERAL ( - SELECT S.views, S.forwards - FROM crossposting_snapshots S - WHERE S.channel = CP.channel - AND S.meme_id = CP.meme_id - AND S.telegram_message_id = CP.telegram_message_id - ORDER BY S.snapshot_at DESC - LIMIT 1 - ) SNAP ON TRUE - LEFT JOIN user_meme_reaction R - ON R.meme_id = M.id - AND R.user_id = :user_id - WHERE CP.channel IN ('tgchannelru', 'tgchannelen') - AND CP.created_at >= (NOW() AT TIME ZONE 'UTC') - interval '7 days' - AND CP.created_at < (NOW() AT TIME ZONE 'UTC') - interval '24 hours' - AND M.status = 'published' - AND M.type = 'image' - AND M.duplicate_of IS NULL - AND M.telegram_file_id IS NOT NULL - AND R.meme_id IS NULL - AND SNAP.views >= 50 - AND SNAP.forwards >= 1 - AND NOT EXISTS ( - SELECT 1 FROM user_tg_chat_membership OLD - WHERE OLD.user_tg_id = :user_id - AND OLD.chat_id = CASE CP.channel - WHEN 'tgchannelru' THEN CAST(:ru_chat_id AS bigint) - ELSE CAST(:en_chat_id AS bigint) END - ) - ORDER BY SNAP.forwards DESC, SNAP.views DESC, CP.created_at DESC - LIMIT 1 - """ - row = await fetch_one( - text(query), - { - "user_id": user_id, - "ru_chat_id": TELEGRAM_CHANNEL_RU_CHAT_ID, - "en_chat_id": TELEGRAM_CHANNEL_EN_CHAT_ID, - }, - ) - return _meme_from_row(row, BROADCAST_CHANNEL_VIRAL_RECOMMENDED_BY) - - def _meme_from_row(row: dict | None, recommended_by: str) -> MemeData | None: if not row: return None diff --git a/src/recommendations/candidates.py b/src/recommendations/candidates.py index 05706b82..e12f406f 100644 --- a/src/recommendations/candidates.py +++ b/src/recommendations/candidates.py @@ -6,7 +6,6 @@ from src.database import fetch_all, fetch_one from src.recommendations.utils import ( - block_disliked_sources_sql_filter, disliked_source_demote_sql, exclude_meme_ids_sql_filter, ) @@ -36,8 +35,6 @@ COLD_START_ADAPT_RECOMMENDED_BY = "cold_start_adapt" COLD_START_EXPLORE_GUARDED_RECOMMENDED_BY = "cold_start_explore_guarded" COLD_START_ADAPT_GUARDED_RECOMMENDED_BY = "cold_start_adapt_guarded" -COLD_START_EXPLORE_FRESH_RECOMMENDED_BY = "cold_start_explore_fresh" -COLD_START_EXPLORE_FRESH_GUARDED_RECOMMENDED_BY = "cold_start_explore_fresh_guarded" _OCR_TEXT_SQL = "trim(coalesce(M.ocr_result->>'text', M.ocr_result->'raw_result'->>'ocr_text', ''))" TEXT_LIGHT_OCR_FILTER_SQL = f""" @@ -76,19 +73,9 @@ def _build_params( return params -def _cold_start_recommended_by( - engine: str, - candidate_guardrails_enabled: bool, - *, - max_age_days: int | None = None, -) -> str: - if engine == COLD_START_EXPLORE_RECOMMENDED_BY: - if max_age_days: - if candidate_guardrails_enabled: - return COLD_START_EXPLORE_FRESH_GUARDED_RECOMMENDED_BY - return COLD_START_EXPLORE_FRESH_RECOMMENDED_BY - if candidate_guardrails_enabled: - return COLD_START_EXPLORE_GUARDED_RECOMMENDED_BY +def _cold_start_recommended_by(engine: str, candidate_guardrails_enabled: bool) -> str: + if engine == COLD_START_EXPLORE_RECOMMENDED_BY and candidate_guardrails_enabled: + return COLD_START_EXPLORE_GUARDED_RECOMMENDED_BY if engine == COLD_START_ADAPT_RECOMMENDED_BY and candidate_guardrails_enabled: return COLD_START_ADAPT_GUARDED_RECOMMENDED_BY return engine @@ -105,23 +92,16 @@ def _cold_start_params( *, engine: str, candidate_guardrails_enabled: bool, - max_age_days: int | None = None, ) -> dict[str, Any]: params = _build_params( user_id, limit, exclude_meme_ids, - recommended_by=_cold_start_recommended_by( - engine, - candidate_guardrails_enabled, - max_age_days=max_age_days, - ), + recommended_by=_cold_start_recommended_by(engine, candidate_guardrails_enabled), text_light_max_ocr_words=TEXT_LIGHT_MAX_OCR_WORDS, ) if candidate_guardrails_enabled: params["cold_start_guardrail_source_urls"] = list(COLD_START_GUARDRAIL_SOURCE_URLS) - if max_age_days: - params["cold_start_max_age_days"] = max_age_days return params @@ -161,7 +141,6 @@ async def best_uploaded_memes( AND R.meme_id IS NULL AND S.type = 'user upload' {exclude_meme_ids_sql_filter(exclude_meme_ids)} - {block_disliked_sources_sql_filter()} ORDER BY -1 * COALESCE((UMSS.nlikes + 1.) / (UMSS.nlikes + UMSS.ndislikes + 1.), 0.5) @@ -212,7 +191,6 @@ async def like_spread_and_recent_memes( AND MS.nlikes > MS.ndislikes {pool_filter} {exclude_meme_ids_sql_filter(exclude_meme_ids)} - {block_disliked_sources_sql_filter()} ORDER BY -1 * (MS.nlikes - MS.ndislikes) / (MS.nmemes_sent + 1.) * {disliked_source_demote_sql()} @@ -284,7 +262,6 @@ async def _get_lr_smoothed_candidates( {min_sends_filter} {text_light_filter} {exclude_meme_ids_sql_filter(exclude_meme_ids)} - {block_disliked_sources_sql_filter()} ORDER BY -1 * COALESCE((UMSS.nlikes + 1.) / (UMSS.nlikes + UMSS.ndislikes + 1.), 0.5) @@ -372,7 +349,6 @@ async def get_es_ranked( AND R.meme_id IS NULL AND MS.engagement_score > 0 {exclude_meme_ids_sql_filter(exclude_meme_ids)} - {block_disliked_sources_sql_filter()} ORDER BY -1 * COALESCE((UMSS.nlikes + 1.) / (UMSS.nlikes + UMSS.ndislikes + 1.), 0.5) @@ -458,7 +434,6 @@ async def goat( WHERE 1=1 {exclude_meme_ids_sql_filter(exclude_meme_ids)} - {block_disliked_sources_sql_filter()} ORDER BY SCORES.score DESC NULLS LAST LIMIT :limit """ @@ -522,7 +497,6 @@ async def get_recently_liked( AND M.status = 'ok' AND R.meme_id IS NULL {exclude_meme_ids_sql_filter(exclude_meme_ids)} - {block_disliked_sources_sql_filter()} -- Prefer non-disliked sources first (demote mult 1.0 before 0.15) ORDER BY {disliked_source_demote_sql()} DESC, M.id DESC LIMIT :limit @@ -535,7 +509,6 @@ async def cold_start_explore( limit: int = 5, exclude_meme_ids: list[int] = [], candidate_guardrails_enabled: bool = False, - max_age_days: int | None = None, ): """Phase 1 cold start: quality-first selection for new user first impression. @@ -553,26 +526,8 @@ async def cold_start_explore( correction. Used for memes 1-5 (first impression). - - ``max_age_days`` (H10 treatment) keeps the same quality floors but only - among memes created in that window, ranked newest first. Empty pool is - filled by the pipeline from unfiltered cold_start_explore. """ - age_filter_sql = "" - if max_age_days: - age_filter_sql = "AND M.created_at >= NOW() - (:cold_start_max_age_days * INTERVAL '1 day')" - order_sql = ( - """M.created_at DESC, - MS.lr_smoothed DESC NULLS LAST, - (MS.nlikes::float / NULLIF(MS.nlikes + MS.ndislikes, 0)) DESC, - (MS.nlikes + MS.ndislikes) DESC""" - if max_age_days - else """MS.lr_smoothed DESC NULLS LAST, - (MS.nlikes::float / NULLIF(MS.nlikes + MS.ndislikes, 0)) DESC, - (MS.nlikes + MS.ndislikes) DESC""" - ) - query = f""" SELECT M.id @@ -599,13 +554,13 @@ async def cold_start_explore( AND MS.lr_smoothed >= :cold_start_explore_min_lr_smoothed AND (MS.nlikes::float / NULLIF(MS.nlikes + MS.ndislikes, 0)) >= :cold_start_explore_min_raw_like_rate - {age_filter_sql} {TEXT_LIGHT_OCR_FILTER_SQL} {_cold_start_guardrail_source_filter(candidate_guardrails_enabled)} {exclude_meme_ids_sql_filter(exclude_meme_ids)} - {block_disliked_sources_sql_filter()} - ORDER BY {order_sql} + ORDER BY MS.lr_smoothed DESC NULLS LAST, + (MS.nlikes::float / NULLIF(MS.nlikes + MS.ndislikes, 0)) DESC, + (MS.nlikes + MS.ndislikes) DESC LIMIT :limit """ params = _cold_start_params( @@ -614,7 +569,6 @@ async def cold_start_explore( exclude_meme_ids, engine=COLD_START_EXPLORE_RECOMMENDED_BY, candidate_guardrails_enabled=candidate_guardrails_enabled, - max_age_days=max_age_days, ) params.update( { @@ -689,7 +643,6 @@ async def cold_start_adapt( {TEXT_LIGHT_OCR_FILTER_SQL} {_cold_start_guardrail_source_filter(candidate_guardrails_enabled)} {exclude_meme_ids_sql_filter(exclude_meme_ids)} - {block_disliked_sources_sql_filter()} ORDER BY -1 * GREATEST(COALESCE(RR.raw_weight, 0) + 0.5, 0.1) @@ -708,68 +661,6 @@ async def cold_start_adapt( ) -async def viral_shares( - user_id: int, - limit: int = 10, - exclude_meme_ids: list[int] = [], -): - """Memes with proven share-click conversion (unique non-self deep-link clickers). - - Ranks by invited_count / ln(nmemes_sent + e) so high-volume memes do not - dominate purely by raw share-click counts. Quality floor keeps junk out of - the growth-oriented blend slot. - """ - query = f""" - SELECT - M.id - , M.type, M.telegram_file_id, M.caption - , 'viral_shares' AS recommended_by - , COALESCE(MS.nlikes, 0) AS nlikes - , COALESCE(MS.invited_count, 0) AS invited_count - , COALESCE(MS.nmemes_sent, 0) AS nmemes_sent - , COALESCE(MS.lr_smoothed, 0) AS lr_smoothed - , ( - COALESCE(MS.invited_count, 0)::float - / LN(COALESCE(MS.nmemes_sent, 0) + 2.718281828) - ) AS virality_score - - FROM meme M - INNER JOIN meme_stats MS - ON MS.meme_id = M.id - - INNER JOIN user_language L - ON L.language_code = M.language_code - AND L.user_id = :user_id - - LEFT JOIN user_meme_reaction R - ON R.meme_id = M.id - AND R.user_id = :user_id - - LEFT JOIN user_meme_source_stats UMSS - ON UMSS.meme_source_id = M.meme_source_id - AND UMSS.user_id = :user_id - - WHERE 1=1 - AND M.status = 'ok' - AND R.meme_id IS NULL - AND COALESCE(MS.invited_count, 0) > 0 - AND COALESCE(MS.nmemes_sent, 0) >= 20 - AND COALESCE(MS.lr_smoothed, 0) >= 0.05 - {exclude_meme_ids_sql_filter(exclude_meme_ids)} - {block_disliked_sources_sql_filter()} - - ORDER BY - ( - COALESCE(MS.invited_count, 0)::float - / LN(COALESCE(MS.nmemes_sent, 0) + 2.718281828) - ) * {disliked_source_demote_sql()} DESC - , MS.lr_smoothed DESC NULLS LAST - , MS.nlikes DESC NULLS LAST - LIMIT :limit - """ - return await fetch_all(text(query), _build_params(user_id, limit, exclude_meme_ids)) - - class CandidatesRetriever: """CandidatesRetriever class is used for unit testing""" @@ -784,7 +675,6 @@ class CandidatesRetriever: "es_ranked": get_es_ranked, "cold_start_explore": cold_start_explore, "cold_start_adapt": cold_start_adapt, - "viral_shares": viral_shares, } async def get_candidates( diff --git a/src/recommendations/cold_start_experiments.py b/src/recommendations/cold_start_experiments.py deleted file mode 100644 index 16ce4f30..00000000 --- a/src/recommendations/cold_start_experiments.py +++ /dev/null @@ -1,96 +0,0 @@ -"""A/B: fresh (<30d) quality memes for true-new cold start (H10).""" - -from __future__ import annotations - -import hashlib -from typing import Any - -from src.tgbot.service import ( - assign_experiment, - get_experiment_assignment, - get_experiment_variant, -) - -COLD_START_FRESH_VIRAL_EXPERIMENT_ID = "cold_start_fresh_viral_v1" -COLD_START_FRESH_VIRAL_CONTROL = "control" -COLD_START_FRESH_VIRAL_TREATMENT = "treatment_fresh_30d" -COLD_START_FRESH_MAX_AGE_DAYS = 30 -COLD_START_FRESH_ENROLL_MAX_MEMES_SENT = 6 - - -def cold_start_fresh_variant_for_user(user_id: int) -> str: - key = f"{COLD_START_FRESH_VIRAL_EXPERIMENT_ID}:{user_id}" - bucket = int.from_bytes(hashlib.sha256(key.encode("utf-8")).digest()[:8], "big") % 2 - if bucket == 0: - return COLD_START_FRESH_VIRAL_CONTROL - return COLD_START_FRESH_VIRAL_TREATMENT - - -def is_cold_start_fresh_eligible( - nmemes_sent: int, - nsessions: int, - cold_start_account_too_old: bool, -) -> bool: - return ( - nmemes_sent < COLD_START_FRESH_ENROLL_MAX_MEMES_SENT - and (nsessions or 0) <= 1 - and not cold_start_account_too_old - ) - - -def build_cold_start_fresh_assignment(user_id: int) -> tuple[str, dict[str, Any]]: - variant = cold_start_fresh_variant_for_user(user_id) - return variant, { - "assignment_strategy": "sha256(experiment_id:user_id)%2", - "max_age_days": COLD_START_FRESH_MAX_AGE_DAYS, - "primary_read": ( - "first-tap %, first-like %, reached-5; k-factor: non-self share clicks, invites" - ), - } - - -async def get_or_assign_cold_start_fresh_variant( - user_id: int, - *, - nmemes_sent: int, - nsessions: int, - cold_start_account_too_old: bool, -) -> str: - assignment = await get_experiment_assignment( - user_id, - COLD_START_FRESH_VIRAL_EXPERIMENT_ID, - ) - if assignment is not None: - return assignment["variant"] - - if not is_cold_start_fresh_eligible(nmemes_sent, nsessions, cold_start_account_too_old): - return COLD_START_FRESH_VIRAL_CONTROL - - proposed_variant, assignment_metadata = build_cold_start_fresh_assignment(user_id) - inserted = await assign_experiment( - user_id, - COLD_START_FRESH_VIRAL_EXPERIMENT_ID, - proposed_variant, - assignment_metadata, - ) - if inserted: - return proposed_variant - - return ( - await get_experiment_variant(user_id, COLD_START_FRESH_VIRAL_EXPERIMENT_ID) - or COLD_START_FRESH_VIRAL_CONTROL - ) - - -def fresh_max_age_days_for_variant( - variant: str, - *, - nmemes_sent: int, - nsessions: int, - cold_start_account_too_old: bool, -) -> int | None: - if variant != COLD_START_FRESH_VIRAL_TREATMENT: - return None - if not is_cold_start_fresh_eligible(nmemes_sent, nsessions, cold_start_account_too_old): - return None - return COLD_START_FRESH_MAX_AGE_DAYS diff --git a/src/recommendations/meme_queue.py b/src/recommendations/meme_queue.py index 99823a15..4290022d 100644 --- a/src/recommendations/meme_queue.py +++ b/src/recommendations/meme_queue.py @@ -8,19 +8,8 @@ from src.config import settings from src.database import fetch_all, fetch_one from src.recommendations.blender import blend -from src.recommendations.blender_experiments import ( - MATURE_BLENDER_CONTROL_WEIGHTS, - get_mature_blend_weights_with_experiments, - get_text_light_blender_v1_weights, - is_explore_budget_v1_treatment, -) -from src.recommendations.candidates import ( - CandidatesRetriever, -) -from src.recommendations.cold_start_experiments import ( - fresh_max_age_days_for_variant, - get_or_assign_cold_start_fresh_variant, -) +from src.recommendations.blender_experiments import is_explore_budget_v1_treatment +from src.recommendations.candidates import CandidatesRetriever from src.recommendations.pipeline import ( RecommendationBatchPipeline, RecommendationBatchRequest, @@ -36,15 +25,12 @@ "cold_start_adapt", "cold_start_explore_guarded", "cold_start_adapt_guarded", - "cold_start_explore_fresh", - "cold_start_explore_fresh_guarded", } ) COLD_START_GUARDED_RECOMMENDED_BY = frozenset( { "cold_start_explore_guarded", "cold_start_adapt_guarded", - "cold_start_explore_fresh_guarded", } ) @@ -217,12 +203,6 @@ async def clear_meme_queue_for_user(user_id: int) -> None: await redis.delete_by_key(queue_key) -def _trim_error_message(value: str, limit: int = 300) -> str: - if len(value) <= limit: - return value - return value[: limit - 3] + "..." - - async def check_queue(user_id: int) -> bool: """Refill queue if low. Returns True if lock was acquired (work done or skipped). @@ -263,29 +243,6 @@ async def check_queue(user_id: int) -> bool: return True -async def _cold_start_fresh_max_age_days( - user_id: int, - *, - nmemes_sent: int, - nsessions: int, - cold_start_account_too_old: bool, -) -> int | None: - if settings.COLD_START_FRESH_VIRAL_EXPERIMENT_ENABLED is not True: - return None - variant = await get_or_assign_cold_start_fresh_variant( - user_id, - nmemes_sent=nmemes_sent, - nsessions=nsessions, - cold_start_account_too_old=cold_start_account_too_old, - ) - return fresh_max_age_days_for_variant( - variant, - nmemes_sent=nmemes_sent, - nsessions=nsessions, - cold_start_account_too_old=cold_start_account_too_old, - ) - - async def generate_recommendations( user_id: int, limit: int, @@ -337,9 +294,6 @@ async def generate_recommendations( retriever=retriever, blend_func=blend, fetch_all_func=fetch_all, - mature_weights_func=get_mature_blend_weights_with_experiments, - text_light_weights_func=get_text_light_blender_v1_weights, - mature_control_weights=MATURE_BLENDER_CONTROL_WEIGHTS, ) result = await pipeline.run( RecommendationBatchRequest( @@ -356,22 +310,10 @@ async def generate_recommendations( cold_start_candidate_guardrails_enabled=( settings.COLD_START_CANDIDATE_GUARDRAILS_ENABLED ), - cold_start_fresh_max_age_days=await _cold_start_fresh_max_age_days( - user_id, - nmemes_sent=nmemes_sent, - nsessions=nsessions, - cold_start_account_too_old=cold_start_account_too_old, - ), - # FFM-1357: stop new exposure until this overlay has a CEO-owned - # active experiment record. Existing assignment rows remain readable. - text_light_blender_v1_enabled=False, explore_budget_enabled=( settings.EXPLORE_BUDGET_V1_ENABLED is True and await is_explore_budget_v1_treatment(user_id, nmemes_sent) ), - source_diversity_enabled=settings.RECOMMENDATION_SOURCE_DIVERSITY_ENABLED, - shadow_scoring_enabled=settings.RECOMMENDATION_SHADOW_SCORING_ENABLED, - diagnostics_sample_rate=settings.RECOMMENDATION_DIAGNOSTICS_SAMPLE_RATE, ) ) candidates = result.selected @@ -383,8 +325,7 @@ async def generate_recommendations( except Exception as error: result.diagnostics.outcome = "enqueue_failure" result.diagnostics.error_type = type(error).__name__ - result.diagnostics.error_message = _trim_error_message(str(error)) - record_recommendation_batch_diagnostics(result.diagnostics, force_full=True) + record_recommendation_batch_diagnostics(result.diagnostics) raise record_recommendation_batch_diagnostics(result.diagnostics) diff --git a/src/recommendations/pipeline.py b/src/recommendations/pipeline.py index f7c9ef98..c8d50d6b 100644 --- a/src/recommendations/pipeline.py +++ b/src/recommendations/pipeline.py @@ -1,18 +1,14 @@ from __future__ import annotations import asyncio -import json import logging -import random import uuid -from contextlib import nullcontext from dataclasses import dataclass, field from time import perf_counter from typing import Any, Awaitable, Callable, Mapping, Sequence from sqlalchemy import text -from src.config import settings from src.feed_turn.planner import ( COLD_START_1, COLD_START_2, @@ -23,24 +19,12 @@ ) from src.observability.sentry import sentry_log_extra from src.recommendations.blender import blend as default_blend -from src.recommendations.blender_experiments import ( - MATURE_BLENDER_CONTROL_WEIGHTS, - get_mature_blend_weights_with_experiments, - get_text_light_blender_v1_weights, -) from src.recommendations.candidates import CandidatesRetriever from src.recommendations.utils import ( - block_disliked_sources_sql_filter, disliked_source_demote_sql, exclude_meme_ids_sql_filter, ) -try: - import sentry_sdk -except Exception: # pragma: no cover - sentry is optional in local tooling - sentry_sdk = None - - logger = logging.getLogger(__name__) LOW_SENT_POOL_MIN_REACTIONS_FOR_QUALITY_GATE = 10 @@ -68,15 +52,9 @@ list[Candidate], ] FetchAllFunc = Callable[[Any, Mapping[str, Any] | None], Awaitable[list[Candidate]]] -MatureWeightsFunc = Callable[[int], Awaitable[Mapping[str, float]]] -TextLightWeightsFunc = Callable[[int, dict[str, float]], Awaitable[Mapping[str, float]]] LowSentFetcher = Callable[ ["RecommendationBatchRequest", int, list[int]], Awaitable[list[Candidate]] ] -ShadowScorer = Callable[ - [list[Candidate], "RecommendationBatchRequest"], - Awaitable[Mapping[str, Any]] | Mapping[str, Any], -] SwallowWarning = tuple[str, tuple[Any, ...]] @@ -93,12 +71,7 @@ class RecommendationBatchRequest: random_seed: int | None = None cold_start_nsessions_gate_enabled: bool = False cold_start_candidate_guardrails_enabled: bool = False - cold_start_fresh_max_age_days: int | None = None - text_light_blender_v1_enabled: bool = False explore_budget_enabled: bool = False - source_diversity_enabled: bool = False - shadow_scoring_enabled: bool = True - diagnostics_sample_rate: float = 0.01 @dataclass(frozen=True) @@ -106,10 +79,8 @@ class EngineRunDiagnostics: engine: str duration_ms: int candidate_count: int - candidate_ids: tuple[int, ...] = () kwargs_keys: tuple[str, ...] = () error_type: str | None = None - error_message: str | None = None def compact(self) -> dict[str, Any]: return { @@ -120,30 +91,6 @@ def compact(self) -> dict[str, Any]: "error_type": self.error_type, } - def full(self, include_candidate_ids: bool) -> dict[str, Any]: - payload = self.compact() | {"error_message": self.error_message} - if include_candidate_ids: - payload["candidate_ids"] = list(self.candidate_ids) - return payload - - -@dataclass(frozen=True) -class ShadowScoreSummary: - candidate_count: int - quality_feature_count: int - virality_feature_count: int - avg_quality_score: float | None - avg_virality_score: float | None - - def to_dict(self) -> dict[str, Any]: - return { - "candidate_count": self.candidate_count, - "quality_feature_count": self.quality_feature_count, - "virality_feature_count": self.virality_feature_count, - "avg_quality_score": self.avg_quality_score, - "avg_virality_score": self.avg_virality_score, - } - @dataclass class RecommendationBatchDiagnostics: @@ -160,8 +107,6 @@ class RecommendationBatchDiagnostics: cold_start_nsessions_gate_enabled: bool = False cold_start_candidate_guardrails_enabled: bool = False cold_start_candidate_guardrails_applied: bool = False - cold_start_fresh_applied: bool = False - diagnostics_sample_rate: float = 0.01 maturity_stage: str | None = None duration_ms: int = 0 selected_count: int = 0 @@ -169,15 +114,8 @@ class RecommendationBatchDiagnostics: low_sent_count: int = 0 fallback_used: str | None = None last_resort_used: bool = False - source_diversity_enabled: bool = False - source_diversity_reordered_count: int = 0 - shadow_scoring_enabled: bool = False - shadow_score_summary: ShadowScoreSummary | None = None - shadow_scoring_payload: Mapping[str, Any] | None = None outcome: str = "success" error_type: str | None = None - error_message: str | None = None - selected_ids: tuple[int, ...] = () engine_runs: list[EngineRunDiagnostics] = field(default_factory=list) @property @@ -211,7 +149,6 @@ def compact(self) -> dict[str, Any]: "cold_start_candidate_guardrails_applied": ( self.cold_start_candidate_guardrails_applied ), - "cold_start_fresh_applied": self.cold_start_fresh_applied, "queue_len_before": self.queue_len_before, "exclude_count": self.exclude_count, "selected_count": self.selected_count, @@ -223,42 +160,10 @@ def compact(self) -> dict[str, Any]: "low_sent_count": self.low_sent_count, "fallback_used": self.fallback_used, "last_resort_used": self.last_resort_used, - "source_diversity_enabled": self.source_diversity_enabled, - "source_diversity_reordered_count": self.source_diversity_reordered_count, - "shadow_scoring_enabled": self.shadow_scoring_enabled, - "shadow_scored_count": ( - None - if self.shadow_score_summary is None - else self.shadow_score_summary.candidate_count - ), "outcome": self.outcome, "error_type": self.error_type, } - def full(self, include_candidate_ids: bool = True) -> dict[str, Any]: - payload = self.compact() | { - "user_id": self.user_id, - "error_message": self.error_message, - "engine_runs": [ - engine_run.full(include_candidate_ids=include_candidate_ids) - for engine_run in self.engine_runs - ], - "shadow_score_summary": ( - None if self.shadow_score_summary is None else self.shadow_score_summary.to_dict() - ), - "shadow_scoring_payload": self.shadow_scoring_payload, - } - if include_candidate_ids: - payload["selected_ids"] = list(self.selected_ids) - return payload - - def should_emit_full(self, sample_rate: float) -> bool: - if self.outcome != "success" or self.engine_error_count > 0: - return True - if settings.ENVIRONMENT.is_testing: - return True - return random.random() < max(0.0, min(1.0, sample_rate)) - @dataclass(frozen=True) class RecommendationBatchResult: @@ -266,21 +171,8 @@ class RecommendationBatchResult: diagnostics: RecommendationBatchDiagnostics -def record_recommendation_batch_diagnostics( - diagnostics: RecommendationBatchDiagnostics, - *, - force_full: bool = False, -) -> None: - sample_rate = diagnostics.limit and diagnostics.diagnostics_sample_rate - include_full = force_full or diagnostics.should_emit_full(float(sample_rate or 0.0)) - compact = diagnostics.compact() - extra = sentry_log_extra({"recommendation_batch": compact}, recommendation_full=include_full) - if include_full: - extra["ff_recommendation_batch_full_json"] = json.dumps( - diagnostics.full(include_candidate_ids=True), - default=str, - ) - +def record_recommendation_batch_diagnostics(diagnostics: RecommendationBatchDiagnostics) -> None: + extra = sentry_log_extra({"recommendation_batch": diagnostics.compact()}) level = ( logging.WARNING if diagnostics.outcome != "success" or diagnostics.engine_error_count @@ -297,19 +189,11 @@ def __init__( blend_func: BlendFunc | None = None, fetch_all_func: FetchAllFunc | None = None, low_sent_fetcher: LowSentFetcher | None = None, - shadow_scorer: ShadowScorer | None = None, - mature_weights_func: MatureWeightsFunc = get_mature_blend_weights_with_experiments, - text_light_weights_func: TextLightWeightsFunc = get_text_light_blender_v1_weights, - mature_control_weights: Mapping[str, float] = MATURE_BLENDER_CONTROL_WEIGHTS, ) -> None: self.retriever = retriever or CandidatesRetriever() self.blend_func = blend_func or default_blend self.fetch_all_func = fetch_all_func self.low_sent_fetcher = low_sent_fetcher - self.shadow_scorer = shadow_scorer - self.mature_weights_func = mature_weights_func - self.text_light_weights_func = text_light_weights_func - self.mature_control_weights = dict(mature_control_weights) async def run(self, request: RecommendationBatchRequest) -> RecommendationBatchResult: diagnostics = RecommendationBatchDiagnostics( @@ -327,41 +211,20 @@ async def run(self, request: RecommendationBatchRequest) -> RecommendationBatchR cold_start_candidate_guardrails_enabled=( request.cold_start_candidate_guardrails_enabled ), - diagnostics_sample_rate=request.diagnostics_sample_rate, - source_diversity_enabled=request.source_diversity_enabled, - shadow_scoring_enabled=request.shadow_scoring_enabled, ) started_at = perf_counter() - with _start_sentry_span("recommendations.batch", "RecommendationBatchPipeline.run") as span: - try: - selected = await self._select_batch(request, diagnostics) - if request.source_diversity_enabled: - selected, reordered_count = diversify_candidates_by_source(selected) - diagnostics.source_diversity_reordered_count = reordered_count - - if request.shadow_scoring_enabled: - if self.shadow_scorer is None: - diagnostics.shadow_score_summary = summarize_shadow_scores(selected) - else: - shadow_payload = self.shadow_scorer(selected, request) - if hasattr(shadow_payload, "__await__"): - shadow_payload = await shadow_payload - diagnostics.shadow_scoring_payload = shadow_payload - - diagnostics.duration_ms = _elapsed_ms(started_at) - diagnostics.selected_count = len(selected) - diagnostics.selected_ids = _candidate_ids(selected) - _set_sentry_span_data(span, diagnostics) - return RecommendationBatchResult(selected=selected, diagnostics=diagnostics) - except Exception as error: - diagnostics.duration_ms = _elapsed_ms(started_at) - diagnostics.outcome = "failure" - diagnostics.error_type = type(error).__name__ - diagnostics.error_message = _trim(str(error)) - _set_sentry_span_data(span, diagnostics) - record_recommendation_batch_diagnostics(diagnostics, force_full=True) - raise + try: + selected = await self._select_batch(request, diagnostics) + except Exception as error: + diagnostics.duration_ms = _elapsed_ms(started_at) + diagnostics.outcome = "failure" + diagnostics.error_type = type(error).__name__ + record_recommendation_batch_diagnostics(diagnostics) + raise + diagnostics.duration_ms = _elapsed_ms(started_at) + diagnostics.selected_count = len(selected) + return RecommendationBatchResult(selected=selected, diagnostics=diagnostics) async def _select_batch( self, @@ -378,8 +241,6 @@ async def _select_batch( diagnostics, limit=request.limit, exclude_ids=list(request.meme_ids_in_queue), - use_recently_liked_blender_v2=True, - use_text_light_blender_v1=request.text_light_blender_v1_enabled, ) candidates = await self._get_low_sent_candidates( @@ -399,8 +260,6 @@ async def _select_batch( diagnostics, limit=remaining_limit, exclude_ids=exclude_ids, - use_recently_liked_blender_v2=False, - use_text_light_blender_v1=False, ) if _is_idle_return(request): # The first meme after an idle gap is a proven one, not a label. @@ -425,8 +284,6 @@ async def _select_by_plan( *, limit: int, exclude_ids: list[int], - use_recently_liked_blender_v2: bool, - use_text_light_blender_v1: bool, ) -> list[Candidate]: if limit <= 0: return [] @@ -454,12 +311,7 @@ async def _select_by_plan( return candidates return await self._run_fallbacks(plan, request, limit, exclude_ids, diagnostics) - weights = await self._blend_weights( - plan, - request.user_id, - use_recently_liked_blender_v2, - use_text_light_blender_v1, - ) + weights = dict(plan.blend_weights) if request.explore_budget_enabled and plan.maturity_stage == MATURE: weights = with_fresh_viral_weight(weights) candidates_dict = await self._fetch_candidates_dict( @@ -474,7 +326,7 @@ async def _select_by_plan( blended = self.blend_func( candidates_dict, weights, - _fixed_pos_for_weights(plan, weights), + dict(plan.fixed_pos), limit, request.random_seed, ) @@ -502,18 +354,6 @@ async def _fetch_primary_engine( diagnostics, queued_or_selected_ahead=len(exclude_ids), ) - if request.cold_start_fresh_max_age_days and engine == "cold_start_explore": - diagnostics.cold_start_fresh_applied = True - segments = [ - ( - segment_limit, - { - **kwargs, - "max_age_days": request.cold_start_fresh_max_age_days, - }, - ) - for segment_limit, kwargs in segments - ] for segment_limit, kwargs in segments: if segment_limit <= 0: @@ -554,23 +394,6 @@ async def _fetch_primary_engine( ) return selected - if ( - engine == "cold_start_explore" - and request.cold_start_fresh_max_age_days - and len(selected) < limit - ): - fill_limit = limit - len(selected) - fill = await self._fetch_engine( - engine, - request.user_id, - fill_limit, - segment_exclude_ids, - diagnostics, - ) - excluded = set(segment_exclude_ids) - fill = [candidate for candidate in fill if candidate.get("id") not in excluded] - selected.extend(fill[:fill_limit]) - return selected def _primary_engine_fetch_segments( @@ -609,25 +432,6 @@ def _primary_engine_fetch_segments( segments.append((last_position - guarded_end, {})) return segments - async def _blend_weights( - self, - plan: CandidateSelectionPlan, - user_id: int, - use_recently_liked_blender_v2: bool, - use_text_light_blender_v1: bool, - ) -> dict[str, float]: - if plan.maturity_stage != MATURE: - weights = dict(plan.blend_weights) - elif use_recently_liked_blender_v2: - weights = dict(await self.mature_weights_func(user_id)) - else: - weights = dict(self.mature_control_weights) - - if use_text_light_blender_v1 and "lr_smoothed" in weights: - weights = dict(await self.text_light_weights_func(user_id, weights)) - - return weights - async def _run_fallbacks( self, plan: CandidateSelectionPlan, @@ -703,7 +507,6 @@ async def _fetch_engine( exclude_mem_ids=exclude_ids, **kwargs, ), - span_op="recommendations.engine", kwargs_keys=tuple(sorted(kwargs.keys())), swallow_errors=swallow_errors, swallow_warning=( @@ -718,45 +521,39 @@ async def _fetch_with_diagnostics( diagnostics: RecommendationBatchDiagnostics, candidates_awaitable: Awaitable[list[Candidate]], *, - span_op: str, kwargs_keys: tuple[str, ...] = (), swallow_errors: bool = False, swallow_warning: SwallowWarning | None = None, ) -> list[Candidate]: started_at = perf_counter() - with _start_sentry_span(span_op, source_name) as span: - try: - candidates = await candidates_awaitable - except Exception as error: - duration_ms = _elapsed_ms(started_at) - _record_engine_run( - diagnostics, - source_name, - duration_ms, - [], - kwargs_keys=kwargs_keys, - error=error, - ) - _set_engine_span_data(span, source_name, duration_ms, 0, error) - if swallow_errors: - if swallow_warning is None: - logger.warning("%s failed", source_name, exc_info=True) - else: - message, args = swallow_warning - logger.warning(message, *args, exc_info=True) - return [] - raise - - duration_ms = _elapsed_ms(started_at) + try: + candidates = await candidates_awaitable + except Exception as error: _record_engine_run( diagnostics, source_name, - duration_ms, - candidates, + _elapsed_ms(started_at), + [], kwargs_keys=kwargs_keys, + error=error, ) - _set_engine_span_data(span, source_name, duration_ms, len(candidates), None) - return candidates + if swallow_errors: + if swallow_warning is None: + logger.warning("%s failed", source_name, exc_info=True) + else: + message, args = swallow_warning + logger.warning(message, *args, exc_info=True) + return [] + raise + + _record_engine_run( + diagnostics, + source_name, + _elapsed_ms(started_at), + candidates, + kwargs_keys=kwargs_keys, + ) + return candidates async def _drop_bad_fresh( self, @@ -813,7 +610,6 @@ async def _get_low_sent_candidates( limit, exclude_ids, ), - span_op="recommendations.engine", ) diagnostics.low_sent_count = len(candidates) return candidates @@ -864,35 +660,9 @@ async def _fetch_query_pool( pool_name, diagnostics, fetch_all_func(text(query), params), - span_op="recommendations.sql_pool", ) -def diversify_candidates_by_source(candidates: list[Candidate]) -> tuple[list[Candidate], int]: - if not candidates or all(_source_key(candidate) is None for candidate in candidates): - return candidates, 0 - - seen_sources: set[Any] = set() - first_pass: list[Candidate] = [] - repeated_sources: list[Candidate] = [] - for candidate in candidates: - source_key = _source_key(candidate) - if source_key is None or source_key not in seen_sources: - first_pass.append(candidate) - if source_key is not None: - seen_sources.add(source_key) - continue - repeated_sources.append(candidate) - - diversified = first_pass + repeated_sources - reordered_count = sum( - 1 - for before, after in zip(candidates, diversified) - if _candidate_id(before) != _candidate_id(after) - ) - return diversified, reordered_count - - def _is_idle_return(request: RecommendationBatchRequest) -> bool: """Empty queue == the batch's first meme is the next one shown. The Redis queue expires 1h after the last refill, so this is the post-push slot.""" @@ -907,38 +677,6 @@ def with_fresh_viral_weight(weights: Mapping[str, float]) -> dict[str, float]: return scaled -def _fixed_pos_for_weights( - plan: CandidateSelectionPlan, - weights: Mapping[str, float], -) -> dict[int, str]: - fixed_pos = dict(plan.fixed_pos) - for position, engine in list(fixed_pos.items()): - if engine == "lr_smoothed" and "text_light_lr_smoothed" in weights: - fixed_pos[position] = "text_light_lr_smoothed" - return fixed_pos - - -def summarize_shadow_scores(candidates: Sequence[Candidate]) -> ShadowScoreSummary: - quality_scores: list[float] = [] - virality_scores: list[float] = [] - for candidate in candidates: - quality_score = _quality_score(candidate) - if quality_score is not None: - quality_scores.append(quality_score) - - virality_score = _virality_score(candidate) - if virality_score is not None: - virality_scores.append(virality_score) - - return ShadowScoreSummary( - candidate_count=len(candidates), - quality_feature_count=len(quality_scores), - virality_feature_count=len(virality_scores), - avg_quality_score=_average(quality_scores), - avg_virality_score=_average(virality_scores), - ) - - def _low_sent_query(exclude_ids: list[int], explore_budget: bool = False) -> str: reactions = "(COALESCE(MS.nlikes, 0) + COALESCE(MS.ndislikes, 0))" if explore_budget: @@ -952,7 +690,6 @@ def _low_sent_query(exclude_ids: list[int], explore_budget: bool = False) -> str OR (COALESCE(MS.nlikes, 0) + 1.) / ({reactions} + 2.) >= {EXPLORE_BUDGET_LOW_SENT_MIN_SMOOTHED_LR} ) - {block_disliked_sources_sql_filter()} """ source_order = f"{disliked_source_demote_sql()} DESC," else: @@ -1056,10 +793,8 @@ def _record_engine_run( engine=engine, duration_ms=duration_ms, candidate_count=len(candidates), - candidate_ids=_candidate_ids(candidates), kwargs_keys=kwargs_keys, error_type=None if error is None else type(error).__name__, - error_message=None if error is None else _trim(str(error)), ) ) @@ -1077,104 +812,5 @@ def _candidate_id(candidate: Mapping[str, Any]) -> Any: return candidate.get("id") -def _source_key(candidate: Mapping[str, Any]) -> Any: - source_key = candidate.get("meme_source_id") - if source_key is not None: - return source_key - return candidate.get("source_id") - - -def _quality_score(candidate: Mapping[str, Any]) -> float | None: - score = _first_number(candidate, ("lr_smoothed", "engagement_score", "quality_score")) - if score is not None: - return score - - nlikes = _first_number(candidate, ("nlikes",)) - ndislikes = _first_number(candidate, ("ndislikes",)) - if nlikes is None and ndislikes is None: - return None - - likes = nlikes or 0.0 - dislikes = ndislikes or 0.0 - return (likes + 1.0) / (likes + dislikes + 2.0) - - -def _virality_score(candidate: Mapping[str, Any]) -> float | None: - return _first_number( - candidate, - ( - "invited_count", - "share_clicks", - "unique_share_clickers", - "forwards", - "nshares", - "platform_shares", - ), - ) - - -def _first_number(candidate: Mapping[str, Any], keys: tuple[str, ...]) -> float | None: - for key in keys: - value = candidate.get(key) - if isinstance(value, (int, float)): - return float(value) - return None - - -def _average(values: list[float]) -> float | None: - if not values: - return None - return round(sum(values) / len(values), 6) - - def _elapsed_ms(started_at: float) -> int: return int((perf_counter() - started_at) * 1000) - - -def _trim(value: str, limit: int = 300) -> str: - if len(value) <= limit: - return value - return value[: limit - 3] + "..." - - -def _start_sentry_span(op: str, name: str): - if sentry_sdk is None: - return nullcontext() - try: - return sentry_sdk.start_span(op=op, name=name) - except Exception: - return nullcontext() - - -def _set_sentry_span_data(span: Any, diagnostics: RecommendationBatchDiagnostics) -> None: - if span is None: - return - compact = diagnostics.compact() - try: - span.set_tag("ff.module", "recommendations") - span.set_tag("recommendation.outcome", diagnostics.outcome) - if diagnostics.maturity_stage: - span.set_tag("recommendation.maturity_stage", diagnostics.maturity_stage) - for key, value in compact.items(): - span.set_data(f"recommendation.{key}", value) - except Exception: - return - - -def _set_engine_span_data( - span: Any, - engine: str, - duration_ms: int, - candidate_count: int, - error: Exception | None, -) -> None: - if span is None: - return - try: - span.set_tag("recommendation.engine", engine) - span.set_data("recommendation.engine.duration_ms", duration_ms) - span.set_data("recommendation.engine.candidate_count", candidate_count) - if error: - span.set_data("recommendation.engine.error_type", type(error).__name__) - except Exception: - return diff --git a/src/recommendations/utils.py b/src/recommendations/utils.py index 875c3926..7a46c7f6 100644 --- a/src/recommendations/utils.py +++ b/src/recommendations/utils.py @@ -9,52 +9,6 @@ def exclude_meme_ids_sql_filter(exclude_meme_ids: list[int], meme_id_column: str return "" -def block_disliked_sources_sql_filter( - *, - enabled: bool | None = None, - min_reactions: int | None = None, - min_dislike_to_like_ratio: float | None = None, - meme_source_column: str = "M.meme_source_id", -) -> str: - """Hard-exclude sources the user clearly hates (optional, off by default). - - Skip/dislike in the feed UI is closer to TikTok "next" than "not interested". - Hard exclusion of every majority-dislike source empties pools and treats skips - as bans. Prefer :func:`disliked_source_demote_sql` for ranking demotion. - - Hard block only when ``ndislikes >= ratio * nlikes`` with enough reactions. - """ - if enabled is None: - from src.config import settings - - enabled = settings.RECOMMENDATION_BLOCK_DISLIKED_SOURCES - if not enabled: - return "" - - if min_reactions is None: - from src.config import settings - - min_reactions = settings.RECOMMENDATION_BLOCK_DISLIKED_MIN_REACTIONS - if min_dislike_to_like_ratio is None: - from src.config import settings - - min_dislike_to_like_ratio = settings.RECOMMENDATION_BLOCK_DISLIKED_RATIO - - min_reactions = max(1, int(min_reactions)) - ratio = float(min_dislike_to_like_ratio) - # ratio is config float, not user input - return f""" - AND NOT EXISTS ( - SELECT 1 - FROM user_meme_source_stats umss_block - WHERE umss_block.user_id = :user_id - AND umss_block.meme_source_id = {meme_source_column} - AND (umss_block.nlikes + umss_block.ndislikes) >= {min_reactions} - AND umss_block.ndislikes >= {ratio} * GREATEST(umss_block.nlikes, 1) - ) - """ - - def disliked_source_demote_sql( umss_alias: str = "UMSS", *, diff --git a/tests/recommendations/test_blender_experiments.py b/tests/recommendations/test_blender_experiments.py index 099cc49e..6a7b902e 100644 --- a/tests/recommendations/test_blender_experiments.py +++ b/tests/recommendations/test_blender_experiments.py @@ -6,349 +6,11 @@ EXPLORE_BUDGET_V1_CONTROL, EXPLORE_BUDGET_V1_EXPERIMENT_ID, EXPLORE_BUDGET_V1_TREATMENT, - MATURE_BLENDER_CONTROL_WEIGHTS, - MATURE_BLENDER_TREATMENT_WEIGHTS, - RECENTLY_LIKED_BLENDER_V2_CONTROL, - RECENTLY_LIKED_BLENDER_V2_ENROLLMENT_FROZEN_AT, - RECENTLY_LIKED_BLENDER_V2_EXCLUDED, - RECENTLY_LIKED_BLENDER_V2_EXPERIMENT_ID, - RECENTLY_LIKED_BLENDER_V2_SAMPLE_GATE_PER_VARIANT, - RECENTLY_LIKED_BLENDER_V2_TREATMENT, - TEXT_LIGHT_BLENDER_V1_CONTROL, - TEXT_LIGHT_BLENDER_V1_EXPERIMENT_ID, - TEXT_LIGHT_BLENDER_V1_MAX_OCR_WORDS, - TEXT_LIGHT_BLENDER_V1_TREATMENT, - _lr_quartile_from_boundaries, - build_recently_liked_blender_v2_assignment, - build_text_light_blender_v1_assignment, explore_budget_v1_variant_for_user, - get_mature_blend_weights_with_experiments, - get_or_assign_recently_liked_blender_v2_variant, - get_or_assign_text_light_blender_v1_variant, - get_recent_7d_lr_assignment_metrics, - get_recent_7d_lr_quartile_boundaries, - get_recently_liked_blender_v2_weights, - get_text_light_blender_v1_weights, is_explore_budget_v1_treatment, ) -def test_assignment_excludes_high_volume_low_lr_skipper(): - variant, metadata = build_recently_liked_blender_v2_assignment( - 101, - { - "likes_7d": 9, - "reactions_7d": 51, - "lr_7d": 9 / 51, - "lr_quartile": 1, - }, - ) - - assert variant == RECENTLY_LIKED_BLENDER_V2_EXCLUDED - assert metadata["high_volume_skipper"] is True - assert metadata["excluded_reason"] == "lr_7d_below_20pct_and_reactions_7d_above_50" - assert metadata["assigned_weights"] == MATURE_BLENDER_CONTROL_WEIGHTS - - -def test_assignment_persists_stratification_metadata_for_eligible_user(): - variant, metadata = build_recently_liked_blender_v2_assignment( - 202, - { - "likes_7d": 20, - "reactions_7d": 60, - "lr_7d": 1 / 3, - "lr_quartile": 2, - }, - ) - - assert variant in {RECENTLY_LIKED_BLENDER_V2_CONTROL, RECENTLY_LIKED_BLENDER_V2_TREATMENT} - assert metadata["lr_quartile"] == 2 - assert metadata["reactions_7d"] == 60 - assert metadata["high_volume_skipper"] is False - assert metadata["sample_gate_per_variant"] == RECENTLY_LIKED_BLENDER_V2_SAMPLE_GATE_PER_VARIANT - assert metadata["day3_guardrail"]["primary_read_rule"] == "sample_gate_per_variant" - - -def test_assignment_is_stable_within_lr_quartile(): - metrics = { - "likes_7d": 20, - "reactions_7d": 60, - "lr_7d": 1 / 3, - "lr_quartile": 3, - } - - first, first_metadata = build_recently_liked_blender_v2_assignment(303, metrics) - second, second_metadata = build_recently_liked_blender_v2_assignment(303, metrics) - - assert first == second - assert first_metadata == second_metadata - - -def test_text_light_assignment_replaces_lr_engine_only_for_treatment(): - base_weights = { - "best_uploaded_memes": 0.1, - "lr_smoothed": 0.3, - "recently_liked": 0.2, - } - - variant, metadata = build_text_light_blender_v1_assignment(404, base_weights) - - assert variant in {TEXT_LIGHT_BLENDER_V1_CONTROL, TEXT_LIGHT_BLENDER_V1_TREATMENT} - assert metadata["max_ocr_words"] == TEXT_LIGHT_BLENDER_V1_MAX_OCR_WORDS - assert metadata["filtered_engine"] == "text_light_lr_smoothed" - if variant == TEXT_LIGHT_BLENDER_V1_TREATMENT: - assert "lr_smoothed" not in metadata["assigned_weights"] - assert metadata["assigned_weights"]["text_light_lr_smoothed"] == 0.3 - else: - assert metadata["assigned_weights"] == base_weights - - -@pytest.mark.asyncio -async def test_mature_weights_never_enroll_viral_shares_after_kill(): - with patch( - "src.recommendations.blender_experiments.assign_experiment", - new_callable=AsyncMock, - ) as assign: - weights = await get_mature_blend_weights_with_experiments(909) - - assert weights == MATURE_BLENDER_CONTROL_WEIGHTS - assert "viral_shares" not in weights - assign.assert_not_awaited() - - -def test_lr_quartile_uses_cached_boundaries(): - boundaries = (0.2, 0.5, 0.8) - - assert _lr_quartile_from_boundaries(0.2, boundaries) == 1 - assert _lr_quartile_from_boundaries(0.21, boundaries) == 2 - assert _lr_quartile_from_boundaries(0.51, boundaries) == 3 - assert _lr_quartile_from_boundaries(0.81, boundaries) == 4 - - -@pytest.mark.asyncio -async def test_lr_assignment_metrics_only_fetch_current_user_reactions(): - with ( - patch( - "src.recommendations.blender_experiments.get_recent_7d_lr_quartile_boundaries", - new_callable=AsyncMock, - return_value=(0.2, 0.5, 0.8), - ) as get_boundaries, - patch( - "src.recommendations.blender_experiments.fetch_one", - new_callable=AsyncMock, - return_value={"likes_7d": 35, "reactions_7d": 100, "lr_7d": 0.35}, - ) as fetch_metrics, - ): - metrics = await get_recent_7d_lr_assignment_metrics(101) - - get_boundaries.assert_awaited_once() - fetch_metrics.assert_awaited_once() - query_text = str(fetch_metrics.await_args.args[0]) - assert "NTILE" not in query_text - assert "GROUP BY umr.user_id" not in query_text - assert metrics == { - "likes_7d": 35, - "reactions_7d": 100, - "lr_7d": 0.35, - "lr_quartile": 2, - "lr_quartile_boundaries": [0.2, 0.5, 0.8], - } - - -@pytest.mark.asyncio -async def test_lr_quartile_boundaries_do_not_recompute_without_cache_lock(): - with ( - patch( - "src.recommendations.blender_experiments._get_cached_lr_quartile_boundaries", - new_callable=AsyncMock, - return_value=None, - ), - patch( - "src.recommendations.blender_experiments.redis_client.set", - new_callable=AsyncMock, - return_value=None, - ), - patch( - "src.recommendations.blender_experiments.sleep", - new_callable=AsyncMock, - ), - patch( - "src.recommendations.blender_experiments._calculate_recent_7d_lr_quartile_boundaries", - new_callable=AsyncMock, - ) as calculate_boundaries, - ): - boundaries = await get_recent_7d_lr_quartile_boundaries() - - assert boundaries is None - calculate_boundaries.assert_not_awaited() - - -@pytest.mark.asyncio -async def test_existing_recently_liked_blender_v2_assignment_wins(): - with patch( - "src.recommendations.blender_experiments.get_experiment_assignment", - new_callable=AsyncMock, - return_value={"variant": RECENTLY_LIKED_BLENDER_V2_TREATMENT}, - ) as get_assignment: - variant = await get_or_assign_recently_liked_blender_v2_variant(101) - - assert variant == RECENTLY_LIKED_BLENDER_V2_TREATMENT - get_assignment.assert_awaited_once_with(101, RECENTLY_LIKED_BLENDER_V2_EXPERIMENT_ID) - - -@pytest.mark.asyncio -async def test_frozen_recently_liked_blender_v2_does_not_assign_new_user(): - with ( - patch( - "src.recommendations.blender_experiments.get_experiment_assignment", - new_callable=AsyncMock, - return_value=None, - ) as get_assignment, - patch( - "src.recommendations.blender_experiments.get_recent_7d_lr_assignment_metrics", - new_callable=AsyncMock, - ) as get_metrics, - patch( - "src.recommendations.blender_experiments.assign_experiment", - new_callable=AsyncMock, - ) as assign, - patch( - "src.recommendations.blender_experiments.get_experiment_variant", - new_callable=AsyncMock, - ) as get_variant, - ): - variant = await get_or_assign_recently_liked_blender_v2_variant(202) - - assert variant == RECENTLY_LIKED_BLENDER_V2_CONTROL - assert RECENTLY_LIKED_BLENDER_V2_ENROLLMENT_FROZEN_AT == "2026-07-23T18:05:54Z" - get_assignment.assert_awaited_once_with(202, RECENTLY_LIKED_BLENDER_V2_EXPERIMENT_ID) - get_metrics.assert_not_awaited() - assign.assert_not_awaited() - get_variant.assert_not_awaited() - - -@pytest.mark.asyncio -async def test_recently_liked_blender_v2_race_rereads_winning_assignment(): - with ( - patch( - "src.recommendations.blender_experiments.get_experiment_assignment", - new_callable=AsyncMock, - return_value=None, - ), - patch( - "src.recommendations.blender_experiments.RECENTLY_LIKED_BLENDER_V2_ENROLLMENT_FROZEN", - False, - ), - patch( - "src.recommendations.blender_experiments.get_recent_7d_lr_assignment_metrics", - new_callable=AsyncMock, - return_value={ - "likes_7d": 20, - "reactions_7d": 60, - "lr_7d": 1 / 3, - "lr_quartile": 2, - }, - ), - patch( - "src.recommendations.blender_experiments.assign_experiment", - new_callable=AsyncMock, - return_value=False, - ), - patch( - "src.recommendations.blender_experiments.get_experiment_variant", - new_callable=AsyncMock, - return_value=RECENTLY_LIKED_BLENDER_V2_CONTROL, - ), - ): - variant = await get_or_assign_recently_liked_blender_v2_variant(202) - - assert variant == RECENTLY_LIKED_BLENDER_V2_CONTROL - - -@pytest.mark.asyncio -async def test_recently_liked_blender_v2_defers_assignment_without_real_boundaries(): - with ( - patch( - "src.recommendations.blender_experiments.get_experiment_assignment", - new_callable=AsyncMock, - return_value=None, - ), - patch( - "src.recommendations.blender_experiments.RECENTLY_LIKED_BLENDER_V2_ENROLLMENT_FROZEN", - False, - ), - patch( - "src.recommendations.blender_experiments.get_recent_7d_lr_assignment_metrics", - new_callable=AsyncMock, - return_value=None, - ), - patch( - "src.recommendations.blender_experiments.assign_experiment", - new_callable=AsyncMock, - ) as assign, - patch( - "src.recommendations.blender_experiments.get_experiment_variant", - new_callable=AsyncMock, - ) as get_variant, - ): - variant = await get_or_assign_recently_liked_blender_v2_variant(202) - - assert variant == RECENTLY_LIKED_BLENDER_V2_CONTROL - assign.assert_not_awaited() - get_variant.assert_not_awaited() - - -@pytest.mark.asyncio -async def test_recently_liked_blender_v2_weights_ship_default_after_closeout(): - # Experiment closed: no assignment path; always return shipped mature default. - weights = await get_recently_liked_blender_v2_weights(100) - - assert weights == MATURE_BLENDER_CONTROL_WEIGHTS - assert weights["recently_liked"] == 0.3 - assert weights["lr_smoothed"] == 0.35 - - -@pytest.mark.asyncio -async def test_existing_text_light_blender_v1_assignment_wins(): - with patch( - "src.recommendations.blender_experiments.get_experiment_assignment", - new_callable=AsyncMock, - return_value={"variant": TEXT_LIGHT_BLENDER_V1_TREATMENT}, - ) as get_assignment: - variant = await get_or_assign_text_light_blender_v1_variant( - 101, - MATURE_BLENDER_CONTROL_WEIGHTS, - ) - - assert variant == TEXT_LIGHT_BLENDER_V1_TREATMENT - get_assignment.assert_awaited_once_with(101, TEXT_LIGHT_BLENDER_V1_EXPERIMENT_ID) - - -@pytest.mark.asyncio -async def test_text_light_blender_v1_treatment_replaces_lr_weight(): - with patch( - "src.recommendations.blender_experiments.get_or_assign_text_light_blender_v1_variant", - new_callable=AsyncMock, - return_value=TEXT_LIGHT_BLENDER_V1_TREATMENT, - ): - weights = await get_text_light_blender_v1_weights(100, MATURE_BLENDER_CONTROL_WEIGHTS) - - assert "lr_smoothed" not in weights - assert weights["text_light_lr_smoothed"] == MATURE_BLENDER_CONTROL_WEIGHTS["lr_smoothed"] - - -@pytest.mark.asyncio -async def test_text_light_blender_v1_control_keeps_base_weights(): - with patch( - "src.recommendations.blender_experiments.get_or_assign_text_light_blender_v1_variant", - new_callable=AsyncMock, - return_value=TEXT_LIGHT_BLENDER_V1_CONTROL, - ): - weights = await get_text_light_blender_v1_weights(100, MATURE_BLENDER_TREATMENT_WEIGHTS) - - assert weights == MATURE_BLENDER_TREATMENT_WEIGHTS - - def test_explore_budget_v1_split_is_deterministic_and_balanced(): variants = [explore_budget_v1_variant_for_user(user_id) for user_id in range(2000)] diff --git a/tests/recommendations/test_block_disliked_sources.py b/tests/recommendations/test_block_disliked_sources.py deleted file mode 100644 index 0604e68f..00000000 --- a/tests/recommendations/test_block_disliked_sources.py +++ /dev/null @@ -1,140 +0,0 @@ -"""Source affinity policies: demote (default) vs hard-block (optional).""" - -import pytest -import pytest_asyncio -from tests.factories import ( - cleanup_test_data, - create_meme, - create_meme_source, - create_meme_stats, - create_user, - create_user_language, - create_user_meme_source_stats, -) - -from src.database import engine -from src.recommendations.candidates import CandidatesRetriever -from src.recommendations.utils import ( - block_disliked_sources_sql_filter, - disliked_source_demote_sql, -) - -USER_ID = 10001 -SOURCE_GOOD = 10001 -SOURCE_BAD = 10002 - - -@pytest_asyncio.fixture() -async def disliked_source_data(): - async with engine.connect() as conn: - await create_user(conn, id=USER_ID) - await create_user_language(conn, user_id=USER_ID, language_code="ru") - await create_meme_source(conn, id=SOURCE_GOOD, type="telegram", url="https://t.me/good") - await create_meme_source(conn, id=SOURCE_BAD, type="telegram", url="https://t.me/bad") - - await create_user_meme_source_stats( - conn, user_id=USER_ID, meme_source_id=SOURCE_GOOD, nlikes=10, ndislikes=1 - ) - # Soft majority dislike (demote), not 3x hate (hard block) - await create_user_meme_source_stats( - conn, user_id=USER_ID, meme_source_id=SOURCE_BAD, nlikes=2, ndislikes=8 - ) - - for mid in range(10001, 10006): - await create_meme(conn, id=mid, meme_source_id=SOURCE_GOOD) - await create_meme_stats( - conn, - meme_id=mid, - nlikes=20, - ndislikes=5, - nmemes_sent=40, - lr_smoothed=0.2, - engagement_score=0.15, - raw_impr_rank=0, - invited_count=2, - ) - for mid in range(10011, 10016): - await create_meme(conn, id=mid, meme_source_id=SOURCE_BAD) - await create_meme_stats( - conn, - meme_id=mid, - nlikes=20, - ndislikes=5, - nmemes_sent=40, - lr_smoothed=0.2, - engagement_score=0.15, - raw_impr_rank=0, - invited_count=2, - ) - await conn.commit() - - yield - - async with engine.connect() as conn: - await cleanup_test_data(conn) - - -def test_hard_block_off_by_default_string(): - # Explicit disabled - assert block_disliked_sources_sql_filter(enabled=False) == "" - - -def test_hard_block_requires_ratio_when_enabled(): - frag = block_disliked_sources_sql_filter( - enabled=True, min_reactions=15, min_dislike_to_like_ratio=3.0 - ) - assert "NOT EXISTS" in frag - assert ">= 15" in frag - assert "3.0" in frag or "3." in frag - - -def test_demote_sql_returns_case_expression(): - frag = disliked_source_demote_sql(enabled=True, min_reactions=5, multiplier=0.15) - assert "CASE" in frag - assert "0.15" in frag - - -@pytest.mark.asyncio -async def test_hard_block_disabled_still_returns_bad_source_candidates( - disliked_source_data, monkeypatch -): - """Default hard-block OFF: majority-dislike sources remain eligible (demoted only).""" - monkeypatch.setattr("src.config.settings.RECOMMENDATION_BLOCK_DISLIKED_SOURCES", False) - monkeypatch.setattr("src.config.settings.RECOMMENDATION_DEMOTE_DISLIKED_SOURCES", True) - - retriever = CandidatesRetriever() - results = await retriever.get_candidates("lr_smoothed", USER_ID, limit=50) - ids = {r["id"] for r in results} - # Good sources preferred but bad may still appear with soft demote if limit large - assert ids & set(range(10001, 10006)), "expected good-source memes" - - -@pytest.mark.asyncio -async def test_hard_block_strong_hate_excludes_source(disliked_source_data, monkeypatch): - monkeypatch.setattr("src.config.settings.RECOMMENDATION_BLOCK_DISLIKED_SOURCES", True) - # Soft majority 8:2 is not 3x — should NOT hard-block - monkeypatch.setattr("src.config.settings.RECOMMENDATION_BLOCK_DISLIKED_MIN_REACTIONS", 5) - monkeypatch.setattr("src.config.settings.RECOMMENDATION_BLOCK_DISLIKED_RATIO", 3.0) - - retriever = CandidatesRetriever() - results = await retriever.get_candidates("lr_smoothed", USER_ID, limit=50) - ids = {r["id"] for r in results} - # 8 dislikes vs 2 likes = 4x >= 3x and n=10 — wait 8 >= 3*2=6, yes hard block! - # 8 >= 6 and n=10 >= 5 → hard blocked with ratio 3 - # Actually nlikes=2 ndislikes=8, 8 >= 3*2 = 6, yes blocked - bad = ids & set(range(10011, 10016)) - assert not bad, f"strong-ratio hard block should exclude: {bad}" - - -@pytest.mark.asyncio -async def test_hard_block_high_ratio_only(disliked_source_data, monkeypatch): - """With ratio 5, 8:2 source is not hard-blocked (8 < 5*2=10).""" - monkeypatch.setattr("src.config.settings.RECOMMENDATION_BLOCK_DISLIKED_SOURCES", True) - monkeypatch.setattr("src.config.settings.RECOMMENDATION_BLOCK_DISLIKED_MIN_REACTIONS", 5) - monkeypatch.setattr("src.config.settings.RECOMMENDATION_BLOCK_DISLIKED_RATIO", 5.0) - - retriever = CandidatesRetriever() - results = await retriever.get_candidates("lr_smoothed", USER_ID, limit=50) - ids = {r["id"] for r in results} - # May include bad-source under soft demote only - assert ids & set(range(10001, 10006)) diff --git a/tests/recommendations/test_broadcast_pick.py b/tests/recommendations/test_broadcast_pick.py index bd84a555..b1427bad 100644 --- a/tests/recommendations/test_broadcast_pick.py +++ b/tests/recommendations/test_broadcast_pick.py @@ -5,7 +5,6 @@ import pytest from src.recommendations.broadcast_pick import ( - BROADCAST_CHANNEL_VIRAL_RECOMMENDED_BY, BROADCAST_HQ_RECOMMENDED_BY, BROADCAST_RECOMMENDED_BY, pick_reengagement_meme, @@ -14,16 +13,8 @@ from src.storage.schemas import MemeData -def _disable_channel_viral(monkeypatch) -> None: - monkeypatch.setattr( - "src.recommendations.broadcast_pick.settings.BROADCAST_CHANNEL_VIRAL_PICK_ENABLED", - False, - ) - - @pytest.mark.asyncio async def test_hq_pick_uses_sql_row_and_hq_label(monkeypatch): - _disable_channel_viral(monkeypatch) monkeypatch.setattr( "src.recommendations.broadcast_pick.settings.BROADCAST_HIGH_QUALITY_PICK_ENABLED", True, @@ -51,7 +42,6 @@ async def test_hq_pick_uses_sql_row_and_hq_label(monkeypatch): @pytest.mark.asyncio async def test_hq_empty_falls_back_to_queue(monkeypatch): - _disable_channel_viral(monkeypatch) monkeypatch.setattr( "src.recommendations.broadcast_pick.settings.BROADCAST_HIGH_QUALITY_PICK_ENABLED", True, @@ -89,7 +79,6 @@ async def test_hq_empty_falls_back_to_queue(monkeypatch): @pytest.mark.asyncio async def test_hq_disabled_uses_queue_only(monkeypatch): - _disable_channel_viral(monkeypatch) monkeypatch.setattr( "src.recommendations.broadcast_pick.settings.BROADCAST_HIGH_QUALITY_PICK_ENABLED", False, @@ -121,64 +110,3 @@ async def test_hq_disabled_uses_queue_only(monkeypatch): fetch_one.assert_not_called() assert meme is queue_meme assert label == BROADCAST_RECOMMENDED_BY - - -@pytest.mark.asyncio -async def test_channel_viral_pick_wins_over_hq(monkeypatch): - monkeypatch.setattr( - "src.recommendations.broadcast_pick.settings.BROADCAST_CHANNEL_VIRAL_PICK_ENABLED", - True, - ) - monkeypatch.setattr( - "src.recommendations.broadcast_pick.settings.BROADCAST_HIGH_QUALITY_PICK_ENABLED", - True, - ) - row = { - "id": 10084067, - "type": MemeType.IMAGE, - "telegram_file_id": "viral-file", - "caption": None, - "nlikes": 12, - } - with patch( - "src.recommendations.broadcast_pick.fetch_one", - new_callable=AsyncMock, - return_value=row, - ) as fetch_one: - meme, label = await pick_reengagement_meme(7) - - fetch_one.assert_awaited_once() - assert label == BROADCAST_CHANNEL_VIRAL_RECOMMENDED_BY - assert meme is not None - assert meme.id == 10084067 - assert meme.recommended_by == BROADCAST_CHANNEL_VIRAL_RECOMMENDED_BY - - -@pytest.mark.asyncio -async def test_channel_viral_empty_falls_back_to_hq(monkeypatch): - monkeypatch.setattr( - "src.recommendations.broadcast_pick.settings.BROADCAST_CHANNEL_VIRAL_PICK_ENABLED", - True, - ) - monkeypatch.setattr( - "src.recommendations.broadcast_pick.settings.BROADCAST_HIGH_QUALITY_PICK_ENABLED", - True, - ) - hq_row = { - "id": 42, - "type": MemeType.IMAGE, - "telegram_file_id": "file-42", - "caption": "hi", - "nlikes": 99, - } - with patch( - "src.recommendations.broadcast_pick.fetch_one", - new_callable=AsyncMock, - side_effect=[None, hq_row], - ) as fetch_one: - meme, label = await pick_reengagement_meme(7) - - assert fetch_one.await_count == 2 - assert label == BROADCAST_HQ_RECOMMENDED_BY - assert meme is not None - assert meme.id == 42 diff --git a/tests/recommendations/test_cold_start_fresh.py b/tests/recommendations/test_cold_start_fresh.py deleted file mode 100644 index b5416dd9..00000000 --- a/tests/recommendations/test_cold_start_fresh.py +++ /dev/null @@ -1,176 +0,0 @@ -from unittest.mock import AsyncMock, patch - -import pytest - -from src.recommendations.candidates import ( - COLD_START_EXPLORE_FRESH_GUARDED_RECOMMENDED_BY, - COLD_START_EXPLORE_FRESH_RECOMMENDED_BY, - cold_start_explore, -) -from src.recommendations.cold_start_experiments import ( - COLD_START_FRESH_MAX_AGE_DAYS, - COLD_START_FRESH_VIRAL_CONTROL, - COLD_START_FRESH_VIRAL_TREATMENT, - cold_start_fresh_variant_for_user, - fresh_max_age_days_for_variant, - is_cold_start_fresh_eligible, -) -from src.recommendations.pipeline import ( - RecommendationBatchPipeline, - RecommendationBatchRequest, -) - - -def _fetch_call_sql_and_params(fetch_all: AsyncMock) -> tuple[str, dict]: - query, params = fetch_all.call_args.args - return str(query), params - - -def test_fresh_variant_is_stable_split(): - control_ids = [ - user_id - for user_id in range(200) - if cold_start_fresh_variant_for_user(user_id) == COLD_START_FRESH_VIRAL_CONTROL - ] - treatment_ids = [ - user_id - for user_id in range(200) - if cold_start_fresh_variant_for_user(user_id) == COLD_START_FRESH_VIRAL_TREATMENT - ] - assert len(control_ids) > 40 - assert len(treatment_ids) > 40 - assert cold_start_fresh_variant_for_user(control_ids[0]) == COLD_START_FRESH_VIRAL_CONTROL - assert cold_start_fresh_variant_for_user(treatment_ids[0]) == COLD_START_FRESH_VIRAL_TREATMENT - - -def test_fresh_eligibility_and_max_age(): - assert is_cold_start_fresh_eligible(0, 1, False) is True - assert is_cold_start_fresh_eligible(6, 1, False) is False - assert is_cold_start_fresh_eligible(0, 2, False) is False - assert is_cold_start_fresh_eligible(0, 1, True) is False - assert ( - fresh_max_age_days_for_variant( - COLD_START_FRESH_VIRAL_TREATMENT, - nmemes_sent=0, - nsessions=1, - cold_start_account_too_old=False, - ) - == COLD_START_FRESH_MAX_AGE_DAYS - ) - assert ( - fresh_max_age_days_for_variant( - COLD_START_FRESH_VIRAL_CONTROL, - nmemes_sent=0, - nsessions=1, - cold_start_account_too_old=False, - ) - is None - ) - - -@pytest.mark.asyncio -async def test_cold_start_explore_fresh_filters_age_and_ranks_newest(): - with patch( - "src.recommendations.candidates.fetch_all", - new_callable=AsyncMock, - return_value=[], - ) as fetch_all: - await cold_start_explore(123, limit=5, max_age_days=30) - - query_sql, params = _fetch_call_sql_and_params(fetch_all) - assert "cold_start_max_age_days" in query_sql - assert params["cold_start_max_age_days"] == 30 - assert params["recommended_by"] == COLD_START_EXPLORE_FRESH_RECOMMENDED_BY - assert "M.created_at DESC" in query_sql - - -@pytest.mark.asyncio -async def test_cold_start_explore_fresh_guarded_label(): - with patch( - "src.recommendations.candidates.fetch_all", - new_callable=AsyncMock, - return_value=[], - ) as fetch_all: - await cold_start_explore( - 123, - limit=5, - max_age_days=30, - candidate_guardrails_enabled=True, - ) - - _, params = _fetch_call_sql_and_params(fetch_all) - assert params["recommended_by"] == COLD_START_EXPLORE_FRESH_GUARDED_RECOMMENDED_BY - - -@pytest.mark.asyncio -async def test_generate_recommendations_passes_fresh_max_age_for_treatment(): - from src.recommendations.meme_queue import generate_recommendations - - retriever = AsyncMock() - retriever.get_candidates = AsyncMock(return_value=[]) - retriever.get_candidates_dict = AsyncMock(return_value={}) - - with ( - patch( - "src.recommendations.meme_queue.get_user_info", - new_callable=AsyncMock, - return_value={"nmemes_sent": 0, "nsessions": 1, "type": "user"}, - ), - patch( - "src.recommendations.meme_queue.redis.get_all_memes_in_queue_by_key", - new_callable=AsyncMock, - return_value=[], - ), - patch( - "src.recommendations.meme_queue.get_or_assign_cold_start_fresh_variant", - new_callable=AsyncMock, - return_value=COLD_START_FRESH_VIRAL_TREATMENT, - ), - patch("src.recommendations.meme_queue.settings") as settings, - patch( - "src.recommendations.meme_queue.RecommendationBatchPipeline.run", - new_callable=AsyncMock, - ) as run, - ): - settings.COLD_START_FRESH_VIRAL_EXPERIMENT_ENABLED = True - settings.COLD_START_NSESSIONS_GATE_ENABLED = False - settings.COLD_START_CANDIDATE_GUARDRAILS_ENABLED = False - settings.RECOMMENDATION_SOURCE_DIVERSITY_ENABLED = False - settings.RECOMMENDATION_SHADOW_SCORING_ENABLED = False - settings.RECOMMENDATION_DIAGNOSTICS_SAMPLE_RATE = 0.0 - run.return_value.selected = [] - await generate_recommendations(7, limit=5, nmemes_sent=0, retriever=retriever) - - request = run.await_args.args[0] - assert request.cold_start_fresh_max_age_days == COLD_START_FRESH_MAX_AGE_DAYS - - -@pytest.mark.asyncio -async def test_pipeline_prefers_fresh_then_fills_classic_cs(): - retriever = AsyncMock() - - async def get_candidates(engine, user_id, limit, exclude_mem_ids=None, **kwargs): - if kwargs.get("max_age_days"): - return [{"id": 1, "recommended_by": "cold_start_explore_fresh"}] - return [ - {"id": 1, "recommended_by": "cold_start_explore"}, - {"id": 2, "recommended_by": "cold_start_explore"}, - {"id": 3, "recommended_by": "cold_start_explore"}, - ] - - retriever.get_candidates.side_effect = get_candidates - pipeline = RecommendationBatchPipeline(retriever=retriever) - result = await pipeline.run( - RecommendationBatchRequest( - user_id=7, - limit=3, - nmemes_sent=0, - nsessions=1, - cold_start_fresh_max_age_days=30, - ) - ) - - assert [row["id"] for row in result.selected] == [1, 2, 3] - assert result.selected[0]["recommended_by"] == "cold_start_explore_fresh" - assert result.diagnostics.cold_start_fresh_applied is True - assert retriever.get_candidates.await_count >= 2 diff --git a/tests/recommendations/test_engine_contracts.py b/tests/recommendations/test_engine_contracts.py index 09105ef0..e49a66a1 100644 --- a/tests/recommendations/test_engine_contracts.py +++ b/tests/recommendations/test_engine_contracts.py @@ -108,7 +108,6 @@ async def base_data(): "text_light_lr_smoothed", "like_spread_and_recent_memes", "es_ranked", - "viral_shares", ] # Engines that require user_meme_source_stats via INNER JOIN diff --git a/tests/recommendations/test_meme_queue.py b/tests/recommendations/test_meme_queue.py index 5c5df500..f52339d1 100644 --- a/tests/recommendations/test_meme_queue.py +++ b/tests/recommendations/test_meme_queue.py @@ -3,10 +3,7 @@ import pytest -from src.recommendations.blender_experiments import ( - MATURE_BLENDER_CONTROL_WEIGHTS, - MATURE_BLENDER_TREATMENT_WEIGHTS, -) +from src.feed_turn.planner import MATURE_BLEND_WEIGHTS from src.recommendations.candidates import CandidatesRetriever from src.recommendations.meme_queue import ( _cold_start_allowed_by_realtime_state, @@ -26,16 +23,6 @@ def _patch_user_info(nsessions: int = 0, nmemes_sent: int = 0, **extra): ) -@pytest.fixture(autouse=True) -def disable_cold_start_fresh_experiment(): - with patch( - "src.recommendations.meme_queue._cold_start_fresh_max_age_days", - new_callable=AsyncMock, - return_value=None, - ): - yield - - @pytest.fixture(autouse=True) def mock_redis(): """Mock Redis and user_info calls — these tests validate blending logic, not Redis.""" @@ -55,11 +42,6 @@ def mock_redis(): "src.recommendations.meme_queue.redis.add_memes_to_queue_by_key", new_callable=AsyncMock, ), - patch( - "src.recommendations.meme_queue.get_text_light_blender_v1_weights", - new_callable=AsyncMock, - side_effect=lambda user_id, weights: weights, - ), patch( "src.recommendations.meme_queue._get_realtime_cold_start_routing_state", new_callable=AsyncMock, @@ -333,9 +315,6 @@ class TestRetriever(CandidatesRetriever): with patch("src.recommendations.meme_queue.settings") as settings: settings.COLD_START_NSESSIONS_GATE_ENABLED = False settings.COLD_START_CANDIDATE_GUARDRAILS_ENABLED = False - settings.RECOMMENDATION_SOURCE_DIVERSITY_ENABLED = False - settings.RECOMMENDATION_SHADOW_SCORING_ENABLED = False - settings.RECOMMENDATION_DIAGNOSTICS_SAMPLE_RATE = 0.0 candidates = await generate_recommendations( TEST_USER_ID, 10, nmemes_sent=3, retriever=TestRetriever() ) @@ -576,11 +555,6 @@ def capture_blend(candidates_dict, weights_dict, fixed_pos=None, limit=0, random with ( patch("src.recommendations.meme_queue.blend", side_effect=capture_blend), - patch( - "src.recommendations.meme_queue.get_mature_blend_weights_with_experiments", - new_callable=AsyncMock, - return_value=MATURE_BLENDER_CONTROL_WEIGHTS, - ) as get_weights, ): candidates = await generate_recommendations( TEST_USER_ID, 10, 200, TestRetriever(), random_seed=102 @@ -589,134 +563,11 @@ def capture_blend(candidates_dict, weights_dict, fixed_pos=None, limit=0, random assert len(candidates) == 10 # lr_smoothed is pinned at position 0 assert candidates[0]["id"] in [7, 8, 9, 10] - assert captured_weights == MATURE_BLENDER_CONTROL_WEIGHTS - get_weights.assert_awaited_once_with(TEST_USER_ID) - - -@pytest.mark.asyncio -async def test_generate_above_100_treatment_uses_recently_liked_weights(): - captured_weights = None - - async def one_candidate(self, user_id, limit=10, exclude_meme_ids=[], **kw): - return [{"id": 4, "recommended_by": "recently_liked"}] - - class TestRetriever(CandidatesRetriever): - engine_map = { - "best_uploaded_memes": one_candidate, - "like_spread_and_recent_memes": one_candidate, - "lr_smoothed": one_candidate, - "recently_liked": one_candidate, - "goat": one_candidate, - "es_ranked": one_candidate, - } - - def capture_blend(candidates_dict, weights_dict, fixed_pos=None, limit=0, random_seed=None): - nonlocal captured_weights - captured_weights = weights_dict - return [{"id": 4, "recommended_by": "recently_liked"}] - - with ( - patch( - "src.recommendations.meme_queue.get_mature_blend_weights_with_experiments", - new_callable=AsyncMock, - return_value=MATURE_BLENDER_TREATMENT_WEIGHTS, - ) as get_weights, - patch("src.recommendations.meme_queue.blend", side_effect=capture_blend), - ): - candidates = await generate_recommendations( - TEST_USER_ID, 10, 200, TestRetriever(), random_seed=102 - ) - - assert candidates == [{"id": 4, "recommended_by": "recently_liked"}] - # After v2 closeout, treatment weights == shipped control default. - assert captured_weights == MATURE_BLENDER_TREATMENT_WEIGHTS - assert captured_weights == MATURE_BLENDER_CONTROL_WEIGHTS - assert captured_weights["recently_liked"] == 0.3 - get_weights.assert_awaited_once_with(TEST_USER_ID) - - -@pytest.mark.asyncio -async def test_text_light_blender_v1_is_disabled_in_queue_generation(): - captured_weights = None - captured_fixed_pos = None - - async def one_candidate(self, user_id, limit=10, exclude_meme_ids=[], **kw): - return [{"id": 4, "recommended_by": "generic"}] - - class TestRetriever(CandidatesRetriever): - engine_map = { - "best_uploaded_memes": one_candidate, - "like_spread_and_recent_memes": one_candidate, - "lr_smoothed": one_candidate, - "recently_liked": one_candidate, - "goat": one_candidate, - "es_ranked": one_candidate, - } - - treatment_weights = dict(MATURE_BLENDER_CONTROL_WEIGHTS) - lr_weight = treatment_weights.pop("lr_smoothed") - treatment_weights["text_light_lr_smoothed"] = lr_weight - - def capture_blend(candidates_dict, weights_dict, fixed_pos=None, limit=0, random_seed=None): - nonlocal captured_weights, captured_fixed_pos - captured_weights = weights_dict - captured_fixed_pos = fixed_pos - return [{"id": 4, "recommended_by": "lr_smoothed"}] - - with ( - patch( - "src.recommendations.meme_queue.get_mature_blend_weights_with_experiments", - new_callable=AsyncMock, - return_value=MATURE_BLENDER_CONTROL_WEIGHTS, - ), - patch( - "src.recommendations.meme_queue.get_text_light_blender_v1_weights", - new_callable=AsyncMock, - return_value=treatment_weights, - ) as get_text_light_weights, - patch("src.recommendations.meme_queue.blend", side_effect=capture_blend), - ): - candidates = await generate_recommendations( - TEST_USER_ID, 10, 200, TestRetriever(), random_seed=102 - ) - - assert candidates == [{"id": 4, "recommended_by": "lr_smoothed"}] - assert captured_weights == MATURE_BLENDER_CONTROL_WEIGHTS - assert captured_fixed_pos == {0: "lr_smoothed"} - get_text_light_weights.assert_not_awaited() + assert captured_weights == dict(MATURE_BLEND_WEIGHTS) @pytest.mark.asyncio -async def test_recently_liked_blender_v2_assignment_is_mature_only(): - async def empty(self, user_id, limit=10, exclude_meme_ids=[], **kw): - return [] - - class TestRetriever(CandidatesRetriever): - engine_map = { - "best_uploaded_memes": empty, - "like_spread_and_recent_memes": empty, - "lr_smoothed": empty, - "recently_liked": empty, - "goat": empty, - "es_ranked": empty, - "cold_start_explore": empty, - "cold_start_adapt": empty, - "text_light_lr_smoothed": empty, - } - - with patch( - "src.recommendations.meme_queue.get_mature_blend_weights_with_experiments", - new_callable=AsyncMock, - return_value=MATURE_BLENDER_TREATMENT_WEIGHTS, - ) as get_weights: - await generate_recommendations(TEST_USER_ID, 10, 29, TestRetriever()) - await generate_recommendations(TEST_USER_ID, 10, 99, TestRetriever()) - - get_weights.assert_not_awaited() - - -@pytest.mark.asyncio -async def test_recently_liked_blender_v2_skips_moderator_assignment(): +async def test_moderator_mature_uses_default_weights(): async def one_candidate(self, user_id, limit=10, exclude_meme_ids=[], **kw): return [{"id": 4, "recommended_by": "recently_liked"}] @@ -738,11 +589,6 @@ def capture_blend(candidates_dict, weights_dict, fixed_pos=None, limit=0, random return [{"id": 4, "recommended_by": "recently_liked"}] with ( - patch( - "src.recommendations.meme_queue.get_mature_blend_weights_with_experiments", - new_callable=AsyncMock, - return_value=MATURE_BLENDER_TREATMENT_WEIGHTS, - ) as get_weights, patch( "src.recommendations.meme_queue.get_user_info", new_callable=AsyncMock, @@ -760,8 +606,7 @@ def capture_blend(candidates_dict, weights_dict, fixed_pos=None, limit=0, random ) assert candidates == [{"id": 4, "recommended_by": "recently_liked"}] - assert captured_weights == MATURE_BLENDER_CONTROL_WEIGHTS - get_weights.assert_not_awaited() + assert captured_weights == dict(MATURE_BLEND_WEIGHTS) @pytest.mark.asyncio @@ -779,17 +624,12 @@ class TestRetriever(CandidatesRetriever): "es_ranked": empty, } - with patch( - "src.recommendations.meme_queue.get_mature_blend_weights_with_experiments", - new_callable=AsyncMock, - return_value=MATURE_BLENDER_CONTROL_WEIGHTS, - ): - # All engines empty → empty result - candidates = await generate_recommendations(TEST_USER_ID, 10, 200, TestRetriever()) - assert len(candidates) == 0 + # All engines empty → empty result + candidates = await generate_recommendations(TEST_USER_ID, 10, 200, TestRetriever()) + assert len(candidates) == 0 - # Same for high meme count — no fallback engine anymore - candidates = await generate_recommendations(TEST_USER_ID, 10, 1200, TestRetriever()) + # Same for high meme count — no fallback engine anymore + candidates = await generate_recommendations(TEST_USER_ID, 10, 1200, TestRetriever()) assert len(candidates) == 0 @@ -1015,11 +855,6 @@ async def test_gate_on_mature_user_unchanged(): with ( _patch_user_info(nsessions=5, nmemes_sent=120), patch("src.config.settings.COLD_START_NSESSIONS_GATE_ENABLED", True), - patch( - "src.recommendations.meme_queue.get_mature_blend_weights_with_experiments", - new_callable=AsyncMock, - return_value=MATURE_BLENDER_TREATMENT_WEIGHTS, - ) as get_weights, ): candidates = await generate_recommendations( TEST_USER_ID, 10, nmemes_sent=120, retriever=retriever, random_seed=42 @@ -1027,7 +862,6 @@ async def test_gate_on_mature_user_unchanged(): sources = {c["recommended_by"] for c in candidates} assert "cold_start_explore" not in sources assert "cold_start_adapt" not in sources - get_weights.assert_awaited_once_with(TEST_USER_ID) @pytest.mark.asyncio diff --git a/tests/recommendations/test_pipeline.py b/tests/recommendations/test_pipeline.py index 131d6b83..63416a28 100644 --- a/tests/recommendations/test_pipeline.py +++ b/tests/recommendations/test_pipeline.py @@ -1,8 +1,8 @@ -import json from typing import Any import pytest +from src.feed_turn.planner import MATURE_BLEND_WEIGHTS from src.recommendations.candidates import CandidatesRetriever from src.recommendations.pipeline import ( RecommendationBatchPipeline, @@ -117,26 +117,15 @@ async def _empty_low_sent_fetcher( return [] -async def _identity_text_light_weights( - user_id: int, - weights: dict[str, float], -) -> dict[str, float]: - del user_id - return weights - - def _pipeline( retriever: FakeRetriever, *, low_sent_fetcher=_empty_low_sent_fetcher, - shadow_scorer=None, ) -> RecommendationBatchPipeline: return RecommendationBatchPipeline( retriever=retriever, blend_func=_fake_blend, low_sent_fetcher=low_sent_fetcher, - shadow_scorer=shadow_scorer, - text_light_weights_func=_identity_text_light_weights, ) @@ -441,8 +430,9 @@ async def test_blend_engine_failure_is_recorded_but_other_engines_continue(): result = await _pipeline(retriever).run(_request(nmemes_sent=50)) assert _ids(result.selected) == [101, 102] + assert result.diagnostics.engine_error_count == 1 assert "goat" in repr(result.diagnostics) - assert "temporarily unavailable" in repr(result.diagnostics) + assert "RuntimeError" in repr(result.diagnostics) @pytest.mark.asyncio @@ -516,9 +506,6 @@ async def fetch_all(query, params): captured["bad_fresh_ids"] = params["meme_ids"] return [{"id": 102}] - async def mature_weights(user_id): - return {"lr_smoothed": 0.6, "like_spread_and_recent_memes": 0.4} - retriever = FakeRetriever( { "lr_smoothed": [_meme(101), _meme(102)], @@ -530,25 +517,19 @@ async def mature_weights(user_id): blend_func=blend, fetch_all_func=fetch_all, low_sent_fetcher=_empty_low_sent_fetcher, - mature_weights_func=mature_weights, - text_light_weights_func=_identity_text_light_weights, ) result = await pipeline.run(_request(limit=5, nmemes_sent=200, explore_budget_enabled=True)) - assert captured["weights"] == pytest.approx( - {"lr_smoothed": 0.51, "like_spread_and_recent_memes": 0.34, "fresh_viral": 0.15} - ) + total = sum(MATURE_BLEND_WEIGHTS.values()) + expected = {engine: weight * 0.85 for engine, weight in MATURE_BLEND_WEIGHTS.items()} + expected["fresh_viral"] = total * 0.15 + assert captured["weights"] == pytest.approx(expected) assert captured["fixed_pos"] == {0: "like_spread_and_recent_memes"} assert captured["bad_fresh_ids"] == [101, 102, 301] assert _ids(result.selected) == [101, 301] -async def _lr_only_mature_weights(user_id: int) -> dict[str, float]: - del user_id - return {"lr_smoothed": 1.0} - - @pytest.mark.asyncio async def test_explore_budget_control_keeps_current_feed(): async def fetch_all(query, params): @@ -560,130 +541,9 @@ async def fetch_all(query, params): blend_func=_fake_blend, fetch_all_func=fetch_all, low_sent_fetcher=_empty_low_sent_fetcher, - mature_weights_func=_lr_only_mature_weights, - text_light_weights_func=_identity_text_light_weights, ) result = await pipeline.run(_request(limit=5, nmemes_sent=200)) assert _ids(result.selected) == [101] assert "fresh_viral" not in {call["engine"] for call in retriever.calls} - - -@pytest.mark.asyncio -async def test_source_diversity_is_disabled_by_default(): - retriever = FakeRetriever( - { - "lr_smoothed": [ - _meme(101, "lr_smoothed", meme_source_id=1), - _meme(102, "lr_smoothed", meme_source_id=1), - _meme(103, "lr_smoothed", meme_source_id=2), - ] - } - ) - - result = await _pipeline(retriever).run(_request(nmemes_sent=50)) - - assert _ids(result.selected) == [101, 102, 103] - - -@pytest.mark.asyncio -async def test_source_diversity_enabled_reorders_only_when_source_metadata_exists(): - retriever = FakeRetriever( - { - "lr_smoothed": [ - _meme(101, "lr_smoothed", meme_source_id=1), - _meme(102, "lr_smoothed", meme_source_id=1), - _meme(103, "lr_smoothed", meme_source_id=2), - _meme(104, "lr_smoothed", meme_source_id=1), - ] - } - ) - - with_source_metadata = await _pipeline(retriever).run( - _request(nmemes_sent=50, source_diversity_enabled=True) - ) - - assert _ids(with_source_metadata.selected) == [101, 103, 102, 104] - - retriever_without_metadata = FakeRetriever( - { - "lr_smoothed": [ - _meme(201, "lr_smoothed"), - _meme(202, "lr_smoothed"), - _meme(203, "lr_smoothed"), - ] - } - ) - - without_source_metadata = await _pipeline(retriever_without_metadata).run( - _request(nmemes_sent=50, source_diversity_enabled=True) - ) - - assert _ids(without_source_metadata.selected) == [201, 202, 203] - - -@pytest.mark.asyncio -async def test_shadow_scoring_records_summary_without_changing_selected_order(): - async def shadow_scorer( - candidates: list[dict[str, Any]], - request: RecommendationBatchRequest, - ) -> dict[str, Any]: - del request - return { - "summary": { - "candidate_count": len(candidates), - "top_shadow_id": 103, - }, - "scores": { - 101: 0.1, - 102: 0.5, - 103: 0.9, - }, - } - - retriever = FakeRetriever( - { - "lr_smoothed": [ - _meme(101, "lr_smoothed"), - _meme(102, "lr_smoothed"), - _meme(103, "lr_smoothed"), - ] - } - ) - - result = await _pipeline(retriever, shadow_scorer=shadow_scorer).run( - _request(nmemes_sent=50, shadow_scoring_enabled=True) - ) - - assert _ids(result.selected) == [101, 102, 103] - assert result.diagnostics.shadow_scoring_payload == { - "summary": { - "candidate_count": 3, - "top_shadow_id": 103, - }, - "scores": { - 101: 0.1, - 102: 0.5, - 103: 0.9, - }, - } - - -@pytest.mark.asyncio -async def test_compact_diagnostics_exclude_candidate_ids(): - retriever = FakeRetriever( - { - "lr_smoothed": [ - _meme(101, "lr_smoothed"), - _meme(102, "lr_smoothed"), - ] - } - ) - - result = await _pipeline(retriever).run(_request(nmemes_sent=50)) - - compact_json = json.dumps(result.diagnostics.compact(), sort_keys=True) - assert "selected_ids" not in compact_json - assert "candidate_ids" not in compact_json - assert result.diagnostics.full()["selected_ids"] == [101, 102] From 864c6fb2fe5149e9c768344be4634ac8ba7d59b5 Mon Sep 17 00:00:00 2001 From: Daniil Okhlopkov <5613295+ohld@users.noreply.github.com> Date: Sat, 31 Oct 2026 13:26:24 -0300 Subject: [PATCH 2/2] docs(specs): drop deleted blender/overlay from recommendations stage table --- specs/recommendations.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/specs/recommendations.md b/specs/recommendations.md index 1ba31332..530aea1a 100644 --- a/specs/recommendations.md +++ b/specs/recommendations.md @@ -38,8 +38,8 @@ Removed engines: `fast_dopamine`, `classic`, `multiply_all_scores`, `selected_so | Stage | Trigger | Engines | Source | |-------|---------|---------|--------| | Cold start | nmemes_sent < 30 | 3-phase adaptive with text-light guards: explore/adapt/fallback avoid OCR text above 30 words | [`pipeline.py`](../src/recommendations/pipeline.py) | -| Growing | 30-100 | Control uses `lr_smoothed`; the text-light overlay is disabled unless explicitly enabled by a registered experiment | [`pipeline.py`](../src/recommendations/pipeline.py) | -| Mature | 100+ | Recently-liked blender v2; the text-light overlay is disabled unless explicitly enabled by a registered experiment | [`pipeline.py`](../src/recommendations/pipeline.py) | +| Growing | 30-100 | `lr_smoothed`-led blend (see `src/feed_turn/planner.py`) | [`pipeline.py`](../src/recommendations/pipeline.py) | +| Mature | 100+ | `MATURE_BLEND_WEIGHTS` blend (`src/feed_turn/planner.py`); `explore_budget_v1` adds `fresh_viral` for treatment | [`pipeline.py`](../src/recommendations/pipeline.py) | | Moderator/Admin | user_type check | 75% low_sent_pool + 25% regular (by maturity) | [`pipeline.py`](../src/recommendations/pipeline.py) | `fixed_pos={0: "lr_smoothed"}` forces first position to `lr_smoothed` in blended mode.