""" 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, proportional weight redistribution, and effective weight serialization. - Publication Eligibility Gates: Decoupled public chart assigning consecutive ranks while enforcing multi-platform evidence tiers (Top 10 = Tier 3, Top 50 = Tier >= 2) conditioned on strict identity verification, status_flag validation, and UTC data freshness thresholds. Fail-closed on missing timestamp columns, failed lookups, or unverified statuses. """ from typing import Optional, Tuple, Union 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.log1p(np.maximum(valid.astype(float), 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) @staticmethod def _is_signal_fresh( timestamp_val: object, max_days: float, reference_time: Optional[pd.Timestamp] = None ) -> bool: """ Parses UTC ISO 8601 string or pd.Timestamp and checks staleness. Fails closed (returns False) on missing, null, or unparseable timestamps. """ if pd.isna(timestamp_val): return False try: ts = pd.to_datetime(timestamp_val, utc=True) ref = reference_time or pd.Timestamp.now(tz="UTC") age_days = (ref - ts).total_seconds() / 86400.0 return 0.0 <= age_days <= max_days except Exception: return False 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) status_col = df["status_flag"] if "status_flag" in df.columns else pd.Series("COMPLETE", index=df.index) valid_mask = ( df["subscriber_count"].notna() & (df["subscriber_count"] >= 10000) & df["daily_view_velocity"].notna() & (df["daily_view_velocity"] > 0.0) & (status_col != "INACTIVE_OR_DORMANT") & (status_col != "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) status_col = out["status_flag"] if "status_flag" in out.columns else pd.Series("COMPLETE", index=out.index) has_baseline = ( out["subscriber_count"].notna() & out["lifetime_views"].notna() & (status_col != "FAILED_LOOKUP") ) out["is_rank_eligible"] = ( has_baseline & (out.get("confidence_score", pd.Series(1.0, index=out.index)) >= min_confidence) & (status_col.isin(["COMPLETE", "CATALOG_ACTIVE_NO_RECENT_UPLOAD"])) ) filled_momentum = out["momentum_30d"].fillna(0.0).clip(lower=-100.0, upper=300.0) if "momentum_30d" in out.columns else pd.Series(0.0, index=out.index) score_momentum = self._min_max_scale(filled_momentum) filled_velocity = out["daily_view_velocity"].fillna(0.0) if "daily_view_velocity" in out.columns else pd.Series(0.0, index=out.index) score_velocity = self._log_scale(filled_velocity) log_subs = self._log_scale(out["subscriber_count"].fillna(0.0)) if "subscriber_count" in out.columns else pd.Series(0.0, index=out.index) log_lifetime = self._log_scale(out["lifetime_views"].fillna(0.0)) if "lifetime_views" in out.columns else pd.Series(0.0, index=out.index) score_reach = 0.60 * log_subs + 0.40 * log_lifetime score_consistency = self._log_scale(out["video_count"].fillna(0.0)) if "video_count" in out.columns else pd.Series(0.0, index=out.index) 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.get("confidence_score", pd.Series(1.0, index=out.index)).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, raw rank, and effective weights. 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() # Schema aliasing for raw input fallbacks if "production_hiffi_score" not in out.columns and "hiffi_score" in out.columns: out["production_hiffi_score"] = out["hiffi_score"] if "production_hiffi_rank" not in out.columns and "hiffi_rank" in out.columns: out["production_hiffi_rank"] = out["hiffi_rank"] 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) # Explainability: Format explicit percentage breakdown per artist def format_weights(row: pd.Series) -> str: parts = [f"{col}:{val*100.0:.1f}%" for col, val in row.items() if val > 0] return ",".join(parts) if parts else "NONE" out["effective_weights"] = normalized_weights.apply(format_weights, axis=1) 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 continuous 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 detect_metric_anomalies(self, df: pd.DataFrame) -> pd.DataFrame: """ Flags suspicious metric patterns without altering raw scores: - SUSPICIOUS_IG_TO_YT_RATIO: IG followers > 10x YouTube lifetime views. - EXTREME_SPOTIFY_VELOCITY_RATIO: Spotify daily change > 25% of monthly listeners. - Populates 'anomaly_flags' column (semicolon-delimited or 'CLEAN'). """ out = df.copy() flags = [[] for _ in range(len(out))] # 1. Instagram to YouTube Lifetime Ratio Check if "instagram_followers" in out.columns and "lifetime_views" in out.columns: ig = pd.to_numeric(out["instagram_followers"], errors="coerce") yt_views = pd.to_numeric(out["lifetime_views"], errors="coerce") suspicious_ratio = (ig.notna() & yt_views.notna() & (yt_views > 0) & (ig > 10.0 * yt_views)) for idx, is_suspicious in enumerate(suspicious_ratio): if is_suspicious: flags[idx].append("SUSPICIOUS_IG_TO_YT_RATIO") # 2. Spotify Velocity to Monthly Listeners Check if "spotify_daily_change" in out.columns and "spotify_monthly_listeners" in out.columns: sp_change = pd.to_numeric(out["spotify_daily_change"], errors="coerce") sp_monthly = pd.to_numeric(out["spotify_monthly_listeners"], errors="coerce") extreme_velocity = ( sp_change.notna() & sp_monthly.notna() & (sp_monthly > 1000) & (sp_change.abs() > 0.25 * sp_monthly) ) for idx, is_extreme in enumerate(extreme_velocity): if is_extreme: flags[idx].append("EXTREME_SPOTIFY_VELOCITY_RATIO") out["anomaly_flags"] = [";".join(f) if f else "CLEAN" for f in flags] return out def apply_publication_eligibility_gates( self, df: pd.DataFrame, reference_time: Optional[pd.Timestamp] = None, return_metadata: bool = False ) -> Union[pd.DataFrame, Tuple[pd.DataFrame, dict]]: """ Constructs a publication ranking layer decoupled from raw scores: - Top 10 requires all 3 platforms (Tier 3), strictly verified and fresh. - Top 50 requires at least 2 platforms (Tier >= 2), strictly verified and fresh. - Freshness Thresholds: YouTube <= 7d, Spotify <= 7d, Instagram <= 14d. - Integrity Gate: Artists with status_flag == 'FAILED_LOOKUP' are disqualified from YouTube credit. - Fail-closed: missing timestamp columns, stale records, or unverified statuses evaluate strictly to False and are logged to evidence_exclusions. - Preserves raw hiffi_xp_score and raw hiffi_xp_rank completely unchanged. - Produces consecutive, unique ranks in 'hiffi_xp_published_rank' (1 to N). """ out = df.copy() ref_time = reference_time or pd.Timestamp.now(tz="UTC") exclusions = [[] for _ in range(len(out))] # 1. YouTube Validation (Score present + status not FAILED_LOOKUP + Freshness <= 7 days) status_col = out["status_flag"] if "status_flag" in out.columns else pd.Series("COMPLETE", index=out.index) has_yt_val = out["production_hiffi_score"].notna() if "production_hiffi_score" in out.columns else pd.Series(False, index=out.index) is_lookup_failed = status_col == "FAILED_LOOKUP" if "youtube_updated_at" in out.columns: yt_fresh = out["youtube_updated_at"].apply( lambda x: self._is_signal_fresh(x, max_days=7.0, reference_time=ref_time) ) else: yt_fresh = pd.Series(False, index=out.index) for idx, (val, failed, fresh) in enumerate(zip(has_yt_val, is_lookup_failed, yt_fresh)): if failed: exclusions[idx].append("FAILED_YOUTUBE_LOOKUP") elif val and not fresh: exclusions[idx].append("STALE_OR_MISSING_YOUTUBE_TIMESTAMP") valid_yt = has_yt_val & (~is_lookup_failed) & yt_fresh # 2. Spotify Validation (Telemetry present + Freshness <= 7 days) has_sp_val = out["spotify_monthly_listeners"].notna() if "spotify_monthly_listeners" in out.columns else pd.Series(False, index=out.index) if "spotify_updated_at" in out.columns: sp_fresh = out["spotify_updated_at"].apply( lambda x: self._is_signal_fresh(x, max_days=7.0, reference_time=ref_time) ) else: sp_fresh = pd.Series(False, index=out.index) for idx, (val, fresh) in enumerate(zip(has_sp_val, sp_fresh)): if val and not fresh: exclusions[idx].append("STALE_OR_MISSING_SPOTIFY_TIMESTAMP") valid_sp = has_sp_val & sp_fresh # 3. Instagram Validation (Followers present + Approved Status + Freshness <= 14 days) has_ig_val = out["instagram_followers"].notna() if "instagram_followers" in out.columns else pd.Series(False, index=out.index) verified_statuses = {"AUTOMATED_VERIFIED", "MANUAL_VERIFIED"} if "instagram_verification_status" in out.columns: ig_verified = out["instagram_verification_status"].isin(verified_statuses) else: ig_verified = pd.Series(False, index=out.index) for idx, (val, ver) in enumerate(zip(has_ig_val, ig_verified)): if val and not ver: exclusions[idx].append("UNVERIFIED_INSTAGRAM_IDENTITY") if "instagram_updated_at" in out.columns: ig_fresh = out["instagram_updated_at"].apply( lambda x: self._is_signal_fresh(x, max_days=14.0, reference_time=ref_time) ) else: ig_fresh = pd.Series(False, index=out.index) for idx, (val, ver, fresh) in enumerate(zip(has_ig_val, ig_verified, ig_fresh)): if val and ver and not fresh: exclusions[idx].append("STALE_OR_MISSING_INSTAGRAM_TIMESTAMP") valid_ig = has_ig_val & ig_verified & ig_fresh out["platform_evidence_tier"] = ( valid_yt.astype(int) + valid_sp.astype(int) + valid_ig.astype(int) ) out["evidence_exclusions"] = [";".join(e) if e else "NONE" for e in exclusions] # 4. Deterministic sorting 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: Select Top 10 from Tier 3 candidates 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()) 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: Select Positions 11 to 50 from Tier >= 2 candidates 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 all remaining artists strictly by score order final_remaining = pool.drop(index=assigned_indices) assigned_indices.extend(final_remaining.index.tolist()) # 5. Assign consecutive 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