"""Step 10 ensemble — weighted z-blend, top_n cut, 1-based rank, reranker dominance, empty guard.""" from __future__ import annotations import polars as pl from config import EnsembleSettings, Settings from rank import ensemble def _pool() -> pl.DataFrame: return pl.DataFrame({ "candidate_id": ["c1", "c2", "c3", "c4"], "reranker_score": [2.0, 1.0, 0.5, 0.1], "granite_score": [0.8, 0.7, 0.6, 0.5], "composite_score": [1.0, 2.0, 0.5, 0.2], }) def test_final_score_rank_and_top_n() -> None: kept = ensemble(_pool(), Settings(ensemble=EnsembleSettings(top_n=2))) assert kept.height == 2 assert {"final_score", "rank"} <= set(kept.columns) assert kept["rank"].to_list() == [1, 2] # 1-based, in sorted order def test_reranker_dominates_blend() -> None: full = ensemble(_pool(), Settings(ensemble=EnsembleSettings(top_n=4))) # c1 has the top reranker_score (weight 0.50) → should rank #1 assert full.sort("rank")["candidate_id"].to_list()[0] == "c1" def test_missing_signal_contributes_zero() -> None: pool = _pool().drop("granite_score", "composite_score") # only reranker drives the blend kept = ensemble(pool, Settings(ensemble=EnsembleSettings(top_n=4))) assert kept["candidate_id"].to_list() == ["c1", "c2", "c3", "c4"] # reranker desc order def test_empty_pool_passthrough() -> None: out = ensemble(_pool().head(0), Settings()) assert out.height == 0 and {"final_score", "rank"} <= set(out.columns)