diff --git a/docs/analyst/crossposting-v5-shadow.sql b/docs/analyst/crossposting-v5-shadow.sql index 752b8fca..656e3fc1 100644 --- a/docs/analyst/crossposting-v5-shadow.sql +++ b/docs/analyst/crossposting-v5-shadow.sql @@ -14,7 +14,7 @@ WITH dec AS ( SELECT d.channel, d.decided_at, d.picked_meme_id, (d.candidates->0->>'shadow_v5_pick_meme_id')::int AS v5_pick, (d.candidates->0->>'score_v5')::float AS score_v5, - (d.candidates->0->>'final_score')::float AS score_v4 + COALESCE(d.candidates->0->'scores'->>'v4', d.candidates->0->>'final_score')::float AS score_v4 FROM crossposting_decision_log d WHERE d.channel IN ('tgchannelru', 'tgchannelen') AND d.candidates->0->>'shadow_v5_pick_meme_id' IS NOT NULL diff --git a/experiments/2026-08-10-crosspost-v4-and-taste-cohort.md b/experiments/2026-08-10-crosspost-v4-and-taste-cohort.md index e0556d4e..00b16d20 100644 --- a/experiments/2026-08-10-crosspost-v4-and-taste-cohort.md +++ b/experiments/2026-08-10-crosspost-v4-and-taste-cohort.md @@ -11,7 +11,7 @@ check_on: 2026-08-17 decided: 2026-10-30 result: hit rate 32.1% to 41.0% (+8.9 pp, z~2.4, n 299/373), f1k 26.2 to 28.9; manual posts also +3.4 f1k (channel drift) learning: Bot like volume, not like rate, predicts channel forwards; part of the lift is channel drift -code: CROSSPOST_RU_MEME_LIKE_VOLUME_ENABLED, src/crossposting/service.py (score_version=4) +code: src/crossposting/scoring.py LIVE (score_version=4) readout: docs/analyst/crossposting-v4-like-volume.sql --- @@ -48,7 +48,7 @@ fill **Metrics after** only. | Field | Value | |-------|--------| -| Code | `CROSSPOST_RU_MEME_LIKE_VOLUME_ENABLED` (default ON) | +| Code | `LIVE["tgchannelru"] = "v4"` in `src/crossposting/scoring.py` (was `CROSSPOST_RU_MEME_LIKE_VOLUME_ENABLED` until 2026-10-31) | | Started | 2026-08-10 ~05:20 UTC first post | | Kill | env `false` | @@ -176,7 +176,7 @@ Readout: [`docs/analyst/readouts/2026-10-30-experiment-closure.md`](../docs/anal |-------|--------| | **ID** | `crosspost_meme_level_v1` / **score_version=4** | | **Status** | **KEEP at 100% 2026-10-30**: hit rate 32.1 → 41.0% (+8.9 pp, z≈2.4), f1k 26.2 → 28.9 | -| **Kill switch** | `CROSSPOST_RU_MEME_LIKE_VOLUME_ENABLED=false` → v2 | +| **Kill switch** | set `LIVE["tgchannelru"] = "v2"` in `src/crossposting/scoring.py` | | **Readout offline** | `docs/analyst/readouts/2026-08-09-crosspost-meme-level-offline.md` | | **Readout online** | `docs/analyst/crossposting-v4-like-volume.sql` | | **n offline** | 611 RU image posts / 120d | diff --git a/experiments/2026-10-30-crosspost-v5-shadow.md b/experiments/2026-10-30-crosspost-v5-shadow.md index 1390b2da..60ce8456 100644 --- a/experiments/2026-10-30-crosspost-v5-shadow.md +++ b/experiments/2026-10-30-crosspost-v5-shadow.md @@ -11,7 +11,7 @@ check_on: 2026-11-14 decided: result: unknown learning: -code: src/crossposting/shadow_score.py +code: src/crossposting/scoring.py readout: docs/analyst/crossposting-v5-shadow.sql --- @@ -50,8 +50,8 @@ EN score_v5 = theta_src · exp(user_taste_z) ``` Logged per candidate in `crossposting_decision_log.candidates[]`: `theta_src`, -`user_taste_z`, `score_v5`, `shadow_v5_rank`, `shadow_v5_pick_meme_id`, -`shadow_version = v5_src_beta_taste_v1`. +`user_taste_z`, `score_v5`, `shadow_v5_rank`, `shadow_v5_pick_meme_id` +(`shadow_version = v5_src_beta_taste_v1` until the 2026-10-31 scoring refactor). ## Online vs offline differences (known limits) @@ -78,3 +78,12 @@ Counterfactual zf exists only for v5 picks that got posted later. ## Metrics after _(fill on 2026-11-14)_ + +## Log +- 2026-10-31: scoring refactor: every crosspost score now lives in + `src/crossposting/scoring.py` (`FACTORS`, `SCORES`, `LIVE`). Promote this shadow = + set `LIVE[channel]` to `"v5"` (score_version 5); readouts unchanged, new rows + also carry `factors`, `scores`, `ranks`, `picks`, `picked_by` dicts. Caveat (unchanged): + shadow scores rank only the production top-5 (`PRESELECT`), not the whole pool. + Replay before/after: same pick and top-5 on 120 local fixture runs, the last 40 prod + decisions and the current prod pool (both channels). diff --git a/experiments/2026-10-31-crosspost-content-tags.md b/experiments/2026-10-31-crosspost-content-tags.md index a0922afa..412eb86a 100644 --- a/experiments/2026-10-31-crosspost-content-tags.md +++ b/experiments/2026-10-31-crosspost-content-tags.md @@ -11,7 +11,7 @@ check_on: 2026-11-20 decided: result: unknown learning: -code: Describe Memes content tags, src/crossposting/shadow_score.py +code: Describe Memes content tags, src/crossposting/scoring.py readout: docs/analyst/crossposting-content-tags-gates.py, docs/analyst/crossposting-content-tags-shadow.sql --- @@ -80,3 +80,12 @@ Also report `crossposting-content-tags-shadow.sql`: _(fill on 2026-11-20)_ - 2026-10-31 ~16:00 UTC: prompt example changed from `false` to `true or false` for both booleans (models echoed the example). Count Gate A only on tags with calculated_at after this deploy. + +## Log +- 2026-10-31: scoring refactor: every crosspost score now lives in + `src/crossposting/scoring.py` (`FACTORS`, `SCORES`, `LIVE`). Promote this shadow = + set `LIVE[channel]` to `"v5c"` (score_version 6); readouts unchanged, new rows + also carry `factors`, `scores`, `ranks`, `picks`, `picked_by` dicts. Caveat (unchanged): + shadow scores rank only the production top-5 (`PRESELECT`), not the whole pool. + Replay before/after: same pick and top-5 on 120 local fixture runs, the last 40 prod + decisions and the current prod pool (both channels). diff --git a/experiments/2026-11-01-crosspost-src-forward-rate.md b/experiments/2026-11-01-crosspost-src-forward-rate.md index cb8d9b1a..7361af66 100644 --- a/experiments/2026-11-01-crosspost-src-forward-rate.md +++ b/experiments/2026-11-01-crosspost-src-forward-rate.md @@ -11,7 +11,7 @@ check_on: 2026-11-28 decided: result: unknown learning: -code: meme_raw_telegram.forwards, Collect Channel Stats flow, src/crossposting/shadow_score.py +code: meme_raw_telegram.forwards, Collect Channel Stats flow, src/crossposting/scoring.py readout: docs/analyst/crossposting-src-fwd-shadow.sql --- @@ -73,3 +73,12 @@ Run `crossposting-src-fwd-shadow.sql` on 2026-11-28 and bootstrap the Spearman g ## Metrics after _(fill on 2026-11-28)_ + +## Log +- 2026-10-31: scoring refactor: every crosspost score now lives in + `src/crossposting/scoring.py` (`FACTORS`, `SCORES`, `LIVE`). Promote this shadow = + set `LIVE[channel]` to `"v5f"` (score_version 7); readouts unchanged, new rows + also carry `factors`, `scores`, `ranks`, `picks`, `picked_by` dicts. Caveat (unchanged): + shadow scores rank only the production top-5 (`PRESELECT`), not the whole pool. + Replay before/after: same pick and top-5 on 120 local fixture runs, the last 40 prod + decisions and the current prod pool (both channels). diff --git a/specs/channel-growth-optimization.md b/specs/channel-growth-optimization.md index 364c551c..07930e77 100644 --- a/specs/channel-growth-optimization.md +++ b/specs/channel-growth-optimization.md @@ -53,7 +53,7 @@ Train a model that predicts whether a meme will grow the channel (high fwd/1k) B ## Architecture - Stats collector: src/flows/crossposting/stats_collector.py (Prefect flow, every 6h) -- Scoring: src/crossposting/service.py (get_next_meme_for_tgchannelru/en) +- Scoring: src/crossposting/scoring.py (factors, scores, LIVE); pool + log in service.py - CTAs: src/flows/crossposting/meme.py (CTAS list, weighted by repetition) - Env vars: TELEGRAM_API_ID, TELEGRAM_API_HASH, TELEGRAM_SESSION_STRING (same as e2e smoke tests) - Schedule: scripts/serve_flows.py (RU: 8,10,11,12,14,16 MSK, EN: 8,10,14,18,20 MSK) diff --git a/specs/crossposting-source-quality.md b/specs/crossposting-source-quality.md index e460bb76..326a0174 100644 --- a/specs/crossposting-source-quality.md +++ b/specs/crossposting-source-quality.md @@ -26,7 +26,8 @@ Validated finding from Phase 1 (do not relitigate): ## Implementation shape -In `src/crossposting/service.py`, both `get_next_meme_for_tgchannelru` and `get_next_meme_for_tgchannelen`: +Original shape (now `_POOL_QUERY` in `src/crossposting/service.py` + `src_quality_mult` in `src/crossposting/scoring.py`; +note the shipped SQL gave sources with n<5 a 0.5 multiplier, not the neutral 1.0 planned here): ```sql WITH src_quality AS ( diff --git a/src/config.py b/src/config.py index 0998cfa6..d7f1d30d 100644 --- a/src/config.py +++ b/src/config.py @@ -73,10 +73,6 @@ class Config(BaseSettings): # Owner flips on after deploy. Set HOLDOUT_PCT=0 after the v1 readout. MONTHLY_REMINDER_ENABLED: bool = False MONTHLY_REMINDER_HOLDOUT_PCT: int = 50 - # RU crosspost ranker: multiply score by ln(nlikes+1) (meme volume, not LR). - # Offline 2026-08-09: src×log1p(likes) top-20% lift ~1.14 on time-split. - # Kill switch reverts @fastfoodmemes to score_version=2 behavior. - CROSSPOST_RU_MEME_LIKE_VOLUME_ENABLED: bool = True PREFECT_API_URL: str | None = None diff --git a/src/crossposting/scoring.py b/src/crossposting/scoring.py new file mode 100644 index 00000000..f91fb139 --- /dev/null +++ b/src/crossposting/scoring.py @@ -0,0 +1,114 @@ +"""Crosspost scoring: every live and shadow score, in one place. + +A *factor* is a named number computed from one candidate row (``None`` when its +inputs are missing). A *score* is the product of named factors. Candidates are +preselected from the whole pool by ``PRESELECT`` (needs only pool columns), the +top-N get shadow inputs attached, then ``LIVE`` picks among them. + +Promote a shadow score on a channel: edit ``LIVE``. Add a score: add its factors +to ``FACTORS`` and a tuple to ``SCORES`` (plus a ``SCORE_VERSION`` int). +Shadow scores are only ranked within the PRESELECT top-N. +""" + +from __future__ import annotations + +import math +from collections.abc import Callable +from typing import Any + +Row = dict[str, Any] + +# impr_penalty is 0, not the intended 0.8 (RU) / 0.5 (EN): the old SQL bound it into +# `CASE ... THEN 1 ELSE :impr_penalty END`, which Postgres typed integer, so asyncpg +# sent 0. Production has always zeroed memes with raw_impr_rank > 1; kept as-is. +CHANNEL_PARAMS: dict[str, dict[str, Any]] = { + "tgchannelru": {"impr_penalty": 0.0, "age_threshold": 7, "language_code": "ru"}, + "tgchannelen": {"impr_penalty": 0.0, "age_threshold": 90, "language_code": "en"}, +} + + +def _src_quality_mult(r: Row, channel: str) -> float: + # Mirrors the old SQL: LEAST(2, GREATEST(0.5, signal / NULLIF(median, 0))). + # GREATEST ignores NULL, so a source without enough history scores 0.5, not 1.0. + signal, median = r.get("src_signal"), r.get("median_signal") + if signal is None or not median: + return 0.5 + return max(0.5, min(2.0, float(signal) / float(median))) + + +def _v5_base(r: Row, channel: str) -> float | None: + """2026-10-30 walk-forward study (experiments/2026-10-30-crosspost-v5-shadow.md). + + RU: ln(nlikes+1) * theta_src * exp(0.5 * user_taste_z); EN: theta_src * exp(user_taste_z). + """ + if r.get("theta_src") is None: + return None + theta, taste = float(r["theta_src"]), float(r.get("user_taste_z") or 0.0) + if channel == "tgchannelru": + return math.log(r["nlikes"] + 1.0) * theta * math.exp(0.5 * taste) + return theta * math.exp(taste) + + +FACTORS: dict[str, Callable[[Row, str], float | None]] = { + # v2 production factors (meme_stats columns are NOT NULL) + "lr_factor": lambda r, c: (r["nlikes"] + 1.0) / (r["nlikes"] + r["ndislikes"] + 1), + "impr_factor": lambda r, c: ( + 1.0 if r["raw_impr_rank"] <= 1 else CHANNEL_PARAMS[c]["impr_penalty"] + ), + "age_factor": lambda r, c: 1.0 if r["age_days"] < CHANNEL_PARAMS[c]["age_threshold"] else 0.8, + "caption_factor": lambda r, c: 0.8 if r["has_caption"] else 1.0, + "sent_factor": lambda r, c: ( + 1.0 if r["nmemes_sent"] <= 1 else (r["nlikes"] + r["ndislikes"]) / r["nmemes_sent"] + ), + "src_quality_mult": _src_quality_mult, + "invited_boost": lambda r, c: 1.0 + min(r["invited_count"], 10) * 0.1, + # v4: meme-level like volume (offline 2026-08-09) + "like_volume_factor": lambda r, c: math.log(r["nlikes"] + 1.0), + # v5 shadow: source Beta prior × user taste + "v5_base": _v5_base, + # v5c: RU blind-label study — relatable / tweet screenshots ≈ −0.1 zf each + "content_mult_v1": lambda r, c: ( + 0.85 + ** (int(r.get("relatable") is True) + int(r.get("content_format") == "tweet_screenshot")) + ), + # v5f: within-source forward-rate percentile of the source post; missing → 1.0 + "src_fwd_mult": lambda r, c: ( + 1.0 if r.get("src_fwd_pct") is None else 0.5 + float(r["src_fwd_pct"]) + ), +} + +_V2 = ( + "lr_factor", + "impr_factor", + "age_factor", + "caption_factor", + "sent_factor", + "src_quality_mult", + "invited_boost", +) +SCORES: dict[str, tuple[str, ...]] = { + "v2": _V2, + "v4": (*_V2, "like_volume_factor"), + "v5": ("v5_base",), + "v5c": ("v5_base", "content_mult_v1"), + "v5f": ("v5_base", "src_fwd_mult"), +} +# crossposting.score_version / decision_log.score_version for the picking score. +SCORE_VERSION = {"v2": 2, "v4": 4, "v5": 5, "v5c": 6, "v5f": 7} + +PRESELECT = {"tgchannelru": "v4", "tgchannelen": "v2"} +LIVE = {"tgchannelru": "v4", "tgchannelen": "v2"} + + +def score(row: Row, channel: str, name: str) -> float | None: + vals = [FACTORS[f](row, channel) for f in SCORES[name]] + return None if None in vals else math.prod(vals) + + +def score_all(row: Row, channel: str) -> tuple[dict[str, float | None], dict[str, float | None]]: + factors = {name: f(row, channel) for name, f in FACTORS.items()} + scores = { + name: None if any(factors[f] is None for f in fs) else math.prod(factors[f] for f in fs) + for name, fs in SCORES.items() + } + return factors, scores diff --git a/src/crossposting/service.py b/src/crossposting/service.py index db7ba743..26bf24a8 100644 --- a/src/crossposting/service.py +++ b/src/crossposting/service.py @@ -1,17 +1,19 @@ -import math +import heapq +import logging from typing import Any from sqlalchemy import text from sqlalchemy.dialects.postgresql import insert -from src.config import settings from src.crossposting.constants import Channel -from src.crossposting.shadow_score import ( - SHADOW_V5_VERSION, - attach_v5_ranks, - content_mult_v1, - score_v5, - src_fwd_mult, +from src.crossposting.scoring import ( + CHANNEL_PARAMS, + LIVE, + PRESELECT, + SCORE_VERSION, + SCORES, + score, + score_all, ) from src.database import ( crossposting, @@ -21,10 +23,7 @@ fetch_one, ) -# score_version in crossposting + decision_log -SCORE_VERSION_V2 = 2 -# RU: v2 factors × ln(nlikes+1) — meme-level like volume (docs/analyst/readouts/2026-08-09-…) -SCORE_VERSION_MEME_LIKE_VOLUME = 4 +logger = logging.getLogger(__name__) async def log_meme_sent( @@ -76,156 +75,7 @@ async def posted_recently(channel: Channel, minutes: int = CATCH_UP_GAP_MINUTES) return row is not None -# Per-channel ranker constants (mirror the SQL ORDER BY). -_CHANNEL_PARAMS: dict[str, dict[str, Any]] = { - "tgchannelru": {"impr_penalty": 0.8, "age_threshold": 7, "language_code": "ru"}, - "tgchannelen": {"impr_penalty": 0.5, "age_threshold": 90, "language_code": "en"}, -} - - -def _compute_score_breakdown( - row: dict[str, Any], - channel: str, - *, - like_volume_enabled: bool = False, -) -> dict[str, Any]: - """Reproduce the SQL ORDER BY in Python so each candidate gets a logged score - breakdown. Must stay in sync with the ORDER BY in the SQL queries below.""" - params = _CHANNEL_PARAMS[channel] - nlikes = row.get("nlikes") or 0 - ndislikes = row.get("ndislikes") or 0 - nmemes_sent = row.get("nmemes_sent") or 0 - raw_impr_rank = row.get("raw_impr_rank") - age_days = row.get("age_days") or 0 - invited_count = row.get("invited_count") or 0 - src_signal = row.get("src_signal") - median_signal = row.get("median_signal") - caption_present = row.get("caption") is not None - - denom = nlikes + ndislikes + 1 - lr_factor = (nlikes + 1.0) / denom if denom else 0.5 - impr_factor = ( - 1.0 if (raw_impr_rank is not None and raw_impr_rank <= 1) else params["impr_penalty"] - ) - age_factor = 1.0 if age_days < params["age_threshold"] else 0.8 - caption_factor = 0.8 if caption_present else 1.0 - sent_factor = 1.0 if nmemes_sent <= 1 else (nlikes + ndislikes) / nmemes_sent - - if src_signal is not None and median_signal: - # SQL returns NUMERIC as Decimal; cast both sides for safe arithmetic. - src_quality_mult = max(0.5, min(2.0, float(src_signal) / float(median_signal))) - else: - src_quality_mult = 1.0 - - invited_boost = 1.0 + min(invited_count, 10) * 0.1 - # Meme-level volume (not LR): prefers "well tested in bot" over high-LR sparse. - like_volume_factor = math.log(float(nlikes) + 1.0) if like_volume_enabled else 1.0 - - final_score = ( - lr_factor - * impr_factor - * age_factor - * caption_factor - * sent_factor - * src_quality_mult - * invited_boost - * like_volume_factor - ) - - out = { - "lr_factor": round(lr_factor, 4), - "impr_factor": round(impr_factor, 4), - "age_factor": round(age_factor, 4), - "caption_factor": round(caption_factor, 4), - "sent_factor": round(sent_factor, 4), - "src_quality_mult": round(src_quality_mult, 4), - "invited_boost": round(invited_boost, 4), - "like_volume_factor": round(like_volume_factor, 4), - "like_volume_enabled": like_volume_enabled, - "final_score": round(final_score, 6), - } - return out - - -def _build_decision_log( - channel: str, - score_version: int, - candidates: list[dict[str, Any]], - *, - like_volume_enabled: bool = False, -) -> dict[str, Any] | None: - """Build the kwargs dict for log_ranker_decision from a top-N candidate list. - Returns None if candidates is empty.""" - if not candidates: - return None - picked = candidates[0] - log_candidates = [] - for i, row in enumerate(candidates): - breakdown = _compute_score_breakdown(row, channel, like_volume_enabled=like_volume_enabled) - log_candidate = { - "rank": i + 1, - "meme_id": row["id"], - "source_id": row.get("meme_source_id"), - "nlikes": row.get("nlikes"), - "ndislikes": row.get("ndislikes"), - "raw_impr_rank": row.get("raw_impr_rank"), - "age_days": row.get("age_days"), - "nmemes_sent": row.get("nmemes_sent"), - "invited_count": row.get("invited_count"), - "pre_inbot_share_clicks": row.get("pre_inbot_share_clicks") or 0, - "pre_inbot_share_click_users": row.get("pre_inbot_share_click_users") or 0, - "caption_present": row.get("caption") is not None, - "src_signal": ( - round(float(row["src_signal"]), 4) if row.get("src_signal") is not None else None - ), - **breakdown, - # v5 shadow (log only): see shadow_score.py; None when enrichment failed. - "shadow_version": SHADOW_V5_VERSION, - "theta_src": row.get("theta_src"), - "user_taste_z": row.get("user_taste_z"), - "score_v5": ( - round(score_v5(channel, row["nlikes"], row["theta_src"], row["user_taste_z"]), 6) - if row.get("theta_src") is not None - else None - ), - # v5c shadow (log only): content tags from Describe Memes. - "content_format": row.get("content_format"), - "relatable": row.get("relatable"), - "politics_news": row.get("politics_news"), - "content_mult_v1": content_mult_v1(row.get("content_format"), row.get("relatable")), - # v5f shadow (log only): source post's within-source forward-rate percentile. - "src_fwd_pct": row.get("src_fwd_pct"), - } - v5 = log_candidate["score_v5"] - log_candidate["score_v5c"] = ( - round(v5 * log_candidate["content_mult_v1"], 6) if v5 is not None else None - ) - log_candidate["score_v5f"] = ( - round(v5 * src_fwd_mult(row.get("src_fwd_pct")), 6) if v5 is not None else None - ) - log_candidates.append(log_candidate) - - # Shadow ranks among this top-N only (production order unchanged). - # Keep shadow_* only inside candidates[] — log_ranker_decision kwargs are fixed columns. - log_candidates = attach_v5_ranks(log_candidates) - log_candidates = attach_v5_ranks(log_candidates, version="v5c") - log_candidates = attach_v5_ranks(log_candidates, version="v5f") - - return { - "channel": channel, - "picked_meme_id": picked["id"], - "score_version": score_version, - "median_signal": ( - round(float(picked["median_signal"]), 4) - if picked.get("median_signal") is not None - else None - ), - "pool_size": picked.get("candidate_pool_size"), - "candidates": log_candidates, - } - - -# v5 shadow inputs per candidate, one bounded query per decision (~1.3s on prod). +# Shadow inputs for the top-N, one bounded query per decision (~1.3s on prod). # Labels: zf = ln(1+forwards) - channel×MSK-week mean over image posts 48h..365d old. # ``crossposting.forwards`` is the latest Telethon snapshot (frozen once the post # leaves the 30d sweep) — the same field the 2026-10-30 study used, not a 24h snapshot. @@ -239,7 +89,7 @@ def _build_decision_log( # ponytail: likes limited to sent_at > now()-90d — user_meme_reaction has no valid # meme_id-leading index on prod, so this window keeps the scan bounded. Widen it # (or precompute weights) if user coverage proves too thin. -_V5_SHADOW_QUERY = """ +_SHADOW_INPUTS_QUERY = """ WITH posts AS MATERIALIZED ( SELECT cp.meme_id, cp.created_at, m.meme_source_id AS src, LN(1 + cp.forwards) AS lfw, @@ -307,26 +157,41 @@ def _build_decision_log( THEN (m.ocr_result->>'relatable')::boolean END AS relatable, CASE WHEN jsonb_typeof(m.ocr_result->'politics_news') = 'boolean' THEN (m.ocr_result->>'politics_news')::boolean END AS politics_news, - CASE WHEN sf.n_ref >= 10 THEN sf.pct END::float AS src_fwd_pct + CASE WHEN sf.n_ref >= 10 THEN sf.pct END::float AS src_fwd_pct, + share_clicks.pre_inbot_share_clicks, + share_clicks.pre_inbot_share_click_users FROM meme m CROSS JOIN pbar CROSS JOIN wsd LEFT JOIN src s ON s.src = m.meme_source_id LEFT JOIN taste t ON t.meme_id = m.id LEFT JOIN src_fwd sf ON sf.id = m.raw_meme_id AND sf.meme_source_id = m.meme_source_id + -- in-bot share-link clicks by non-sharers before this decision + CROSS JOIN LATERAL ( + SELECT + COUNT(*) AS pre_inbot_share_clicks, + COUNT(DISTINCT user_id) AS pre_inbot_share_click_users + FROM user_deep_link_log udll + CROSS JOIN LATERAL ( + SELECT substring( + udll.deep_link FROM ('^[ms]_([1-9][0-9]{0,18})_' || m.id || '$') + ) AS sharer_id + ) share_link + WHERE udll.created_at < NOW() + AND CASE + WHEN share_link.sharer_id IS NULL THEN false + WHEN length(share_link.sharer_id) = 19 + AND share_link.sharer_id > '9223372036854775807' THEN false + ELSE udll.user_id <> share_link.sharer_id::bigint + END + ) share_clicks WHERE m.id = ANY(:meme_ids) """ -async def _enrich_candidates_with_v5_shadow( - candidates: list[dict[str, Any]], - *, - channel: str, -) -> list[dict[str, Any]]: - """Attach theta_src + user_taste_z for top-N (does not re-rank).""" +async def _attach_shadow_inputs(candidates: list[dict[str, Any]], channel: str) -> None: + """Attach v5/v5c/v5f inputs + share-click counts to the top-N in place.""" meme_ids = [int(c["id"]) for c in candidates] - if not meme_ids: - return candidates - rows = await fetch_all(text(_V5_SHADOW_QUERY), {"channel": channel, "meme_ids": meme_ids}) + rows = await fetch_all(text(_SHADOW_INPUTS_QUERY), {"channel": channel, "meme_ids": meme_ids}) by_id = {int(r["meme_id"]): r for r in rows} for c in candidates: r = by_id.get(int(c["id"])) @@ -338,7 +203,8 @@ async def _enrich_candidates_with_v5_shadow( c["politics_news"] = r["politics_news"] pct = r["src_fwd_pct"] c["src_fwd_pct"] = round(float(pct), 4) if pct is not None else None - return candidates + c["pre_inbot_share_clicks"] = r["pre_inbot_share_clicks"] + c["pre_inbot_share_click_users"] = r["pre_inbot_share_click_users"] async def log_ranker_decision( @@ -368,19 +234,19 @@ async def log_ranker_decision( _PICKED_FIELDS = ("id", "type", "telegram_file_id", "caption") -def _picked_meme_dict(candidate: dict[str, Any]) -> dict[str, Any]: - return {k: candidate[k] for k in _PICKED_FIELDS} +async def _picked_meme_dict(meme_id: int) -> dict[str, Any]: + row = await fetch_one( + text(f"SELECT {', '.join(_PICKED_FIELDS)} FROM meme WHERE id = :id"), {"id": meme_id} + ) + return dict(row) -_STANDARD_RANKER_QUERY = """ - WITH selected_at AS ( - SELECT NOW() AS decided_at - ), - src_quality AS ( +# Eligible pool, unordered: ranking lives in scoring.py. +_POOL_QUERY = """ + WITH src_quality AS ( SELECT m.meme_source_id, - AVG(cp.forwards * SQRT(GREATEST(cp.views, 1) / 100.0)) AS signal, - COUNT(*) AS n_posts + AVG(cp.forwards * SQRT(GREATEST(cp.views, 1) / 100.0)) AS signal FROM crossposting cp JOIN meme m ON m.id = cp.meme_id WHERE cp.channel = :channel @@ -403,176 +269,120 @@ def _picked_meme_dict(candidate: dict[str, Any]) -> dict[str, Any]: WHERE cp2.channel = :channel AND cp2.created_at > NOW() - INTERVAL '24 hours' AND cp2.telegram_message_id IS NOT NULL - ), - ranked AS ( - SELECT - M.id, M.type, M.telegram_file_id, M.caption, - M.meme_source_id, - MS.nlikes, MS.ndislikes, MS.raw_impr_rank, - MS.age_days, MS.nmemes_sent, MS.invited_count, - SQ.signal AS src_signal, - (SELECT m_signal FROM src_median) AS median_signal, - COUNT(*) OVER () AS candidate_pool_size, - ROW_NUMBER() OVER ( - ORDER BY -1 - * COALESCE((MS.nlikes + 1.) / (MS.nlikes + MS.ndislikes + 1), 0.5) - * CASE WHEN MS.raw_impr_rank <= 1 THEN 1 ELSE :impr_penalty END - * CASE WHEN MS.age_days < :age_threshold THEN 1 ELSE 0.8 END - * CASE WHEN M.caption IS NULL THEN 1 ELSE 0.8 END - * CASE - WHEN MS.nmemes_sent <= 1 THEN 1 - ELSE (MS.nlikes + MS.ndislikes) * 1. / MS.nmemes_sent - END - * COALESCE( - LEAST(2.0, GREATEST(0.5, - SQ.signal / NULLIF((SELECT m_signal FROM src_median), 0) - )), - 1.0 - ) - * (1.0 + LEAST(MS.invited_count, 10) * 0.1) - -- meme-level volume (score_version=4): ln(nlikes+1); weight 0 → 1.0 - * CASE - WHEN :like_volume_enabled THEN LN(COALESCE(MS.nlikes, 0) + 1.0) - ELSE 1.0 - END, - M.id - ) AS candidate_rank - FROM meme M - INNER JOIN meme_stats MS ON MS.meme_id = M.id - LEFT JOIN crossposting CP ON CP.meme_id = M.id AND CP.channel = :channel - LEFT JOIN src_quality SQ ON SQ.meme_source_id = M.meme_source_id - WHERE 1=1 - AND CP.meme_id IS NULL - AND M.status = 'ok' - AND M.language_code = :language_code - AND M.type = 'image' - AND MS.nlikes >= 5 - AND M.meme_source_id NOT IN (SELECT meme_source_id FROM recent_src) - ORDER BY -1 - * COALESCE((MS.nlikes + 1.) / (MS.nlikes + MS.ndislikes + 1), 0.5) - * CASE WHEN MS.raw_impr_rank <= 1 THEN 1 ELSE :impr_penalty END - * CASE WHEN MS.age_days < :age_threshold THEN 1 ELSE 0.8 END - * CASE WHEN M.caption IS NULL THEN 1 ELSE 0.8 END - * CASE - WHEN MS.nmemes_sent <= 1 THEN 1 - ELSE (MS.nlikes + MS.ndislikes) * 1. / MS.nmemes_sent - END - * COALESCE( - LEAST(2.0, GREATEST(0.5, - SQ.signal / NULLIF((SELECT m_signal FROM src_median), 0) - )), - 1.0 - ) - * (1.0 + LEAST(MS.invited_count, 10) * 0.1) - * CASE - WHEN :like_volume_enabled THEN LN(COALESCE(MS.nlikes, 0) + 1.0) - ELSE 1.0 - END, - M.id - LIMIT :limit ) SELECT - ranked.*, - COALESCE(share_clicks.pre_inbot_share_clicks, 0) AS pre_inbot_share_clicks, - COALESCE(share_clicks.pre_inbot_share_click_users, 0) AS pre_inbot_share_click_users - FROM ranked - CROSS JOIN selected_at - LEFT JOIN LATERAL ( - SELECT - COUNT(*) AS pre_inbot_share_clicks, - COUNT(DISTINCT user_id) AS pre_inbot_share_click_users - FROM user_deep_link_log udll - CROSS JOIN LATERAL ( - SELECT substring( - udll.deep_link FROM ('^[ms]_([1-9][0-9]{0,18})_' || ranked.id || '$') - ) AS sharer_id - ) share_link - WHERE udll.created_at < selected_at.decided_at - AND CASE - WHEN share_link.sharer_id IS NULL THEN false - WHEN length(share_link.sharer_id) = 19 - AND share_link.sharer_id > '9223372036854775807' THEN false - ELSE udll.user_id <> share_link.sharer_id::bigint - END - ) share_clicks ON true - ORDER BY ranked.candidate_rank + -- Scoring columns only: the pool is ~130k rows; picked fields load by id. + M.id, M.meme_source_id, (M.caption IS NOT NULL) AS has_caption, + MS.nlikes, MS.ndislikes, MS.raw_impr_rank, + MS.age_days, MS.nmemes_sent, MS.invited_count, + SQ.signal AS src_signal, + (SELECT m_signal FROM src_median) AS median_signal + FROM meme M + INNER JOIN meme_stats MS ON MS.meme_id = M.id + LEFT JOIN crossposting CP ON CP.meme_id = M.id AND CP.channel = :channel + LEFT JOIN src_quality SQ ON SQ.meme_source_id = M.meme_source_id + WHERE CP.meme_id IS NULL + AND M.status = 'ok' + AND M.language_code = :language_code + AND M.type = 'image' + AND MS.nlikes >= 5 + AND M.meme_source_id NOT IN (SELECT meme_source_id FROM recent_src) """ +# Flat per-candidate keys still read by docs/analyst/crossposting-*.sql readouts. +_LEGACY_SHADOWS = ("v5", "v5c", "v5f") -async def _get_next_meme_for_channel( - channel: str, - *, - log_top_n: int, - score_version: int, - like_volume_enabled: bool = False, -) -> tuple[dict[str, Any] | None, dict[str, Any] | None]: - params = { - **_CHANNEL_PARAMS[channel], - "channel": channel, - "limit": log_top_n, - "like_volume_enabled": like_volume_enabled, - } - rows = await fetch_all(text(_STANDARD_RANKER_QUERY), params) - if not rows: - return None, None - # Shadow enrichments must never block posting (log-only signals). - try: - rows = await _enrich_candidates_with_v5_shadow(rows, channel=channel) - except Exception: - import logging - - logging.getLogger(__name__).warning("v5 shadow enrichment failed", exc_info=True) - return ( - _picked_meme_dict(rows[0]), - _build_decision_log( - channel, - score_version, - rows, - like_volume_enabled=like_volume_enabled, - ), - ) +def _round(x: float | None, nd: int) -> float | None: + return None if x is None else round(float(x), nd) -def _ru_like_volume_ranker_enabled() -> bool: - return bool(settings.CROSSPOST_RU_MEME_LIKE_VOLUME_ENABLED) +def _build_decision_log(channel: str, top: list[dict[str, Any]], pool_size: int) -> dict[str, Any]: + """Score the preselected top-N with every score; return log_ranker_decision kwargs.""" + scored = [score_all(r, channel) for r in top] + picks: dict[str, int] = {} + ranks: list[dict[str, int]] = [{} for _ in top] + for name in SCORES: + vals = [s[name] for _, s in scored] + if None in vals: + continue + order = sorted(range(len(top)), key=lambda i: (-vals[i], top[i]["id"])) + for rank, i in enumerate(order, start=1): + ranks[i][name] = rank + picks[name] = top[order[0]]["id"] + # LIVE may need shadow inputs; if they failed to load, fall back to PRESELECT. + picked_by = LIVE[channel] if LIVE[channel] in picks else PRESELECT[channel] -async def get_next_meme_for_tgchannelru( - log_top_n: int = 5, -) -> tuple[dict[str, Any] | None, dict[str, Any] | None]: - """Pick the top crosspost candidate for @fastfoodmemes (RU). - - Returns ``(picked_meme, decision_log)``: - - - ``picked_meme`` — narrow dict suitable for ``MemeData(**...)`` construction - (id, type, telegram_file_id, caption). ``None`` when no candidate passes - filters. - - ``decision_log`` — kwargs dict for ``log_ranker_decision``, with the top-N - candidates and per-candidate score breakdown. ``None`` when no candidates. + log_candidates = [] + for i, (row, (factors, scores)) in enumerate(zip(top, scored, strict=True)): + factors = {k: _round(v, 4) for k, v in factors.items()} + scores = {k: _round(v, 6) for k, v in scores.items()} + log_candidates.append( + { + "rank": i + 1, # PRESELECT order + "meme_id": row["id"], + "source_id": row.get("meme_source_id"), + "nlikes": row.get("nlikes"), + "ndislikes": row.get("ndislikes"), + "raw_impr_rank": row.get("raw_impr_rank"), + "age_days": row.get("age_days"), + "nmemes_sent": row.get("nmemes_sent"), + "invited_count": row.get("invited_count"), + "pre_inbot_share_clicks": row.get("pre_inbot_share_clicks") or 0, + "pre_inbot_share_click_users": row.get("pre_inbot_share_click_users") or 0, + "caption_present": bool(row.get("has_caption")), + "src_signal": _round(row.get("src_signal"), 4), + "theta_src": row.get("theta_src"), + "user_taste_z": row.get("user_taste_z"), + "content_format": row.get("content_format"), + "relatable": row.get("relatable"), + "politics_news": row.get("politics_news"), + "src_fwd_pct": row.get("src_fwd_pct"), + "factors": factors, + "scores": scores, + "ranks": ranks[i], + "picks": picks, + "picked_by": picked_by, + # legacy flat keys (the only shape on rows before this refactor) + **factors, + "final_score": scores[picked_by], + "like_volume_enabled": "like_volume_factor" in SCORES[picked_by], + **{f"score_{n}": scores[n] for n in _LEGACY_SHADOWS}, + **{f"shadow_{n}_rank": ranks[i][n] for n in _LEGACY_SHADOWS if n in picks}, + **{f"shadow_{n}_pick_meme_id": picks[n] for n in _LEGACY_SHADOWS if n in picks}, + } + ) - When ``CROSSPOST_RU_MEME_LIKE_VOLUME_ENABLED`` (default ON), uses - ``score_version=4``: v2 factors × ``ln(nlikes+1)`` so high bot-exposure - memes win within source prior (offline H6). Kill switch → v2. - """ - like_vol = _ru_like_volume_ranker_enabled() - return await _get_next_meme_for_channel( - "tgchannelru", - log_top_n=log_top_n, - score_version=(SCORE_VERSION_MEME_LIKE_VOLUME if like_vol else SCORE_VERSION_V2), - like_volume_enabled=like_vol, - ) + return { + "channel": channel, + "picked_meme_id": picks[picked_by], + "score_version": SCORE_VERSION[picked_by], + "median_signal": _round(top[0].get("median_signal"), 4), + "pool_size": pool_size, + "candidates": log_candidates, + } -async def get_next_meme_for_tgchannelen( - log_top_n: int = 5, +async def get_next_meme_for_channel( + channel: str, log_top_n: int = 5 ) -> tuple[dict[str, Any] | None, dict[str, Any] | None]: - """Same as :func:`get_next_meme_for_tgchannelru` but for @fast_food_memes (EN). + """Pick the crosspost meme for ``channel`` (tgchannelru / tgchannelen). - EN stays on score_version=2 (no like-volume factor) until offline gate passes. + Returns ``(picked_meme, decision_log)``: ``picked_meme`` is a narrow dict for + ``MemeData(**...)``; ``decision_log`` is kwargs for ``log_ranker_decision`` + (its ``score_version`` is also what ``log_meme_sent`` records). + Both ``None`` when no candidate passes the filters. """ - return await _get_next_meme_for_channel( - "tgchannelen", - log_top_n=log_top_n, - score_version=SCORE_VERSION_V2, - like_volume_enabled=False, - ) + params = {"channel": channel, "language_code": CHANNEL_PARAMS[channel]["language_code"]} + rows = await fetch_all(text(_POOL_QUERY), params) + if not rows: + return None, None + pre = PRESELECT[channel] + top = heapq.nsmallest(log_top_n, rows, key=lambda r: (-score(r, channel, pre), r["id"])) + # Shadow inputs must never block posting. + try: + await _attach_shadow_inputs(top, channel) + except Exception: + logger.warning("crosspost shadow inputs failed", exc_info=True) + decision = _build_decision_log(channel, top, pool_size=len(rows)) + return await _picked_meme_dict(decision["picked_meme_id"]), decision diff --git a/src/crossposting/shadow_score.py b/src/crossposting/shadow_score.py deleted file mode 100644 index 8e9e4755..00000000 --- a/src/crossposting/shadow_score.py +++ /dev/null @@ -1,65 +0,0 @@ -"""Crosspost v5 shadow scorer — log only, never changes the production pick. - -From the 2026-10-30 walk-forward study (target zf = log1p(forwards) minus -channel×week mean; see experiments/2026-10-30-crosspost-v5-shadow.md): - - RU: score_v5 = ln(nlikes+1) * theta_src * exp(0.5 * user_taste_z) - EN: score_v5 = theta_src * exp(user_taste_z) (like volume hurts on EN) - -theta_src and user_taste_z come from ``_V5_SHADOW_QUERY`` in service.py. -Version string is frozen for decision_log analytics. -""" - -from __future__ import annotations - -import math -from typing import Any - -SHADOW_V5_VERSION = "v5_src_beta_taste_v1" - - -def score_v5( - channel: str, - nlikes: int | float | None, - theta_src: float, - user_taste_z: float | None, -) -> float: - theta = float(theta_src) - taste = float(user_taste_z or 0.0) - if channel == "tgchannelru": - return math.log(float(nlikes or 0) + 1.0) * theta * math.exp(0.5 * taste) - return theta * math.exp(taste) - - -def content_mult_v1(content_format: str | None, relatable: bool | None) -> float: - """RU blind-label study 2026-10-30: relatable and tweet screenshots ≈ −0.1 zf each. - - Tags come from Describe Memes (meme.ocr_result); missing tags → 1.0. - See experiments/2026-10-31-crosspost-content-tags.md. - """ - return 0.85 ** (int(relatable is True) + int(content_format == "tweet_screenshot")) - - -def src_fwd_mult(src_fwd_pct: float | None) -> float: - """score_v5f = score_v5 * (0.5 + src_fwd_pct); missing pct → 1.0. - - src_fwd_pct: within-source percentile of the source post's forwards/views - (Telethon read at ~20–48h). Study 2026-10-31: Spearman with our crosspost - forwards +0.52 RU / +0.28 EN. See - experiments/2026-11-01-crosspost-src-forward-rate.md. - """ - return 1.0 if src_fwd_pct is None else 0.5 + float(src_fwd_pct) - - -def attach_v5_ranks(candidates: list[dict[str, Any]], version: str = "v5") -> list[dict[str, Any]]: - """Add shadow_{version}_rank (1=best) + shadow_{version}_pick_meme_id without reordering.""" - key = f"score_{version}" - if not candidates or any(c.get(key) is None for c in candidates): - return candidates - order = sorted(range(len(candidates)), key=lambda i: -candidates[i][key]) - for rank, i in enumerate(order, start=1): - candidates[i][f"shadow_{version}_rank"] = rank - pick = candidates[order[0]]["meme_id"] - for c in candidates: - c[f"shadow_{version}_pick_meme_id"] = pick - return candidates diff --git a/src/flows/crossposting/meme.py b/src/flows/crossposting/meme.py index a1926f53..7e0d81a5 100644 --- a/src/flows/crossposting/meme.py +++ b/src/flows/crossposting/meme.py @@ -5,13 +5,9 @@ from prefect import flow, get_run_logger from telegram.constants import ParseMode -from src.config import settings from src.crossposting.constants import Channel from src.crossposting.service import ( - SCORE_VERSION_MEME_LIKE_VOLUME, - SCORE_VERSION_V2, - get_next_meme_for_tgchannelen, - get_next_meme_for_tgchannelru, + get_next_meme_for_channel, log_meme_sent, log_ranker_decision, posted_recently, @@ -217,7 +213,7 @@ async def post_meme_to_tgchannelen(): logger.warning("TG Channel EN posted <1h ago, skipping catch-up slot") return - meme_data, decision = await get_next_meme_for_tgchannelen() + meme_data, decision = await get_next_meme_for_channel(Channel.TG_CHANNEL_EN.value) if meme_data is None: logger.warning("No qualifying meme for TG Channel EN, skipping slot") return @@ -243,7 +239,7 @@ async def post_meme_to_tgchannelen(): Channel.TG_CHANNEL_EN, telegram_message_id=msg.message_id, caption_text=caption_text, - score_version=SCORE_VERSION_V2, + score_version=decision["score_version"], ) await update_meme(next_meme.id, status=MemeStatus.PUBLISHED) await _safe_refresh_crosspost_stats(Channel.TG_CHANNEL_EN, next_meme.id, msg.message_id, logger) @@ -277,7 +273,7 @@ async def post_meme_to_tgchannelru(): logger.warning("TG Channel RU posted <1h ago, skipping catch-up slot") return - meme_data, decision = await get_next_meme_for_tgchannelru() + meme_data, decision = await get_next_meme_for_channel(Channel.TG_CHANNEL_RU.value) if meme_data is None: logger.warning("No qualifying meme for TG Channel RU, skipping slot") return @@ -299,17 +295,12 @@ async def post_meme_to_tgchannelru(): bot, TELEGRAM_CHANNEL_RU_CHAT_ID, next_meme, reply_markup=None ) - ru_score_version = ( - SCORE_VERSION_MEME_LIKE_VOLUME - if settings.CROSSPOST_RU_MEME_LIKE_VOLUME_ENABLED - else SCORE_VERSION_V2 - ) await log_meme_sent( next_meme.id, Channel.TG_CHANNEL_RU, telegram_message_id=msg.message_id, caption_text=caption_text, - score_version=ru_score_version, + score_version=decision["score_version"], ) await update_meme(next_meme.id, status=MemeStatus.PUBLISHED) await _safe_refresh_crosspost_stats(Channel.TG_CHANNEL_RU, next_meme.id, msg.message_id, logger) diff --git a/tests/crossposting/test_scoring.py b/tests/crossposting/test_scoring.py new file mode 100644 index 00000000..90e56633 --- /dev/null +++ b/tests/crossposting/test_scoring.py @@ -0,0 +1,93 @@ +"""Crosspost scoring — pure functions, no DB.""" + +import math + +from src.crossposting import scoring +from src.crossposting.scoring import FACTORS, score +from src.crossposting.service import _build_decision_log + + +def _cand(mid, nlikes, **kw): + return { + "id": mid, + "nlikes": nlikes, + "ndislikes": 10, + "nmemes_sent": 40, + "raw_impr_rank": 1, + "age_days": 3, + "invited_count": 0, + "src_signal": 10.0, + "median_signal": 10.0, + "has_caption": False, + "meme_source_id": 1, + **kw, + } + + +def test_v5_ru_and_en(): + r = _cand(1, 10, theta_src=1.5, user_taste_z=0.4) + assert math.isclose(score(r, "tgchannelru", "v5"), math.log(11) * 1.5 * math.exp(0.2)) + # EN ignores like volume, full-strength taste + assert math.isclose(score(r, "tgchannelen", "v5"), 1.5 * math.exp(0.4)) + assert score(_cand(1, 999, theta_src=2.0), "tgchannelen", "v5") == 2.0 + assert score(_cand(1, 10), "tgchannelru", "v5") is None # no shadow inputs + + +def test_content_and_src_fwd_mults(): + content, fwd = FACTORS["content_mult_v1"], FACTORS["src_fwd_mult"] + assert content(_cand(1, 5), "tgchannelru") == 1.0 + assert math.isclose(content(_cand(1, 5, content_format="tweet_screenshot"), ""), 0.85) + assert math.isclose( + content(_cand(1, 5, content_format="tweet_screenshot", relatable=True), ""), 0.85**2 + ) + assert fwd(_cand(1, 5), "") == 1.0 + assert fwd(_cand(1, 5, src_fwd_pct=0.0), "") == 0.5 + assert fwd(_cand(1, 5, src_fwd_pct=1.0), "") == 1.5 + + +def test_factors_mirror_old_sql_quirks(): + # int-typed CASE bind: the "0.8/0.5" impression penalty was always 0 in prod + assert FACTORS["impr_factor"](_cand(1, 5, raw_impr_rank=2), "tgchannelru") == 0.0 + assert FACTORS["impr_factor"](_cand(1, 5, raw_impr_rank=1), "tgchannelen") == 1.0 + f = FACTORS["src_quality_mult"] + assert f(_cand(1, 5, src_signal=None), "") == 0.5 # GREATEST(0.5, NULL) = 0.5 + assert f(_cand(1, 5, median_signal=0), "") == 0.5 # NULLIF(median, 0) + assert f(_cand(1, 5, src_signal=0.0), "") == 0.5 + assert f(_cand(1, 5, src_signal=100.0), "") == 2.0 + assert f(_cand(1, 5, src_signal=15.0), "") == 1.5 + + +def test_decision_log_ranks_picks_and_legacy_keys(): + tags = {"content_format": "tweet_screenshot", "relatable": True} + cands = [ + _cand(101, 100, theta_src=1.0, user_taste_z=0.0, src_fwd_pct=0.1, **tags), + _cand(102, 100, theta_src=0.9, user_taste_z=0.0, src_fwd_pct=0.9), + _cand(103, 100, theta_src=0.8, user_taste_z=0.0), + ] + log = _build_decision_log("tgchannelru", cands, pool_size=50) + a, b, c = log["candidates"] + assert log["picked_meme_id"] == 101 and log["score_version"] == 4 # live = v4, ties → id + assert a["picked_by"] == "v4" and a["picks"]["v5"] == 101 + assert a["picks"] == {"v2": 101, "v4": 101, "v5": 101, "v5c": 102, "v5f": 102} + assert (a["ranks"]["v5f"], b["ranks"]["v5f"], c["ranks"]["v5f"]) == (3, 1, 2) + # legacy flat keys used by docs/analyst readouts + assert a["score_v5c"] == a["scores"]["v5c"] and a["final_score"] == a["scores"]["v4"] + assert a["shadow_v5f_rank"] == 3 and a["shadow_v5c_pick_meme_id"] == 102 + assert a["lr_factor"] == a["factors"]["lr_factor"] and a["like_volume_enabled"] is True + assert c["src_fwd_pct"] is None and c["score_v5f"] == c["score_v5"] + + +def test_live_shadow_falls_back_to_preselect_without_inputs(monkeypatch): + monkeypatch.setitem(scoring.LIVE, "tgchannelen", "v5") + log = _build_decision_log("tgchannelen", [_cand(101, 100)], pool_size=1) + top = log["candidates"][0] + assert top["picked_by"] == "v2" and log["score_version"] == 2 + assert top["score_v5"] is None and "shadow_v5_rank" not in top + + +def test_live_shadow_picks_when_promoted(monkeypatch): + monkeypatch.setitem(scoring.LIVE, "tgchannelen", "v5") + cands = [_cand(101, 100, theta_src=0.5), _cand(102, 20, theta_src=2.0)] + log = _build_decision_log("tgchannelen", cands, pool_size=2) + assert log["picked_meme_id"] == 102 and log["score_version"] == 5 + assert log["candidates"][0]["final_score"] == log["candidates"][0]["score_v5"] diff --git a/tests/crossposting/test_shadow_score.py b/tests/crossposting/test_shadow_score.py deleted file mode 100644 index f1c7aa14..00000000 --- a/tests/crossposting/test_shadow_score.py +++ /dev/null @@ -1,122 +0,0 @@ -"""v5 shadow score — log only, pure functions.""" - -import math - -from src.crossposting.service import _build_decision_log -from src.crossposting.shadow_score import ( - SHADOW_V5_VERSION, - attach_v5_ranks, - content_mult_v1, - score_v5, - src_fwd_mult, -) - - -def test_score_v5_ru_and_en(): - assert math.isclose(score_v5("tgchannelru", 10, 1.5, 0.4), math.log(11) * 1.5 * math.exp(0.2)) - # EN ignores like volume, full-strength taste - assert math.isclose(score_v5("tgchannelen", 10, 1.5, 0.4), 1.5 * math.exp(0.4)) - assert math.isclose(score_v5("tgchannelen", 999, 2.0, None), 2.0) - - -def test_attach_v5_ranks_keeps_order_and_picks_best(): - cands = [ - {"rank": 1, "meme_id": 1, "score_v5": 1.0}, - {"rank": 2, "meme_id": 2, "score_v5": 9.0}, - {"rank": 3, "meme_id": 3, "score_v5": 3.0}, - ] - out = attach_v5_ranks(cands) - assert [c["meme_id"] for c in out] == [1, 2, 3] - assert [c["shadow_v5_rank"] for c in out] == [3, 1, 2] - assert {c["shadow_v5_pick_meme_id"] for c in out} == {2} - - -def _cand(mid, nlikes, **kw): - return { - "id": mid, - "nlikes": nlikes, - "ndislikes": 10, - "nmemes_sent": 40, - "raw_impr_rank": 1, - "age_days": 3, - "invited_count": 0, - "src_signal": 10.0, - "median_signal": 10.0, - "caption": None, - "meme_source_id": 1, - "candidate_pool_size": 50, - **kw, - } - - -def test_build_decision_log_v5_fields(): - cands = [ - _cand(101, 100, theta_src=0.5, user_taste_z=-1.0), - _cand(102, 20, theta_src=2.0, user_taste_z=1.0), - ] - log = _build_decision_log("tgchannelen", 2, cands) - top = log["candidates"][0] - assert top["rank"] == 1 and top["meme_id"] == 101 # production order preserved - assert top["shadow_version"] == SHADOW_V5_VERSION - assert top["shadow_v5_pick_meme_id"] == 102 - assert "taste_boost_shadow" not in top - - -def test_build_decision_log_without_v5_inputs_logs_none(): - log = _build_decision_log("tgchannelru", 4, [_cand(101, 100)], like_volume_enabled=True) - top = log["candidates"][0] - assert top["score_v5"] is None - assert "shadow_v5_rank" not in top - - -def test_content_mult_v1(): - assert content_mult_v1(None, None) == 1.0 - assert content_mult_v1("single_caption", False) == 1.0 - assert math.isclose(content_mult_v1("tweet_screenshot", None), 0.85) - assert math.isclose(content_mult_v1("single_caption", True), 0.85) - assert math.isclose(content_mult_v1("tweet_screenshot", True), 0.85**2) - - -def test_build_decision_log_v5c_shadow(): - tags = {"content_format": "tweet_screenshot", "relatable": True, "politics_news": False} - cands = [ - _cand(101, 100, theta_src=1.0, user_taste_z=0.0, **tags), - _cand(102, 100, theta_src=0.9, user_taste_z=0.0), - ] - log = _build_decision_log("tgchannelru", 4, cands, like_volume_enabled=True) - top, second = log["candidates"] - assert log["picked_meme_id"] == 101 and top["meme_id"] == 101 # pick unchanged - assert {k: top[k] for k in tags} == tags - assert math.isclose(top["score_v5c"], top["score_v5"] * 0.85**2, rel_tol=1e-5) - assert second["content_mult_v1"] == 1.0 and second["score_v5c"] == second["score_v5"] - assert top["shadow_v5_pick_meme_id"] == 101 # theta 1.0 > 0.9 - assert top["shadow_v5c_pick_meme_id"] == 102 # tags flip the shadow pick - assert (top["shadow_v5c_rank"], second["shadow_v5c_rank"]) == (2, 1) - - -def test_build_decision_log_v5c_none_without_v5(): - top = _build_decision_log("tgchannelru", 4, [_cand(101, 100)])["candidates"][0] - assert top["content_mult_v1"] == 1.0 and top["score_v5c"] is None - assert "shadow_v5c_rank" not in top - - -def test_src_fwd_mult(): - assert src_fwd_mult(None) == 1.0 - assert src_fwd_mult(0.0) == 0.5 - assert src_fwd_mult(1.0) == 1.5 - - -def test_build_decision_log_v5f_shadow(): - cands = [ - _cand(101, 100, theta_src=1.0, user_taste_z=0.0, src_fwd_pct=0.1), - _cand(102, 100, theta_src=0.9, user_taste_z=0.0, src_fwd_pct=0.9), - _cand(103, 100, theta_src=0.8, user_taste_z=0.0), - ] - log = _build_decision_log("tgchannelru", 4, cands, like_volume_enabled=True) - a, b, c = log["candidates"] - assert log["picked_meme_id"] == 101 # production pick unchanged - assert math.isclose(a["score_v5f"], a["score_v5"] * 0.6, rel_tol=1e-5) - assert c["src_fwd_pct"] is None and c["score_v5f"] == c["score_v5"] - assert a["shadow_v5_pick_meme_id"] == 101 - assert a["shadow_v5f_pick_meme_id"] == 102 - assert (a["shadow_v5f_rank"], b["shadow_v5f_rank"], c["shadow_v5f_rank"]) == (3, 1, 2) diff --git a/tests/flows/crossposting/test_catch_up_guard.py b/tests/flows/crossposting/test_catch_up_guard.py index 45414b5c..6d70cfe6 100644 --- a/tests/flows/crossposting/test_catch_up_guard.py +++ b/tests/flows/crossposting/test_catch_up_guard.py @@ -15,17 +15,13 @@ def error(self, *a, **k): ... @pytest.mark.asyncio @pytest.mark.parametrize( - "flow, picker", - [ - (flow_mod.post_meme_to_tgchannelru, "get_next_meme_for_tgchannelru"), - (flow_mod.post_meme_to_tgchannelen, "get_next_meme_for_tgchannelen"), - ], + "flow", [flow_mod.post_meme_to_tgchannelru, flow_mod.post_meme_to_tgchannelen] ) -async def test_skips_when_posted_recently(monkeypatch, flow, picker): +async def test_skips_when_posted_recently(monkeypatch, flow): monkeypatch.setattr(flow_mod, "get_run_logger", lambda: _Log()) monkeypatch.setattr(flow_mod, "posted_recently", AsyncMock(return_value=True)) pick = AsyncMock(return_value=(None, None)) - monkeypatch.setattr(flow_mod, picker, pick) + monkeypatch.setattr(flow_mod, "get_next_meme_for_channel", pick) await flow.fn() pick.assert_not_awaited() @@ -35,6 +31,6 @@ async def test_runs_picker_when_not_posted_recently(monkeypatch): monkeypatch.setattr(flow_mod, "get_run_logger", lambda: _Log()) monkeypatch.setattr(flow_mod, "posted_recently", AsyncMock(return_value=False)) pick = AsyncMock(return_value=(None, None)) - monkeypatch.setattr(flow_mod, "get_next_meme_for_tgchannelru", pick) + monkeypatch.setattr(flow_mod, "get_next_meme_for_channel", pick) await flow_mod.post_meme_to_tgchannelru.fn() pick.assert_awaited_once() diff --git a/tests/test_crossposting_meme.py b/tests/test_crossposting_meme.py index ab661a57..7860a520 100644 --- a/tests/test_crossposting_meme.py +++ b/tests/test_crossposting_meme.py @@ -1,3 +1,4 @@ +import math from datetime import UTC, datetime, timedelta import pytest @@ -5,14 +6,9 @@ from sqlalchemy import delete, text from sqlalchemy.dialects.postgresql import insert -from src.crossposting.service import ( - SCORE_VERSION_MEME_LIKE_VOLUME, - SCORE_VERSION_V2, - _compute_score_breakdown, - get_next_meme_for_tgchannelen, - get_next_meme_for_tgchannelru, - log_ranker_decision, -) +from src.crossposting import scoring +from src.crossposting.scoring import score +from src.crossposting.service import get_next_meme_for_channel, log_ranker_decision from src.database import ( crossposting, crossposting_decision_log, @@ -80,14 +76,11 @@ def test_like_volume_factor_raises_score_for_more_likes(): "invited_count": 0, "src_signal": 10.0, "median_signal": 10.0, - "caption": None, + "has_caption": False, } - low = _compute_score_breakdown({**base, "nlikes": 10}, "tgchannelru", like_volume_enabled=True) - high = _compute_score_breakdown({**base, "nlikes": 80}, "tgchannelru", like_volume_enabled=True) - off = _compute_score_breakdown({**base, "nlikes": 80}, "tgchannelru", like_volume_enabled=False) - assert high["like_volume_factor"] > low["like_volume_factor"] - assert high["final_score"] > low["final_score"] - assert off["like_volume_factor"] == 1.0 + v4 = score({**base, "nlikes": 80}, "tgchannelru", "v4") + assert v4 > score(base, "tgchannelru", "v4") + assert v4 / score({**base, "nlikes": 80}, "tgchannelru", "v2") == pytest.approx(math.log(81)) # ── Integration tests for get_next_meme_for_tgchannelru ranker ─────────── @@ -171,7 +164,7 @@ async def test_select_excludes_source_posted_within_24h(clean_xpost): await create_meme_stats(conn, meme_id=10004, nlikes=10, ndislikes=2) await conn.commit() - picked, decision = await get_next_meme_for_tgchannelru() + picked, decision = await get_next_meme_for_channel("tgchannelru") assert picked is not None, "Source B candidate should remain selectable" assert picked["id"] == 10004, ( "diversity cap must exclude source 10001 and prefer source 10003 candidate" @@ -179,16 +172,12 @@ async def test_select_excludes_source_posted_within_24h(clean_xpost): assert decision is not None assert decision["picked_meme_id"] == 10004 assert decision["candidates"][0]["meme_id"] == 10004 - assert decision["score_version"] == SCORE_VERSION_MEME_LIKE_VOLUME + assert decision["score_version"] == 4 @pytest.mark.asyncio -async def test_like_volume_prefers_higher_nlikes_same_source(clean_xpost, monkeypatch): +async def test_like_volume_prefers_higher_nlikes_same_source(clean_xpost): """score_version=4: ln(nlikes+1) re-ranks within equal quality.""" - monkeypatch.setattr( - "src.crossposting.service.settings.CROSSPOST_RU_MEME_LIKE_VOLUME_ENABLED", - True, - ) async with engine.connect() as conn: await create_meme_source(conn, id=10500, language_code="ru") await create_meme( @@ -202,10 +191,10 @@ async def test_like_volume_prefers_higher_nlikes_same_source(clean_xpost, monkey await create_meme_stats(conn, meme_id=10502, nlikes=80, ndislikes=20, nmemes_sent=200) await conn.commit() - picked, decision = await get_next_meme_for_tgchannelru() + picked, decision = await get_next_meme_for_channel("tgchannelru") assert picked is not None assert picked["id"] == 10502 - assert decision["score_version"] == SCORE_VERSION_MEME_LIKE_VOLUME + assert decision["score_version"] == 4 assert ( decision["candidates"][0]["like_volume_factor"] > decision["candidates"][1]["like_volume_factor"] @@ -213,11 +202,8 @@ async def test_like_volume_prefers_higher_nlikes_same_source(clean_xpost, monkey @pytest.mark.asyncio -async def test_ru_kill_switch_falls_back_to_v2(clean_xpost, monkeypatch): - monkeypatch.setattr( - "src.crossposting.service.settings.CROSSPOST_RU_MEME_LIKE_VOLUME_ENABLED", - False, - ) +async def test_live_edit_switches_picking_score(clean_xpost, monkeypatch): + monkeypatch.setitem(scoring.LIVE, "tgchannelru", "v2") async with engine.connect() as conn: await create_meme_source(conn, id=10510, language_code="ru") await create_meme( @@ -226,11 +212,11 @@ async def test_ru_kill_switch_falls_back_to_v2(clean_xpost, monkeypatch): await create_meme_stats(conn, meme_id=10511, nlikes=20, ndislikes=5) await conn.commit() - picked, decision = await get_next_meme_for_tgchannelru() + picked, decision = await get_next_meme_for_channel("tgchannelru") assert picked is not None - assert decision["score_version"] == SCORE_VERSION_V2 + assert decision["score_version"] == 2 + assert decision["candidates"][0]["picked_by"] == "v2" assert decision["candidates"][0]["like_volume_enabled"] is False - assert decision["candidates"][0]["like_volume_factor"] == 1.0 @pytest.mark.asyncio @@ -255,7 +241,7 @@ async def test_select_returns_none_when_all_filtered(clean_xpost): await _insert_crossposting(conn, "tgchannelru", 10013, hours_ago=72, views=100, forwards=5) await conn.commit() - picked, decision = await get_next_meme_for_tgchannelru() + picked, decision = await get_next_meme_for_channel("tgchannelru") assert picked is None assert decision is None @@ -307,7 +293,7 @@ async def test_source_quality_applied_when_n_above_threshold(clean_xpost): await create_meme_stats(conn, meme_id=10250, nlikes=10, ndislikes=2) await conn.commit() - picked, decision = await get_next_meme_for_tgchannelru() + picked, decision = await get_next_meme_for_channel("tgchannelru") assert picked is not None assert picked["id"] == 10150, ( "good_source candidate should outrank bad_source via SQ multiplier" @@ -324,7 +310,7 @@ async def test_source_quality_applied_when_n_above_threshold(clean_xpost): @pytest.mark.asyncio -async def test_source_quality_neutral_when_no_snapshots(clean_xpost): +async def test_source_quality_half_when_no_snapshots(clean_xpost): async with engine.connect() as conn: await create_meme_source(conn, id=10300, language_code="ru") await create_meme( @@ -333,14 +319,15 @@ async def test_source_quality_neutral_when_no_snapshots(clean_xpost): await create_meme_stats(conn, meme_id=10301, nlikes=10, ndislikes=2) await conn.commit() - picked, decision = await get_next_meme_for_tgchannelru() + picked, decision = await get_next_meme_for_channel("tgchannelru") assert picked is not None assert picked["id"] == 10301 assert decision is not None - # Cold-start: src_signal None and src_quality_mult falls through to neutral 1.0 + # Cold-start: src_signal None → 0.5 (what the pre-refactor SQL ORDER BY applied: + # GREATEST(0.5, NULL) = 0.5; the old logged breakdown wrongly said 1.0) only_candidate = decision["candidates"][0] assert only_candidate["src_signal"] is None - assert only_candidate["src_quality_mult"] == 1.0 + assert only_candidate["src_quality_mult"] == 0.5 @pytest.mark.asyncio @@ -418,7 +405,7 @@ async def test_decision_log_shadow_counts_pre_posting_share_clicks(clean_xpost): ) await conn.commit() - picked, decision = await get_next_meme_for_tgchannelru() + picked, decision = await get_next_meme_for_channel("tgchannelru") assert picked is not None assert picked["id"] == 10351 assert decision is not None @@ -447,7 +434,7 @@ async def test_v5_shadow_logs_content_tags_from_ocr_result(clean_xpost): await create_meme_stats(conn, meme_id=10702, nlikes=10, ndislikes=2) await conn.commit() - picked, decision = await get_next_meme_for_tgchannelru() + picked, decision = await get_next_meme_for_channel("tgchannelru") assert picked["id"] == 10702 cand = decision["candidates"][0] assert (cand["content_format"], cand["relatable"], cand["politics_news"]) == ( @@ -495,7 +482,7 @@ async def test_v5f_shadow_logs_src_fwd_pct(clean_xpost): await create_meme_stats(conn, meme_id=10802, nlikes=10, ndislikes=2) await conn.commit() - picked, decision = await get_next_meme_for_tgchannelru() + picked, decision = await get_next_meme_for_channel("tgchannelru") assert picked["id"] == 10802 cand = decision["candidates"][0] assert cand["src_fwd_pct"] == 1.0 @@ -513,7 +500,7 @@ async def test_en_ranker_decision_log_has_shadow_share_fields(clean_xpost): await create_meme_stats(conn, meme_id=10361, nlikes=10, ndislikes=2) await conn.commit() - picked, decision = await get_next_meme_for_tgchannelen() + picked, decision = await get_next_meme_for_channel("tgchannelen") assert picked is not None assert picked["id"] == 10361 assert decision is not None @@ -528,7 +515,7 @@ async def test_ranker_decision_log_records_top5(clean_xpost): async with engine.connect() as conn: await create_meme_source(conn, id=10400, language_code="ru") # Create 7 eligible candidates from the same source (so all pass filters - # and rank by score). LIMIT in get_next_meme_for_tgchannelru is 5 → top-5 logged. + # and rank by score). log_top_n defaults to 5 → top-5 logged. for i in range(7): mid = 10401 + i await create_meme( @@ -543,7 +530,7 @@ async def test_ranker_decision_log_records_top5(clean_xpost): await create_meme_stats(conn, meme_id=mid, nlikes=20 - i, ndislikes=2) await conn.commit() - picked, decision = await get_next_meme_for_tgchannelru() + picked, decision = await get_next_meme_for_channel("tgchannelru") assert picked is not None assert decision is not None # Persist via the actual logger (exercising the SQL path) @@ -568,7 +555,7 @@ async def test_ranker_decision_log_records_top5(clean_xpost): row = rows[0] assert row["channel"] == "tgchannelru" assert row["picked_meme_id"] == picked["id"] - assert row["score_version"] == SCORE_VERSION_MEME_LIKE_VOLUME + assert row["score_version"] == 4 assert row["candidate_pool_size"] == 7 # 7 eligible memes assert len(row["candidates"]) == 5 # top-5 logged (LIMIT 5) # Each entry has the documented score breakdown keys