hiffi-artist-intelligence/core/ranking_engine.py

import math
from datetime import datetime, timezone
from typing import Any, Dict, Optional, Tuple, Union
import numpy as np
import pandas as pd


class HIFFIRankingEngine:
    """
    HIFFI-XP Core Ranking Engine & Governance Suite.
    - 45/35/20 Bounded log-scaled scoring with dynamic weight redistribution.
    - Publication eligibility gates (timestamp freshness, identity verification, status flags).
    - Statistical anomaly detection for ratio spikes and unnatural patterns.
    """

    def __init__(self):
        self.MAX_SPOTIFY = 120_000_000.0
        self.MAX_YT_SUBS = 70_000_000.0
        self.MAX_YT_VELOCITY = 5_000_000.0
        self.MAX_INSTAGRAM = 150_000_000.0

        self.w_sp = 0.45
        self.w_yt = 0.35
        self.w_ig = 0.20

    def _log_scale(self, val: float, max_bound: float) -> float:
        if pd.isna(val) or val <= 0:
            return 0.0
        s = math.log10(float(val) + 1.0)
        mx = math.log10(float(max_bound) + 1.0)
        return float(np.clip((s / mx) * 100.0, 0.0, 100.0))

    def calculate_breakout_velocity(self, df: pd.DataFrame) -> pd.Series:
        if df.empty:
            return pd.Series(dtype=float)
        growth_7d = pd.to_numeric(df.get("growth_velocity_7d", 0.0), errors="coerce").fillna(0.0)
        growth_30d = pd.to_numeric(df.get("growth_velocity_30d", 0.0), errors="coerce").fillna(0.0)
        raw_bvi = (growth_7d * 0.6 + growth_30d * 0.4) * 10.0
        return pd.Series(np.clip(raw_bvi, 0.0, 100.0), index=df.index).round(2)

    def calculate_hiffi_xp(self, df: pd.DataFrame) -> pd.DataFrame:
        out = df.copy()

        has_sp = out.get("spotify_monthly_listeners", pd.Series(0, index=out.index)).fillna(0).gt(0).astype(int)
        has_yt = out.get("subscriber_count", pd.Series(0, index=out.index)).fillna(0).gt(0).astype(int)
        has_ig = out.get("instagram_followers", pd.Series(0, index=out.index)).fillna(0).gt(0).astype(int)

        if "subscriber_count" not in out.columns:
            has_yt = out.get("lifetime_views", pd.Series(0, index=out.index)).fillna(0).gt(0).astype(int)
            if has_yt.sum() == 0 and "status_flag" in out.columns:
                has_yt = (out["status_flag"] == "COMPLETE").astype(int)

        out["platform_evidence_tier"] = has_sp + has_yt + has_ig

        confidence_map = {3: 1.00, 2: 0.85, 1: 0.65, 0: 0.50}
        out["confidence_score"] = out["platform_evidence_tier"].map(confidence_map).fillna(1.0)

        scores = []
        effective_weight_strings = []

        for _, row in out.iterrows():
            sp_val = row.get("spotify_monthly_listeners")
            sp_clean = float(sp_val) if pd.notna(sp_val) and float(sp_val) > 0 else 0.0

            yt_sub = row.get("subscriber_count")
            yt_sub_clean = float(yt_sub) if pd.notna(yt_sub) and float(yt_sub) > 0 else 0.0

            yt_vel = row.get("daily_view_velocity")
            yt_vel_clean = float(yt_vel) if pd.notna(yt_vel) and float(yt_vel) > 0 else 0.0

            ig_val = row.get("instagram_followers")
            ig_clean = float(ig_val) if pd.notna(ig_val) and float(ig_val) > 0 else 0.0

            s_sp = self._log_scale(sp_clean, self.MAX_SPOTIFY)
            s_yt = self._log_scale(yt_sub_clean, self.MAX_YT_SUBS) * 0.70 + self._log_scale(yt_vel_clean, self.MAX_YT_VELOCITY) * 0.30
            s_ig = self._log_scale(ig_clean, self.MAX_INSTAGRAM)

            active_weights = {}
            if sp_clean > 0:
                active_weights["sp"] = self.w_sp
            if yt_sub_clean > 0 or yt_vel_clean > 0:
                active_weights["yt"] = self.w_yt
            if ig_clean > 0:
                active_weights["ig"] = self.w_ig

            total_active = sum(active_weights.values())

            if total_active > 0:
                norm_sp = active_weights.get("sp", 0.0) / total_active
                norm_yt = active_weights.get("yt", 0.0) / total_active
                norm_ig = active_weights.get("ig", 0.0) / total_active

                raw_score = (s_sp * norm_sp) + (s_yt * norm_yt) + (s_ig * norm_ig)
                conf = confidence_map.get(row.get("platform_evidence_tier", 3), 1.0)
                final_score = float(round(raw_score * conf, 2))
                wt_str = f"sp:{round(norm_sp*100,1)}%,yt:{round(norm_yt*100,1)}%,ig:{round(norm_ig*100,1)}%"
            else:
                prod_score = row.get("production_hiffi_score")
                if pd.notna(prod_score):
                    final_score = float(round(float(prod_score), 2))
                else:
                    final_score = 0.0
                wt_str = "unindexed"

            scores.append(final_score)
            effective_weight_strings.append(wt_str)

        out["hiffi_xp_score"] = scores
        out["effective_weights"] = effective_weight_strings

        if "catalog_rank" not in out.columns or out["catalog_rank"].isna().any():
            out["catalog_rank"] = out["hiffi_xp_score"].rank(ascending=False, method="min").astype(int)
        out["hiffi_xp_published_rank"] = out["catalog_rank"]

        return out

    def score_and_rank(self, df: pd.DataFrame) -> pd.DataFrame:
        return self.calculate_hiffi_xp(df)

    def apply_publication_eligibility_gates(
        self,
        df: pd.DataFrame,
        reference_time: Optional[Any] = None,
        return_metadata: bool = False
    ) -> Union[pd.DataFrame, Tuple[pd.DataFrame, Dict[str, Any]]]:
        out = df.copy()

        if reference_time is None:
            ref_dt = datetime.now(timezone.utc)
        elif isinstance(reference_time, datetime):
            ref_dt = reference_time if reference_time.tzinfo else reference_time.replace(tzinfo=timezone.utc)
        else:
            ref_dt = pd.to_datetime(reference_time, utc=True).to_pydatetime()

        max_staleness_days = 7.0

        def is_fresh(ts_val: Any) -> bool:
            if pd.isna(ts_val) or ts_val is None:
                return False
            try:
                dt = pd.to_datetime(ts_val, utc=True).to_pydatetime()
                diff = (ref_dt - dt).total_seconds() / 86400.0
                return 0.0 <= diff <= max_staleness_days
            except Exception:
                return False

        has_yt_col = "youtube_updated_at" in out.columns
        has_sp_col = "spotify_updated_at" in out.columns
        has_ig_col = "instagram_updated_at" in out.columns
        has_ver_col = "instagram_verification_status" in out.columns

        yt_credits = []
        sp_credits = []
        ig_credits = []
        exclusions_list = []

        for _, row in out.iterrows():
            row_exclusions = []

            # 1. YouTube Evaluation
            status_flag = str(row.get("status_flag", "COMPLETE")).upper()
            if status_flag == "FAILED_LOOKUP":
                yt_ok = False
                row_exclusions.append("FAILED_YOUTUBE_LOOKUP")
            elif not has_yt_col:
                yt_ok = False
                row_exclusions.append("STALE_OR_MISSING_YOUTUBE_TIMESTAMP")
            else:
                yt_ok = is_fresh(row.get("youtube_updated_at"))
                if not yt_ok:
                    row_exclusions.append("STALE_OR_MISSING_YOUTUBE_TIMESTAMP")
            yt_credits.append(int(yt_ok))

            # 2. Spotify Evaluation
            sp_val = row.get("spotify_monthly_listeners")
            has_sp_val = pd.notna(sp_val) and float(sp_val) > 0
            if not has_sp_col:
                sp_ok = False
                row_exclusions.append("STALE_OR_MISSING_SPOTIFY_TIMESTAMP")
            else:
                sp_ok = has_sp_val and is_fresh(row.get("spotify_updated_at"))
                if not sp_ok:
                    row_exclusions.append("STALE_OR_MISSING_SPOTIFY_TIMESTAMP")
            sp_credits.append(int(sp_ok))

            # 3. Instagram Evaluation
            ig_val = row.get("instagram_followers")
            has_ig_val = pd.notna(ig_val) and float(ig_val) > 0

            if not has_ver_col:
                ig_ver_ok = False
            else:
                ver_status = str(row.get("instagram_verification_status", "")).upper()
                ig_ver_ok = ver_status in ("AUTOMATED_VERIFIED", "MANUAL_VERIFIED", "VERIFIED")

            if not has_ig_col:
                ig_fresh = False
                row_exclusions.append("STALE_OR_MISSING_INSTAGRAM_TIMESTAMP")
            else:
                ig_fresh = is_fresh(row.get("instagram_updated_at"))
                if not ig_fresh:
                    row_exclusions.append("STALE_OR_MISSING_INSTAGRAM_TIMESTAMP")

            ig_ok = has_ig_val and ig_ver_ok and ig_fresh
            ig_credits.append(int(ig_ok))

            exclusions_list.append(";".join(row_exclusions) if row_exclusions else "NONE")

        out["gated_yt_credit"] = yt_credits
        out["gated_sp_credit"] = sp_credits
        out["gated_ig_credit"] = ig_credits
        out["platform_evidence_tier"] = [y + s + i for y, s, i in zip(yt_credits, sp_credits, ig_credits)]
        out["evidence_exclusions"] = exclusions_list

        if "hiffi_xp_published_rank" not in out.columns:
            if "hiffi_xp_score" in out.columns:
                out["hiffi_xp_published_rank"] = out["hiffi_xp_score"].rank(ascending=False, method="min").astype(int)
            else:
                out["hiffi_xp_published_rank"] = list(range(1, len(out) + 1))

        if return_metadata:
            top_10 = out.head(10)
            top_50 = out.head(50)
            metadata = {
                "total_records": len(out),
                "top_10_shortfall": bool((top_10["platform_evidence_tier"] < 3).any()),
                "top_50_shortfall": bool((top_50["platform_evidence_tier"] < 3).any()),
                "tier_distribution": out["platform_evidence_tier"].value_counts().to_dict(),
                "reference_time": ref_dt.isoformat(),
            }
            return out, metadata

        return out

    def detect_metric_anomalies(self, df: pd.DataFrame) -> pd.DataFrame:
        out = df.copy()
        flags = []

        for _, row in out.iterrows():
            row_flags = []
            views = float(row.get("lifetime_views") or 0.0)
            ig = float(row.get("instagram_followers") or 0.0)
            sp_month = float(row.get("spotify_monthly_listeners") or 0.0)
            sp_change = float(row.get("spotify_daily_change") or 0.0)

            if views > 0 and ig > 0:
                if (ig / views) > 10.0:
                    row_flags.append("SUSPICIOUS_IG_TO_YT_RATIO")

            if sp_month > 0 and sp_change > 0:
                if (sp_change / sp_month) > 0.20:
                    row_flags.append("EXTREME_SPOTIFY_VELOCITY_RATIO")

            flags.append(";".join(row_flags) if row_flags else "CLEAN")

        out["anomaly_flags"] = flags
        return out