hiffi-artist-intelligence/core/ranking_engine.py

"""
core/ranking_engine.py
Ranks artists according to HIFFI formula stages:
- YouTube Production Baseline: Momentum (35%), Velocity (30%), Reach (20%), Consistency (15%)
- Breakout Velocity Index (BVI, 0-100) with 10k subscriber floor
- Multi-Platform HIFFI-XP Composite (0-100):
  * YouTube Production Score (50%)
  * Spotify Monthly Listeners (25%)
  * Spotify Daily Change / Velocity (5%)
  * Instagram Followers (20%)
  With zero-imputation compliance and proportional weight redistribution.
- Publication Eligibility Gates: Decoupled public chart assigning consecutive ranks
  while enforcing multi-platform evidence tiers for Top 10 and Top 50.
"""

from typing import Optional
import numpy as np
import pandas as pd


class HIFFIRankingEngine:

    @staticmethod
    def _log_scale(series: pd.Series) -> pd.Series:
        valid = series.dropna()
        log_vals = np.log10(1.0 + np.maximum(valid, 0.0))
        s_min, s_max = log_vals.min(), log_vals.max()
        if np.isclose(s_max, s_min) or pd.isna(s_min):
            return pd.Series(0.0, index=series.index)
        scaled = (log_vals - s_min) / (s_max - s_min + 1e-9)
        return scaled.reindex(series.index)

    @staticmethod
    def _min_max_scale(series: pd.Series) -> pd.Series:
        valid = series.dropna()
        s_min, s_max = valid.min(), valid.max()
        if np.isclose(s_max, s_min) or pd.isna(s_min):
            return pd.Series(0.0, index=series.index)
        scaled = (valid - s_min) / (s_max - s_min + 1e-9)
        return scaled.reindex(series.index)

    def calculate_breakout_velocity(self, df: pd.DataFrame) -> pd.Series:
        """
        Computes the Breakout Velocity Index (BVI, 0 - 100).
        Requires at least 10,000 subscribers and positive velocity to qualify.
        """
        out_bvi = pd.Series(np.nan, index=df.index, dtype=float)

        valid_mask = (
            df["subscriber_count"].notna()
            & (df["subscriber_count"] >= 10000)
            & df["daily_view_velocity"].notna()
            & (df["daily_view_velocity"] > 0.0)
            & (df["status_flag"] != "INACTIVE_OR_DORMANT")
            & (df["status_flag"] != "FAILED_LOOKUP")
        )

        if not valid_mask.any():
            return out_bvi

        subset = df[valid_mask]
        subs = subset["subscriber_count"].astype(float)
        daily_vel = subset["daily_view_velocity"].astype(float)
        momentum = subset["momentum_30d"].fillna(0.0).clip(lower=-100.0, upper=300.0)

        v_aud = daily_vel / np.maximum(subs, 1000.0)
        m_signal = (momentum + 100.0) / 400.0
        log_v_aud = np.log10(1.0 + v_aud)
        w_emerge = 1.0 / (1.0 + np.log10(1.0 + np.maximum(subs, 1.0)))

        raw_bvi = (0.65 * log_v_aud + 0.35 * m_signal) * w_emerge

        bvi_min = raw_bvi.min()
        bvi_max = raw_bvi.max()

        if np.isclose(bvi_max, bvi_min):
            scaled_bvi = pd.Series(50.0, index=subset.index)
        else:
            scaled_bvi = ((raw_bvi - bvi_min) / (bvi_max - bvi_min + 1e-9)) * 100.0

        out_bvi.loc[valid_mask] = scaled_bvi.round(2)
        return out_bvi

    def score_and_rank(
        self,
        df: pd.DataFrame,
        min_confidence: float = 0.50,
    ) -> pd.DataFrame:
        """Computes baseline YouTube production ranking."""
        out = df.copy()

        if "confidence_score" in out.columns:
            out["confidence_score"] = pd.to_numeric(
                out["confidence_score"].apply(
                    lambda x: x.get("composite_confidence", 0.0) if isinstance(x, dict) else x
                ),
                errors="coerce"
            ).fillna(0.0)

        out["breakout_velocity_index"] = self.calculate_breakout_velocity(out)

        has_baseline = (
            out["subscriber_count"].notna()
            & out["lifetime_views"].notna()
            & (out["status_flag"] != "FAILED_LOOKUP")
        )

        out["is_rank_eligible"] = (
            has_baseline
            & (out["confidence_score"] >= min_confidence)
            & (out["status_flag"].isin(["COMPLETE", "CATALOG_ACTIVE_NO_RECENT_UPLOAD"]))
        )

        filled_momentum = out["momentum_30d"].fillna(0.0).clip(lower=-100.0, upper=300.0)
        score_momentum = self._min_max_scale(filled_momentum)

        filled_velocity = out["daily_view_velocity"].fillna(0.0)
        score_velocity = self._log_scale(filled_velocity)

        log_subs = self._log_scale(out["subscriber_count"].fillna(0.0))
        log_lifetime = self._log_scale(out["lifetime_views"].fillna(0.0))
        score_reach = 0.60 * log_subs + 0.40 * log_lifetime

        score_consistency = self._log_scale(out["video_count"].fillna(0.0))

        hiffi_raw = (
            0.35 * score_momentum
            + 0.30 * score_velocity
            + 0.20 * score_reach
            + 0.15 * score_consistency
        ) * 100.0

        conf_multiplier = 0.8 + 0.2 * out["confidence_score"].clip(0.0, 1.0)
        out["production_hiffi_score"] = (hiffi_raw * conf_multiplier).round(2)

        eligible_mask = out["is_rank_eligible"]
        out["production_hiffi_rank"] = pd.NA
        out.loc[eligible_mask, "production_hiffi_rank"] = (
            out.loc[eligible_mask, "production_hiffi_score"]
            .rank(ascending=False, method="min")
            .astype(int)
        )

        bvi_eligible_mask = eligible_mask & out["breakout_velocity_index"].notna()
        out["breakout_rank"] = pd.NA
        out.loc[bvi_eligible_mask, "breakout_rank"] = (
            out.loc[bvi_eligible_mask, "breakout_velocity_index"]
            .rank(ascending=False, method="min")
            .astype(int)
        )

        return out

    def calculate_hiffi_xp(self, df: pd.DataFrame) -> pd.DataFrame:
        """
        Calculates HIFFI-XP composite multi-platform score and rank.
        Enforces proportional weight redistribution when signals are missing (NaN).
        Weights: YouTube (0.50), Spotify Monthly (0.25), Spotify Velocity (0.05), Instagram (0.20).
        """
        out = df.copy()

        base_weights = {
            "yt": 0.50,
            "sp_monthly": 0.25,
            "sp_velocity": 0.05,
            "ig": 0.20
        }

        def log_scale_vector(series: pd.Series) -> pd.Series:
            valid = series.dropna()
            if valid.empty:
                return pd.Series(np.nan, index=series.index)
            log_val = np.log1p(np.maximum(0.0, valid.astype(float)))
            min_v, max_v = log_val.min(), log_val.max()
            if np.isclose(max_v, min_v):
                return pd.Series(1.0, index=valid.index)
            scaled = (log_val - min_v) / (max_v - min_v + 1e-9)
            return scaled.reindex(series.index)

        def linear_scale_vector(series: pd.Series) -> pd.Series:
            valid = series.dropna()
            if valid.empty:
                return pd.Series(np.nan, index=series.index)
            min_v, max_v = valid.min(), valid.max()
            if np.isclose(max_v, min_v):
                return pd.Series(0.5, index=valid.index)
            scaled = (valid - min_v) / (max_v - min_v + 1e-9)
            return scaled.reindex(series.index)

        s_yt = log_scale_vector(out["production_hiffi_score"]) if "production_hiffi_score" in out.columns else pd.Series(np.nan, index=out.index)
        s_sp_m = log_scale_vector(out["spotify_monthly_listeners"]) if "spotify_monthly_listeners" in out.columns else pd.Series(np.nan, index=out.index)
        s_sp_v = linear_scale_vector(out["spotify_daily_change"]) if "spotify_daily_change" in out.columns else pd.Series(np.nan, index=out.index)
        s_ig = log_scale_vector(out["instagram_followers"]) if "instagram_followers" in out.columns else pd.Series(np.nan, index=out.index)

        signals = pd.DataFrame({
            "yt": s_yt,
            "sp_monthly": s_sp_m,
            "sp_velocity": s_sp_v,
            "ig": s_ig
        }, index=out.index)

        # Proportional Dynamic Weight Redistribution Matrix
        weight_mask = signals.notna()
        allocated_weights = weight_mask.multiply(pd.Series(base_weights), axis=1)
        weight_sum = allocated_weights.sum(axis=1)

        weight_sum = weight_sum.replace(0.0, np.nan)
        normalized_weights = allocated_weights.divide(weight_sum, axis=0)

        component_scores = (signals.fillna(0.0) * normalized_weights).sum(axis=1) * 100.0
        out["hiffi_xp_score"] = component_scores.round(2)
        out.loc[weight_sum.isna(), "hiffi_xp_score"] = np.nan

        # Deterministic tie-breaking for raw mathematical ranking
        sort_cols = ["hiffi_xp_score", "production_hiffi_score", "spotify_monthly_listeners"]
        present_sort = [c for c in sort_cols if c in out.columns]
        
        valid_mask = out["hiffi_xp_score"].notna()
        sorted_indices = (
            out[valid_mask]
            .sort_values(by=present_sort, ascending=[False] * len(present_sort))
            .index
        )

        out["hiffi_xp_rank"] = pd.NA
        out.loc[sorted_indices, "hiffi_xp_rank"] = range(1, len(sorted_indices) + 1)
        out.loc[valid_mask, "hiffi_xp_rank"] = out.loc[valid_mask, "hiffi_xp_rank"].astype(int)

        if "production_hiffi_rank" in out.columns:
            out["rank_delta"] = pd.to_numeric(out["production_hiffi_rank"], errors="coerce") - pd.to_numeric(out["hiffi_xp_rank"], errors="coerce")
        else:
            out["rank_delta"] = np.nan

        return out

    def apply_publication_eligibility_gates(
        self,
        df: pd.DataFrame,
        return_metadata: bool = False
    ) -> pd.DataFrame:
        """
        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. Strict Signal Verification
        has_yt = out["production_hiffi_score"].notna()
        has_sp = out["spotify_monthly_listeners"].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 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(
            by=present_sort_cols,
            ascending=[False] * len(present_sort_cols)
        ).copy()

        assigned_indices = []
        shortfall_audit = {"top_10_shortfall": False, "top_50_shortfall": False}

        # 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)

        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())

            # 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 contiguous published ranks
        out["hiffi_xp_published_rank"] = pd.NA
        rank_series = pd.Series(
            range(1, len(assigned_indices) + 1),
            index=assigned_indices,
            dtype="Int64"
        )
        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