srilekha/hiffi-artist-intelligence
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: scale HIFFI pipeline to 500 artists with dual BVI ranking, dark dashboard, and Excel audit export

Browse Source
This commit is contained in:
Srilekha committed 2026-10-22 17:20:44 +05:30
1 parent 5d441ffe7c
commit 9710cf9948
14 files changed
+3017 -358
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

core/confidence_scorer.py
+109 -54
View File
@@ -1,60 +1,122 @@
from typing import Dict, Any
"""
core/confidence_scorer.py
Computes composite authenticity confidence scores (0.0 - 1.0)
across YouTube channel titles, MusicBrainz, and Wikidata with
noise-cleaning and cross-referencing.
"""
import re
from typing import Dict, Any, Optional
class ConfidenceScorer:
"""
Computes a weighted authenticity and quality score across multiple data sources:
- YouTube footprint: 40%
- MusicBrainz catalog: 35%
- Wikidata entity: 25%
Evaluates multi-source identity alignment across YouTube, MusicBrainz, and Wikidata.
"""
def __init__(self, yt_weight: float = 0.40, mb_weight: float = 0.35, wd_weight: float = 0.25):
self.w_yt = yt_weight
self.w_mb = mb_weight
self.w_wd = wd_weight
def __init__(
self,
yt_weight: float = 0.40,
mb_weight: float = 0.35,
wd_weight: float = 0.25,
):
self.yt_weight = yt_weight
self.mb_weight = mb_weight
self.wd_weight = wd_weight
@staticmethod
def _clean_string(text: Optional[str]) -> str:
"""Strips common artist channel noise tokens and punctuation."""
if not text:
return ""
s = text.lower()
# Remove suffixes like VEVO, Official, Records, Music, Topic
noise_tokens = [
r"\bvevo\b",
r"\bofficial\b",
r"\btopic\b",
r"\bmusic\b",
r"\brecords\b",
r"\btv\b",
r"\bchannel\b",
]
for token in noise_tokens:
s = re.sub(token, "", s)
# Remove all non-alphanumeric characters
return re.sub(r"[^a-z0-9]", "", s)
def _match_youtube(self, artist_name: str, yt_title: Optional[str]) -> float:
"""Evaluates whether the YouTube channel title matches the artist."""
if not yt_title or not artist_name:
return 0.0
clean_artist = self._clean_string(artist_name)
clean_title = self._clean_string(yt_title)
if not clean_artist or not clean_title:
return 0.0
# Exact match after cleaning (e.g. 'KendrickLamarVEVO' -> 'kendricklamar' == 'kendricklamar')
if clean_artist == clean_title:
return 1.0
# Substring containment
if clean_artist in clean_title or clean_title in clean_artist:
return 0.85
return 0.0
def calculate_score(
self,
yt_metrics: Dict[str, Any],
mb_result: Dict[str, Any],
wd_result: Dict[str, Any]
wd_result: Dict[str, Any],
) -> Dict[str, Any]:
# 1. YouTube sub-score (0.0 to 1.0)
yt_score = 0.0
if yt_metrics.get("status") in ["live_api", "cached"]:
subs = yt_metrics.get("subscribers", 0)
views = yt_metrics.get("lifetime_views", 0)
vids = yt_metrics.get("video_count", 0)
"""
Computes composite confidence based on:
- YouTube channel name alignment (40%)
- MusicBrainz match verification (35%)
- Wikidata entity validation (25%)
- Cross-referencing bonus: MBID linked directly inside Wikidata
"""
artist_name = yt_metrics.get("channel_title") or ""
# If artist name was provided separately in the identifier dict, we compare against channel_title
yt_title = yt_metrics.get("channel_title")
raw_name = yt_metrics.get("name") or yt_title
# Baseline 0.5 for a verified channel existence
yt_score = 0.5
if subs >= 1000:
yt_score += 0.2
elif subs > 0:
yt_score += 0.1
# 1. YouTube Title Alignment Score
yt_score = self._match_youtube(raw_name, yt_title)
if views >= 50000:
yt_score += 0.2
elif views > 0:
yt_score += 0.1
# 2. MusicBrainz Score
mb_match = mb_result.get("verified", False) if isinstance(mb_result, dict) else bool(mb_result)
mb_score = 1.0 if mb_match else 0.0
if vids >= 5:
yt_score += 0.1
# 3. Wikidata Score
wd_match = wd_result.get("verified", False) if isinstance(wd_result, dict) else bool(wd_result)
wd_score = 1.0 if wd_match else 0.0
# 2. MusicBrainz sub-score
mb_score = mb_result.get("confidence", 0.0) if mb_result.get("verified") else 0.0
# 4. Cross-Referencing Bonus: Wikidata linked to MusicBrainz ID
cross_ref_bonus = 0.0
if isinstance(wd_result, dict) and isinstance(mb_result, dict):
wd_mbid = wd_result.get("musicbrainz_id")
actual_mbid = mb_result.get("mbid")
if wd_mbid and actual_mbid and wd_mbid == actual_mbid:
cross_ref_bonus = 0.05 # Extra 5% confidence for direct KG linkage
# 3. Wikidata sub-score
wd_score = wd_result.get("confidence", 0.0) if wd_result.get("verified") else 0.0
# Composite calculation
composite = (
self.yt_weight * yt_score
+ self.mb_weight * mb_score
+ self.wd_weight * wd_score
+ cross_ref_bonus
)
composite = round(min(composite, 1.0), 2)
# Aggregated composite score
composite = (yt_score * self.w_yt) + (mb_score * self.w_mb) + (wd_score * self.w_wd)
composite = round(min(1.0, max(0.0, composite)), 2)
# Confidence category
if composite >= 0.75:
tier = "High"
elif composite >= 0.45:
tier = "Medium"
# Tier classification
if composite >= 0.80:
tier = "High Confidence"
elif composite >= 0.50:
tier = "Medium Confidence"
else:
tier = "Low / Emerging"
@@ -62,16 +124,9 @@ class ConfidenceScorer:
"composite_confidence": composite,
"confidence_tier": tier,
"sub_scores": {
"youtube": round(yt_score, 2),
"musicbrainz": round(mb_score, 2),
"wikidata": round(wd_score, 2)
}
}
if __name__ == "__main__":
scorer = ConfidenceScorer()
dummy_yt = {"status": "live_api", "subscribers": 5000, "lifetime_views": 250000, "video_count": 18}
dummy_mb = {"verified": True, "confidence": 1.0}
dummy_wd = {"verified": True, "confidence": 1.0}
print("Test Scorer Output:")
print(scorer.calculate_score(dummy_yt, dummy_mb, dummy_wd))
"youtube": yt_score,
"musicbrainz": mb_score,
"wikidata": wd_score,
"cross_referenced": cross_ref_bonus > 0.0,
},
}
core/health_detector.py
+87
View File
@@ -0,0 +1,87 @@
"""
core/health_detector.py
Identifies inactive, legacy, zombie, and hyper-viral channels
based on engagement-to-audience ratios and upload cadence.
"""
from typing import Dict, Any
import numpy as np
import pandas as pd
class ChannelHealthDetector:
"""
Evaluates channel vitality by examining daily velocity against subscriber scale.
"""
def __init__(self, zombie_sub_threshold: int = 25000, zombie_velocity_cutoff: float = 100.0):
self.zombie_sub_threshold = zombie_sub_threshold
self.zombie_velocity_cutoff = zombie_velocity_cutoff
def classify_channel(self, row: pd.Series) -> Dict[str, Any]:
subs = row.get("subscriber_count")
daily_vel = row.get("daily_view_velocity")
status_flag = row.get("status_flag", "")
# Handle unresolvable or missing data
if pd.isna(subs) or status_flag == "FAILED_LOOKUP":
return {
"health_status": "UNRESOLVED",
"is_zombie": False,
"health_notes": "Channel could not be resolved via YouTube API."
}
subs = float(subs)
daily_vel = float(daily_vel) if pd.notna(daily_vel) else 0.0
engagement_ratio = daily_vel / max(subs, 1000.0)
# 1. Zombie Channel: Large subscriber base with dead viewership
if subs >= self.zombie_sub_threshold and daily_vel < self.zombie_velocity_cutoff:
return {
"health_status": "ZOMBIE_CHANNEL",
"is_zombie": True,
"health_notes": f"High subscriber base ({int(subs):,}) with negligible viewership ({daily_vel:.1f} views/day)."
}
# 2. Dormant Catalog: No recent uploads and negligible velocity
if status_flag == "INACTIVE_OR_DORMANT" or daily_vel == 0.0:
return {
"health_status": "DORMANT_CATALOG",
"is_zombie": subs >= 10000,
"health_notes": "Zero recent uploads and stalled catalog velocity."
}
# 3. Viral Surge: Recent views exceed subscriber base
if engagement_ratio >= 1.0:
return {
"health_status": "VIRAL_SURGE",
"is_zombie": False,
"health_notes": "Audience reach is exploding beyond existing subscriber base."
}
# 4. Catalog Evergreen: Big catalog keeping views high despite no new release
if status_flag == "CATALOG_ACTIVE_NO_RECENT_UPLOAD":
return {
"health_status": "CATALOG_EVERGREEN",
"is_zombie": False,
"health_notes": "Strong passive catalog views despite release pause."
}
# 5. Healthy Active Channel
return {
"health_status": "ACTIVE_HEALTHY",
"is_zombie": False,
"health_notes": "Consistent release schedule and steady audience engagement."
}
def evaluate_dataframe(self, df: pd.DataFrame) -> pd.DataFrame:
"""Applies health classification across all artist records."""
out = df.copy()
health_results = out.apply(self.classify_channel, axis=1)
out["health_status"] = [r["health_status"] for r in health_results]
out["is_zombie"] = [r["is_zombie"] for r in health_results]
out["health_notes"] = [r["health_notes"] for r in health_results]
return out
core/ranking_engine.py
+155 -71
View File
@@ -1,83 +1,167 @@
"""
core/ranking_engine.py
Ranks artists according to HIFFI formula stages:
- Recent Momentum (35%)
- View Velocity (30%)
- Audience Reach (20%)
- Consistency (15%)
Calculates Breakout Velocity Index (BVI, 0-100) and breakout_rank
with a 10,000 subscriber floor to prevent phantom accounts from ranking.
"""
import numpy as np
import pandas as pd
from typing import Dict, Any
class RankingEngine:
"""
Computes equitable artist rankings using Log Transformation (log1p)
and Min-Max Scaling across skewed social/streaming distribution metrics.
"""
DEFAULT_WEIGHTS = {
"norm_subscribers": 0.30,
"norm_lifetime_views": 0.25,
"norm_recent_avg_views": 0.25,
"norm_video_count": 0.10,
"norm_momentum_30d": 0.10
}
def __init__(self, weights: Dict[str, float] = None):
self.weights = weights or self.DEFAULT_WEIGHTS
class HIFFIRankingEngine:
def normalize_metrics(self, df: pd.DataFrame) -> pd.DataFrame:
"""Applies log10(1 + x) and Min-Max scaling to compress power-law tails."""
df_scored = df.copy()
@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)
metrics_to_transform = [
("yt_subscribers", "norm_subscribers"),
("yt_lifetime_views", "norm_lifetime_views"),
("yt_video_count", "norm_video_count"),
("yt_recent_avg_views", "norm_recent_avg_views"),
("yt_momentum_30d", "norm_momentum_30d")
]
@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)
for raw_col, norm_col in metrics_to_transform:
if raw_col not in df_scored.columns:
df_scored[norm_col] = 0.0
continue
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)
# Step 1: Handle NaNs and negative values, then apply Log10(1 + x)
series = pd.to_numeric(df_scored[raw_col], errors="coerce").fillna(0).clip(lower=0)
log_series = np.log10(1 + series)
min_val = log_series.min()
max_val = log_series.max()
# Step 2: Min-Max Normalization to scale [0.0, 1.0]
if max_val > min_val:
df_scored[norm_col] = (log_series - min_val) / (max_val - min_val)
else:
df_scored[norm_col] = 0.0
# Step 3: Compute weighted base score
df_scored["base_rank_score"] = sum(
df_scored[col] * weight for col, weight in self.weights.items()
# Qualification Gate: Must have real baseline audience (>= 10,000 subs)
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")
)
# Step 4: Multiply by verification confidence to penalize unverified or fake profiles
# Artists with verified MusicBrainz/Wikidata and legitimate channels get full weight
confidence = df_scored.get("composite_confidence", pd.Series(1.0, index=df_scored.index))
# Ensure a minimum floor of 0.2 so emerging artists are ranked on data, not dropped completely
confidence_multiplier = 0.2 + (0.8 * confidence)
# Step 5: Final HIFFI Score on a clean 0 to 100 scale
df_scored["hiffi_score"] = (df_scored["base_rank_score"] * confidence_multiplier * 100).round(2)
# Sort and assign explicit rank ordinal
df_scored = df_scored.sort_values(by="hiffi_score", ascending=False).reset_index(drop=True)
df_scored["hiffi_rank"] = df_scored.index + 1
if not valid_mask.any():
return out_bvi
return df_scored
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)
if __name__ == "__main__":
# Test script with dummy mock distribution
sample_data = pd.DataFrame([
{"Name": "Global Superstar", "yt_subscribers": 15000000, "yt_lifetime_views": 8500000000, "yt_video_count": 120, "yt_recent_avg_views": 2500000, "yt_momentum_30d": 1.2, "composite_confidence": 0.95},
{"Name": "Mid-tier Rising", "yt_subscribers": 250000, "yt_lifetime_views": 45000000, "yt_video_count": 45, "yt_recent_avg_views": 180000, "yt_momentum_30d": 5.4, "composite_confidence": 0.85},
{"Name": "Emerging Talent", "yt_subscribers": 12000, "yt_lifetime_views": 850000, "yt_video_count": 14, "yt_recent_avg_views": 45000, "yt_momentum_30d": 12.8, "composite_confidence": 0.60},
{"Name": "Dormant / Inactive", "yt_subscribers": 500, "yt_lifetime_views": 15000, "yt_video_count": 3, "yt_recent_avg_views": 50, "yt_momentum_30d": 0.0, "composite_confidence": 0.35}
])
engine = RankingEngine()
ranked = engine.normalize_metrics(sample_data)
print("Ranked Test Artists with Log & Min-Max Transformation:")
print(ranked[["hiffi_rank", "Name", "hiffi_score", "norm_subscribers", "norm_lifetime_views"]])
# 1. Audience Velocity Ratio: Views per day relative to subscriber base
v_aud = daily_vel / np.maximum(subs, 1000.0)
# 2. Scaled Momentum Factor: Map [-100, 300] to [0, 1]
m_signal = (momentum + 100.0) / 400.0
# 3. Log-compress velocity to manage power-law peaks
log_v_aud = np.log10(1.0 + v_aud)
# 4. Emergence Dampener: inverse log weight to favor emerging accounts
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
# Min-max scale to 0-100 across valid records
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:
out = df.copy()
# Handle confidence score defensively
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)
# Compute BVI with the 10,000 sub floor
out["breakout_velocity_index"] = self.calculate_breakout_velocity(out)
# Verification & completeness gate
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"]))
)
# 1. Recent Momentum (35%)
filled_momentum = out["momentum_30d"].fillna(0.0).clip(lower=-100.0, upper=300.0)
score_momentum = self._min_max_scale(filled_momentum)
# 2. View Velocity (30%)
filled_velocity = out["daily_view_velocity"].fillna(0.0)
score_velocity = self._log_scale(filled_velocity)
# 3. Audience Reach (20%)
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
# 4. Consistency (15%)
score_consistency = self._log_scale(out["video_count"].fillna(0.0))
# Composite HIFFI Score (0 - 100)
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["hiffi_score"] = (hiffi_raw * conf_multiplier).round(2)
# Primary HIFFI Rank
eligible_mask = out["is_rank_eligible"]
out["hiffi_rank"] = pd.NA
out.loc[eligible_mask, "hiffi_rank"] = (
out.loc[eligible_mask, "hiffi_score"]
.rank(ascending=False, method="min")
.astype(int)
)
# Breakout Rank (only for eligible artists that have a valid BVI)
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.sort_values(
by=["is_rank_eligible", "hiffi_score"], ascending=[False, False]
).reset_index(drop=True)