| """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] |
|
|
|
|
| def test_reranker_dominates_blend() -> None: |
| full = ensemble(_pool(), Settings(ensemble=EnsembleSettings(top_n=4))) |
| |
| assert full.sort("rank")["candidate_id"].to_list()[0] == "c1" |
|
|
|
|
| def test_missing_signal_contributes_zero() -> None: |
| pool = _pool().drop("granite_score", "composite_score") |
| kept = ensemble(pool, Settings(ensemble=EnsembleSettings(top_n=4))) |
| assert kept["candidate_id"].to_list() == ["c1", "c2", "c3", "c4"] |
|
|
|
|
| def test_empty_pool_passthrough() -> None: |
| out = ensemble(_pool().head(0), Settings()) |
| assert out.height == 0 and {"final_score", "rank"} <= set(out.columns) |
|
|
