discovery: add apply_adventurousness — pure popularity-penalty re-rank (aurral-style)

Both Discover rec rows already exclude what you own/watch, so novelty is baked in; the missing lever
is a popularity penalty. apply_adventurousness(items, level) re-ranks dicts (score + optional 0-100
popularity) so globally-popular candidates sink as the dial rises. Pure + reusable across both rec
rows. level<=0 returns the input order unchanged (a copy) — fully additive, no regression; 1.0 applies
the full penalty (a popularity-100 pick loses 70% of its score). Missing popularity is never penalised.

5 seam tests (no-op+copy, demotion, proportional penalty, missing-pop, clamping). 37 pass. Wiring
(scan stores popularity -> routes re-rank live -> Settings slider) is the next increment.
This commit is contained in:
BoulderBadgeDad 2026-06-29 16:03:16 -07:00
parent 052547c04a
commit f0a6d5e696
2 changed files with 86 additions and 0 deletions

View file

@ -39,6 +39,14 @@ def _positive_float(value: object, default: float = 1.0) -> float:
return f if f > 0 else default
def _coerce_float(value: object, default: float = 0.0) -> float:
"""Plain float coercion that keeps 0 / negatives (unlike _positive_float) — popularity can be 0."""
try:
return float(value) # type: ignore[arg-type]
except (TypeError, ValueError):
return default
def _get(row: object, attr: str):
"""Read a field from a dataclass row or a dict row."""
if isinstance(row, dict):
@ -234,6 +242,46 @@ def rank_recommended_artists(
return out[:limit]
# ── Adventurousness re-rank (aurral-style popularity penalty) ────────────────────────────────
# Both Discover rec rows already EXCLUDE what you own / watch, so "novelty" is baked in; the lever we
# were missing is a POPULARITY PENALTY. At higher adventurousness, globally-popular candidates are
# pushed down so the obscure / non-obvious picks surface. Pure + reusable across both rec rows.
_MAX_POP_PENALTY = 0.7 # at level 1.0 a popularity-100 candidate loses 70% of its score
def apply_adventurousness(
items: Sequence[dict],
level: object,
*,
score_key: str = "score",
pop_key: str = "popularity",
tiebreak_key: str = "seed_count",
) -> List[dict]:
"""Re-rank ``items`` (dicts with a numeric score + an optional 0100 popularity) by an
adventurousness-scaled popularity penalty. Returns a NEW list, most-adventurous first.
``level`` is clamped to 0..1. At ``level <= 0`` the input order is returned **unchanged** (a
copy), so the feature is fully additive / no-regression. Items missing a popularity are never
penalised. Adjusted score = ``score × (1 level × MAX_POP_PENALTY × popularity/100)``.
"""
lvl = max(0.0, min(1.0, _coerce_float(level, 0.0)))
if lvl <= 0.0:
return list(items)
def _adjusted(it: object) -> float:
score = _coerce_float(_get(it, score_key), 0.0)
pop = _get(it, pop_key)
if pop is None:
return score
pop_norm = max(0.0, min(1.0, _coerce_float(pop, 0.0) / 100.0))
return score * (1.0 - lvl * _MAX_POP_PENALTY * pop_norm)
return sorted(
items,
key=lambda it: (-_adjusted(it), -_coerce_float(_get(it, tiebreak_key), 0.0), _norm(_get(it, "name"))),
)
def aggregate_candidate_tracks(
recommended_artists: Sequence[RecommendedArtist],
top_tracks_by_artist: Dict[str, Sequence[dict]],

View file

@ -4,6 +4,7 @@ from __future__ import annotations
from core.discovery.listening_recommendations import (
aggregate_candidate_tracks,
apply_adventurousness,
build_recency_weighted_seeds,
choose_mix_fetch_source,
names_match,
@ -314,3 +315,40 @@ def test_rank_threading_changes_winner_within_a_seed():
ranked = rank_recommended_artists(seeds, grouped)
assert [r.name for r in ranked] == ["Close", "Far"]
assert ranked[0].score > ranked[1].score
# ── apply_adventurousness (aurral-style popularity-penalty re-rank) ───────────
def test_adventurousness_zero_is_noop_but_copies():
items = [{"name": "A", "score": 5.0, "popularity": 90},
{"name": "B", "score": 4.0, "popularity": 10}]
out = apply_adventurousness(items, 0.0)
assert [i["name"] for i in out] == ["A", "B"] # order unchanged
assert out == items # same content
assert out is not items # but a fresh list (additive)
def test_adventurousness_demotes_the_popular_one():
# Same score; at full adventurousness the obscure pick (pop 10) overtakes the giant (pop 95).
items = [{"name": "Giant", "score": 5.0, "popularity": 95},
{"name": "Obscure", "score": 5.0, "popularity": 10}]
assert [i["name"] for i in apply_adventurousness(items, 1.0)] == ["Obscure", "Giant"]
def test_adventurousness_penalty_is_proportional_not_absolute():
# A much stronger score still wins despite being more popular — the penalty scales the score.
items = [{"name": "StrongPopular", "score": 10.0, "popularity": 80},
{"name": "WeakObscure", "score": 1.0, "popularity": 0}]
assert apply_adventurousness(items, 0.5)[0]["name"] == "StrongPopular"
def test_adventurousness_missing_popularity_is_unpenalized():
items = [{"name": "Popular", "score": 5.0, "popularity": 100},
{"name": "NoPop", "score": 5.0}]
assert apply_adventurousness(items, 1.0)[0]["name"] == "NoPop"
def test_adventurousness_clamps_level():
items = [{"name": "A", "score": 5.0, "popularity": 100},
{"name": "B", "score": 5.0, "popularity": 0}]
assert [i["name"] for i in apply_adventurousness(items, 5.0)] == ["B", "A"] # >1 clamps to 1
assert [i["name"] for i in apply_adventurousness(items, -2.0)] == ["A", "B"] # <0 clamps to 0 (no-op)