feat(recommendations): collect demographics and advance age cohorts

services/recommender/pipu_rec/wave/store.py

@@ -161,6 +161,9 @@ def load_profile(conn: psycopg.Connection[Any], user_id: str) -> Profile:
     ).fetchone()
     if row and row[3] and len(scores) < ONBOARDING_UNTIL:
         apply_onboarding(conn, scores, row[3])
+        apply_demographics(conn, scores, row[3], user_id)
+        for tid in disliked:
+            scores.pop(tid, None)
     exploration = float(row[0]) if row else 0.15
     taste_cf = parse_vector(row[1]) if row else None
     bandit = {str(k): [float(x) for x in v] for k, v in (row[2] or {}).items()} if row else {}
@@ -212,6 +215,35 @@ def apply_onboarding(conn: psycopg.Connection[Any], scores: dict[str, float], ch
         scores.setdefault(tid, v)
 
 
+def apply_demographics(
+    conn: psycopg.Connection[Any], scores: dict[str, float], choice: dict[str, Any], user_id: str
+) -> None:
+    """A weak cold-start prior from actual likes of peers, without assumed musical stereotypes."""
+    gender, age = choice.get("gender"), choice.get("age_group")
+    if not gender and not age:
+        return
+    rows = conn.execute(
+        """
+        WITH own AS (
+            SELECT onboarding FROM user_profiles_rec WHERE user_id = %s
+        ), peers AS (
+            SELECT p.user_id FROM user_profiles_rec p JOIN users u ON u.id = p.user_id, own
+            WHERE p.user_id <> %s AND NOT u.is_blocked
+                AND (%s::text IS NULL OR p.onboarding->>'gender' = %s)
+                AND (%s::text IS NULL OR rec_age_group(p.onboarding) = rec_age_group(own.onboarding))
+        )
+        SELECT r.track_id::text, count(DISTINCT r.user_id) AS fans
+        FROM track_reactions r JOIN peers p ON p.user_id = r.user_id
+        WHERE r.reaction = 'like'
+        GROUP BY r.track_id HAVING count(DISTINCT r.user_id) >= 2
+        ORDER BY fans DESC, r.track_id LIMIT %s
+        """,
+        (user_id, user_id, gender, gender, age, ONBOARDING_LIMIT),
+    )
+    for tid, fans in rows:
+        scores.setdefault(tid, min(0.25, 0.05 * fans))
+
+
 def seed_tracks(conn: psycopg.Connection[Any], seed_type: str, seed_id: str) -> list[str]:
     """Треки, от которых запущена волна: трек, альбом или исполнитель."""
     queries = {

services/recommender/tests/test_db_integration.py

@@ -181,3 +181,30 @@ def test_impression_features_roundtrip(world: dict[str, Any]) -> None:
         assert any(r[0] == session and r[2] == "complete" for r in rows)
         result = train_and_save(conn, time.time())
         assert "skipped" in result["ranker"], "too few impressions — v1 stays"
+
+
+def test_demographic_prior_advances_age_and_preserves_own_feedback(world: dict[str, Any]) -> None:
+    user = uid()
+    with psycopg.connect(DSN, autocommit=True) as conn:
+        conn.execute("INSERT INTO users (id, username, password_hash) VALUES (%s, %s, 'x')", (user, f"cohort-{user[:6]}"))
+        # Target selected 13–19 twenty years ago; matching listeners are now 30–39.
+        conn.execute(
+            "INSERT INTO user_profiles_rec (user_id, onboarding) VALUES (%s, "
+            "jsonb_build_object('age_group', '13_19', 'gender', 'male', "
+            "'age_group_year', EXTRACT(YEAR FROM CURRENT_DATE)::int - 20))",
+            (user,),
+        )
+        for peer in world["users"][:3]:
+            conn.execute(
+                "INSERT INTO user_profiles_rec (user_id, onboarding) VALUES (%s, %s) "
+                "ON CONFLICT (user_id) DO UPDATE SET onboarding = EXCLUDED.onboarding",
+                (peer, json.dumps({"gender": "male", "age_group": "30_39"})),
+            )
+        profile = load_profile(conn, user)
+        track = world["tracks"][0]  # two matching listeners liked it
+        assert 0 < profile.scores[track] <= 0.25
+        assert not profile.played
+        conn.execute("INSERT INTO track_reactions (user_id, track_id, reaction) VALUES (%s, %s, 'dislike')", (user, track))
+        assert track not in load_profile(conn, user).scores
+        conn.execute("UPDATE user_profiles_rec SET onboarding = %s WHERE user_id = %s", (json.dumps({"gender": "female"}), user))
+        assert not load_profile(conn, user).scores, "no invented preferences when the cohort has no evidence"

services/recommender/tests/test_onboarding.py

@@ -0,0 +1,19 @@
+from unittest.mock import MagicMock
+
+from pipu_rec.wave.store import apply_demographics
+
+
+def test_missing_demographics_do_not_create_a_prior() -> None:
+    conn = MagicMock()
+    scores = {"own": 3.0}
+    apply_demographics(conn, scores, {}, "user")
+    conn.execute.assert_not_called()
+    assert scores == {"own": 3.0}
+
+
+def test_cohort_prior_is_weak_and_never_overwrites_personal_scores() -> None:
+    conn = MagicMock()
+    conn.execute.return_value = [("liked", 10), ("skipped", 10), ("peer", 100), ("small", 2)]
+    scores = {"liked": 3.0, "skipped": -1.0}
+    apply_demographics(conn, scores, {"gender": "male", "age_group": "13_19"}, "user")
+    assert scores == {"liked": 3.0, "skipped": -1.0, "peer": 0.25, "small": 0.1}