feat(cron): finalize top 500 cohort weekly ranking pipeline with delta tracking

scripts/cron_weekly_ranking.py

@@ -0,0 +1,175 @@
+import os
+import sys
+import sqlite3
+import logging
+from datetime import datetime, timezone
+import pandas as pd
+
+BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+if BASE_DIR not in sys.path:
+    sys.path.insert(0, BASE_DIR)
+
+from core.ranking_engine import HIFFIRankingEngine
+
+DB_PATH = os.path.join(BASE_DIR, "data", "hiffi_catalog.db")
+CSV_PATH = os.path.join(BASE_DIR, "data", "hiffi_multiplatform_500.csv")
+LOG_PATH = os.path.join(BASE_DIR, "cron_ranking.log")
+
+logging.basicConfig(
+    filename=LOG_PATH,
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(message)s",
+    datefmt="%Y-%m-%d %H:%M:%S"
+)
+logger = logging.getLogger("HIFFI_CRON")
+
+def run_weekly_ranking_pipeline():
+    start_time = datetime.now(timezone.utc)
+    print("\n========================================================")
+    print("  HIFFI AUTOMATED WEEKLY RANKING PIPELINE (TOP 500 ONLY)")
+    print(f"  Run Timestamp (UTC): {start_time.isoformat()}")
+    print("========================================================\n")
+    logger.info("Starting automated 500-cohort weekly ranking pipeline...")
+
+    if not os.path.exists(DB_PATH):
+        err = f"Database not found at {DB_PATH}"
+        logger.error(err)
+        print(f"ERROR: {err}")
+        return False
+
+    if not os.path.exists(CSV_PATH):
+        err = f"Cohort CSV not found at {CSV_PATH}"
+        logger.error(err)
+        print(f"ERROR: {err}")
+        return False
+
+    conn = sqlite3.connect(DB_PATH)
+    cur = conn.cursor()
+
+    try:
+        # Read the verified 500 cohort IDs directly from the baseline CSV
+        base_cohort_df = pd.read_csv(CSV_PATH)
+        cohort_ids = [str(x) for x in base_cohort_df['artist_id'].dropna().tolist()][:500]
+        placeholders = ",".join([f"'{aid}'" for aid in cohort_ids])
+
+        # 1. Fetch telemetry strictly for the 500 cohort
+        query = f"""
+            WITH latest_telemetry AS (
+                SELECT t.*, ROW_NUMBER() OVER (
+                    PARTITION BY t.artist_id 
+                    ORDER BY t.last_synced_at DESC, t.telemetry_id DESC
+                ) as rn
+                FROM telemetry_metrics t
+                WHERE t.artist_id IN ({placeholders})
+            )
+            SELECT 
+                a.artist_id,
+                a.artist_name,
+                a.country_origin,
+                a.image_url,
+                a.spotify_url,
+                a.youtube_url,
+                a.instagram_url,
+                COALESCE(t.spotify_monthly_listeners, 0) as spotify_monthly_listeners,
+                COALESCE(t.subscribers_count, 0) as subscriber_count,
+                COALESCE(t.daily_view_velocity, 0) as daily_view_velocity,
+                COALESCE(t.instagram_followers, 0) as instagram_followers,
+                r.catalog_rank as previous_rank,
+                r.catalog_intelligence_tier as current_tier
+            FROM artists a
+            LEFT JOIN latest_telemetry t ON a.artist_id = t.artist_id AND t.rn = 1
+            LEFT JOIN rankings r ON a.artist_id = r.artist_id
+            WHERE a.artist_id IN ({placeholders});
+        """
+        df = pd.read_sql_query(query, conn)
+        total_artists = len(df)
+        print(f"Loaded verified cohort: {total_artists} artists.")
+        logger.info(f"Loaded {total_artists} cohort artists for recalculation.")
+
+        # 2. Run Core Scoring Engine (45% SP / 35% YT / 20% IG)
+        engine = HIFFIRankingEngine()
+        ranked_df = engine.calculate_hiffi_xp(df)
+
+        if 'hiffi_xp_score' in ranked_df.columns:
+            score_col = 'hiffi_xp_score'
+        elif 'score' in ranked_df.columns:
+            score_col = 'score'
+        elif 'production_hiffi_score' in ranked_df.columns:
+            score_col = 'production_hiffi_score'
+        else:
+            numeric_cols = ranked_df.select_dtypes(include=['float64', 'float32']).columns
+            score_col = numeric_cols[0]
+
+        # Sort descending by score
+        ranked_df = ranked_df.sort_values(by=score_col, ascending=False).reset_index(drop=True)
+        ranked_df['new_rank'] = ranked_df.index + 1
+
+        # Track rank deltas
+        ranked_df['rank_delta'] = ranked_df.apply(
+            lambda row: int(row['previous_rank'] - row['new_rank'])
+            if pd.notna(row['previous_rank']) and row['previous_rank'] > 0
+            else 0,
+            axis=1
+        )
+
+        def compute_tier(rank_val):
+            if rank_val <= 50:
+                return "TIER_1_ICON"
+            elif rank_val <= 200:
+                return "TIER_2_MAINSTREAM"
+            else:
+                return "TIER_3_RISING"
+
+        ranked_df['tier_rank'] = ranked_df['new_rank']
+        ranked_df['catalog_intelligence_tier'] = ranked_df['new_rank'].apply(compute_tier)
+
+        # 3. Update SQLite database rankings table for the 500 cohort
+        print("Committing 500 cohort ranks to SQLite database...")
+        update_data = []
+        for _, row in ranked_df.iterrows():
+            aid = str(row['artist_id'])
+            c_rank = int(row['new_rank'])
+            t_rank = int(row['tier_rank'])
+            tier_str = str(row['catalog_intelligence_tier'])
+            update_data.append((c_rank, t_rank, tier_str, aid))
+
+        cur.executemany("""
+            UPDATE rankings 
+            SET catalog_rank = ?,
+                tier_rank = ?,
+                catalog_intelligence_tier = ?
+            WHERE artist_id = ?;
+        """, update_data)
+        conn.commit()
+        print(f"Updated database ranking records for {len(update_data)} artists.")
+
+        # 4. Save updated 500 cohort CSV
+        ranked_df['catalog_rank'] = ranked_df['new_rank']
+        if 'name' not in ranked_df.columns and 'artist_name' in ranked_df.columns:
+            ranked_df['name'] = ranked_df['artist_name']
+
+        ranked_df.to_csv(CSV_PATH, index=False)
+        print(f"Exported Top 500 cohort CSV to: {CSV_PATH}")
+
+        # Telemetry summary
+        print("\n--- TOP 5 RANKING SNAPSHOT ---")
+        for _, r in ranked_df.head(5).iterrows():
+            delta_str = f"+{r['rank_delta']}" if r['rank_delta'] > 0 else str(r['rank_delta'])
+            print(f"#{r['new_rank']} {r.get('name', r.get('artist_name'))} | Score: {r[score_col]:.2f} | Tier: {r['catalog_intelligence_tier']} | Delta: {delta_str}")
+
+        elapsed = (datetime.now(timezone.utc) - start_time).total_seconds()
+        print(f"\nWeekly ranking pipeline completed successfully in {elapsed:.2f}s.")
+        logger.info(f"Weekly ranking pipeline finished successfully in {elapsed:.2f}s.")
+        return True
+
+    except Exception as e:
+        conn.rollback()
+        logger.error(f"Weekly ranking run failed: {str(e)}", exc_info=True)
+        print(f"\nERROR running ranking pipeline: {e}")
+        return False
+    finally:
+        conn.close()
+
+if __name__ == "__main__":
+    success = run_weekly_ranking_pipeline()
+    sys.exit(0 if success else 1)