From f40143cb694b6ec617baf38d2dfa26522970b767 Mon Sep 17 00:00:00 2001 From: Srilekha Date: Mon, 26 Oct 2026 12:48:15 +0530 Subject: [PATCH] feat(governance): enforce fail-closed verification, handle shortfall edge cases, pass 23/23 tests --- core/ranking_engine.py | 72 +++++++++++++++++++++++-------- test_multiplatform_ranking.py | 80 ++++++++++++++++++++++++++++++++++- 2 files changed, 134 insertions(+), 18 deletions(-) diff --git a/core/ranking_engine.py b/core/ranking_engine.py index 213580b..f60ce69 100644 --- a/core/ranking_engine.py +++ b/core/ranking_engine.py @@ -233,25 +233,40 @@ class HIFFIRankingEngine: return out - def apply_publication_eligibility_gates(self, df: pd.DataFrame) -> pd.DataFrame: + def apply_publication_eligibility_gates( + self, + df: pd.DataFrame, + return_metadata: bool = False + ) -> pd.DataFrame: """ - Constructs a publication ranking layer separate from raw mathematical scores: - - Top 10 requires all 3 platform signals (Tier 3). - - Top 50 requires at least 2 platform signals (Tier >= 2). - - Preserves hiffi_xp_score and hiffi_xp_rank unchanged. - - Produces consecutive, unique ranks in 'hiffi_xp_published_rank' (1 to N). + Constructs a publication ranking layer decoupled from raw scores: + - Top 10 requires all 3 platforms (Tier 3), strictly verified. + - Top 50 requires at least 2 platforms (Tier >= 2), strictly verified. + - Unverified or unindexed Instagram accounts do NOT contribute to evidence tiers. + - Fail-closed: missing verification column evaluates strictly to False. + - Explicitly tracks shortfalls and degrades gracefully without rank gaps or duplicate integers. """ out = df.copy() - # 1. Determine verified platform presence tier + # 1. Strict Signal Verification has_yt = out["production_hiffi_score"].notna() has_sp = out["spotify_monthly_listeners"].notna() - has_ig = out["instagram_followers"].notna() + + # Fail-closed Instagram verification check + verified_ig_statuses = {"AUTOMATED_VERIFIED", "MANUAL_VERIFIED"} + if "instagram_verification_status" in out.columns: + has_ig = ( + out["instagram_followers"].notna() + & out["instagram_verification_status"].isin(verified_ig_statuses) + ) + else: + has_ig = pd.Series(False, index=out.index) + out["platform_evidence_tier"] = ( has_yt.astype(int) + has_sp.astype(int) + has_ig.astype(int) ) - # 2. Sort the entire pool deterministically by raw composite score + # 2. Sort pool deterministically by raw composite score sort_cols = ["hiffi_xp_score", "production_hiffi_score", "spotify_monthly_listeners"] present_sort_cols = [c for c in sort_cols if c in out.columns] pool = out.sort_values( @@ -260,23 +275,44 @@ class HIFFIRankingEngine: ).copy() assigned_indices = [] + shortfall_audit = {"top_10_shortfall": False, "top_50_shortfall": False} - # Pass 1: Select Top 10 from Tier 3 candidates + # Pass 1: Top 10 from Tier 3 tier3_candidates = pool[pool["platform_evidence_tier"] == 3] top_10_indices = tier3_candidates.head(10).index.tolist() assigned_indices.extend(top_10_indices) - # Pass 2: Select Positions 11 to 50 from remaining Tier >= 2 candidates - remaining_pool = pool.drop(index=assigned_indices) - tier2_plus_candidates = remaining_pool[remaining_pool["platform_evidence_tier"] >= 2] - top_50_fill_indices = tier2_plus_candidates.head(40).index.tolist() - assigned_indices.extend(top_50_fill_indices) + if len(assigned_indices) < min(10, len(pool)): + shortfall_audit["top_10_shortfall"] = True + remaining_pool = pool.drop(index=assigned_indices) + shortfall = min(10 - len(assigned_indices), len(remaining_pool)) + backfill = remaining_pool[remaining_pool["platform_evidence_tier"] >= 2].head(shortfall) + assigned_indices.extend(backfill.index.tolist()) - # Pass 3: Fill Positions 51+ with all remaining artists by score order + # Complete degraded fallback if Tier 2+ is also exhausted + if len(assigned_indices) < min(10, len(pool)): + remaining_pool = pool.drop(index=assigned_indices) + shortfall = min(10 - len(assigned_indices), len(remaining_pool)) + assigned_indices.extend(remaining_pool.head(shortfall).index.tolist()) + + # Pass 2: Positions 11 to 50 from Tier >= 2 + remaining_pool = pool.drop(index=assigned_indices) + needed_for_50 = min(40, len(pool) - len(assigned_indices)) + tier2_plus_candidates = remaining_pool[remaining_pool["platform_evidence_tier"] >= 2] + top_50_fill = tier2_plus_candidates.head(needed_for_50).index.tolist() + assigned_indices.extend(top_50_fill) + + if len(assigned_indices) < min(50, len(pool)): + shortfall_audit["top_50_shortfall"] = True + remaining_pool = pool.drop(index=assigned_indices) + shortfall = min(50 - len(assigned_indices), len(remaining_pool)) + assigned_indices.extend(remaining_pool.head(shortfall).index.tolist()) + + # Pass 3: Positions 51+ with remaining artists by score order final_remaining = pool.drop(index=assigned_indices) assigned_indices.extend(final_remaining.index.tolist()) - # 3. Assign consecutive published ranks without touching raw scores + # 3. Assign contiguous published ranks out["hiffi_xp_published_rank"] = pd.NA rank_series = pd.Series( range(1, len(assigned_indices) + 1), @@ -286,4 +322,6 @@ class HIFFIRankingEngine: out.loc[assigned_indices, "hiffi_xp_published_rank"] = rank_series out["hiffi_xp_published_rank"] = out["hiffi_xp_published_rank"].astype(int) + if return_metadata: + return out, shortfall_audit return out \ No newline at end of file diff --git a/test_multiplatform_ranking.py b/test_multiplatform_ranking.py index bbe632a..37efb8c 100644 --- a/test_multiplatform_ranking.py +++ b/test_multiplatform_ranking.py @@ -377,4 +377,82 @@ def test_publication_eligibility_gates(dataset): # 4. Check ranks are strictly unique and consecutive 1 to N ranks = gated["hiffi_xp_published_rank"].sort_values().tolist() expected_ranks = list(range(1, len(gated) + 1)) - assert ranks == expected_ranks, "Published ranks contain gaps or duplicates." \ No newline at end of file + assert ranks == expected_ranks, "Published ranks contain gaps or duplicates." + +def test_publication_gate_blocks_unverified_identity(): + """Confirms that UNCERTAIN_NEEDS_REVIEW status does not count toward evidence tiers.""" + engine = HIFFIRankingEngine() + test_df = pd.DataFrame([ + { + "name": "Artist Pending Verification", + "production_hiffi_score": 95.0, + "spotify_monthly_listeners": 40_000_000, + "instagram_followers": 20_000_000, + "instagram_verification_status": "UNCERTAIN_NEEDS_REVIEW", + "hiffi_xp_score": 96.0, + "hiffi_xp_rank": 1 + }, + { + "name": "Artist Fully Verified", + "production_hiffi_score": 90.0, + "spotify_monthly_listeners": 35_000_000, + "instagram_followers": 15_000_000, + "instagram_verification_status": "AUTOMATED_VERIFIED", + "hiffi_xp_score": 91.0, + "hiffi_xp_rank": 2 + } + ]) + + gated = engine.apply_publication_eligibility_gates(test_df) + pending_row = gated[gated["name"] == "Artist Pending Verification"].iloc[0] + verified_row = gated[gated["name"] == "Artist Fully Verified"].iloc[0] + + assert pending_row["platform_evidence_tier"] == 2 + assert verified_row["platform_evidence_tier"] == 3 + + +def test_publication_gate_missing_column_fails_closed(): + """Confirms that omitting the verification column fails closed (gives 0 IG credit).""" + engine = HIFFIRankingEngine() + test_df = pd.DataFrame([ + { + "name": "Artist No Status Column", + "production_hiffi_score": 95.0, + "spotify_monthly_listeners": 40_000_000, + "instagram_followers": 20_000_000, + "hiffi_xp_score": 96.0, + "hiffi_xp_rank": 1 + } + ]) + + gated = engine.apply_publication_eligibility_gates(test_df) + assert gated.loc[0, "platform_evidence_tier"] == 2 + + +def test_publication_gate_zero_multiplatform_cohort(): + """Confirms an entirely Tier 1 cohort produces consecutive ranks and flags an audited shortfall.""" + engine = HIFFIRankingEngine() + + records = [ + { + "name": f"Solo Artist {i+1}", + "production_hiffi_score": 85.0 - i, + "spotify_monthly_listeners": np.nan, + "instagram_followers": np.nan, + "instagram_verification_status": "UNVERIFIED", + "hiffi_xp_score": 80.0 - i, + "hiffi_xp_rank": i + 1 + } + for i in range(25) + ] + + tier1_df = pd.DataFrame(records) + gated, audit = engine.apply_publication_eligibility_gates(tier1_df, return_metadata=True) + + # Must retain unique contiguous ordering 1 to 25 + assert gated["hiffi_xp_published_rank"].tolist() == list(range(1, 26)) + # All records remain Tier 1 + assert (gated["platform_evidence_tier"] == 1).all() + # Shortfall must be explicitly reported + assert audit["top_10_shortfall"] is True + assert audit["top_50_shortfall"] is True \ No newline at end of file