crossposting: one scoring module with named factors; LIVE picks per channel

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

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 |

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).

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).

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).

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)

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 (

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
 

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