87 lines
3.3 KiB
Python
87 lines
3.3 KiB
Python
|
"""
|
||
|
|
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
|