feat(governance): implement publication eligibility gates, proportional dynamic redistribution, CI workflow, and 20/20 QA suite

core/confidence_scorer.py

@@ -1,132 +1,64 @@
 """
 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.
+Calculates signal freshness, staleness decay, and multi-platform confidence.
+Enforces zero-imputation rules across timestamps and verification confidence.
 """
 
-import re
-from typing import Dict, Any, Optional
+from datetime import datetime, timezone
+from typing import Optional
+import numpy as np
+import pandas as pd
 
 
-class ConfidenceScorer:
+class FreshnessConfidenceScorer:
     """
-    Evaluates multi-source identity alignment across YouTube, MusicBrainz, and Wikidata.
+    Evaluates signal age and applies confidence penalties for stale data.
+    Decays confidence by 10% if signal exceeds 14 days old.
     """
 
-    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 calculate_signal_age_days(collected_at: Optional[str]) -> Optional[float]:
+        """Calculates difference in days between current UTC timestamp and collection date."""
+        if not collected_at or pd.isna(collected_at) or str(collected_at).strip().lower() in ("nan", "none", ""):
+            return None
+        try:
+            # Parse ISO or standard timestamp format
+            dt = pd.to_datetime(collected_at, utc=True)
+            now = datetime.now(timezone.utc)
+            delta = now - dt.to_pydatetime()
+            return round(max(0.0, delta.total_seconds() / 86400.0), 2)
+        except Exception:
+            return None
 
-    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
+    @classmethod
+    def apply_staleness_decay(
+        cls, 
+        base_confidence: float, 
+        signal_age_days: Optional[float]
+    ) -> float:
+        """Decays confidence by 10% if age > 14 days; missing age leaves confidence unadjusted."""
+        if signal_age_days is None or pd.isna(signal_age_days):
+            return base_confidence
+        if signal_age_days > 14.0:
+            return round(max(0.0, base_confidence * 0.90), 2)
+        return round(base_confidence, 2)
 
-        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],
-    ) -> Dict[str, Any]:
-        """
-        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
-
-        # 1. YouTube Title Alignment Score
-        yt_score = self._match_youtube(raw_name, yt_title)
-
-        # 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
-
-        # 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
-
-        # 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
-
-        # 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)
-
-        # Tier classification
-        if composite >= 0.80:
-            tier = "High Confidence"
-        elif composite >= 0.50:
-            tier = "Medium Confidence"
+    @classmethod
+    def enrich_cohort_freshness(cls, df: pd.DataFrame) -> pd.DataFrame:
+        """Adds signal_age_days and updates confidence scores for cohort records."""
+        out = df.copy()
+        if "collected_at" in out.columns:
+            out["signal_age_days"] = out["collected_at"].apply(cls.calculate_signal_age_days)
         else:
-            tier = "Low / Emerging"
+            out["signal_age_days"] = np.nan
 
-        return {
-            "composite_confidence": composite,
-            "confidence_tier": tier,
-            "sub_scores": {
-                "youtube": yt_score,
-                "musicbrainz": mb_score,
-                "wikidata": wd_score,
-                "cross_referenced": cross_ref_bonus > 0.0,
-            },
-        }
+        # Decay instagram_confidence_score where applicable
+        if "instagram_confidence_score" in out.columns:
+            out["instagram_confidence_score"] = out.apply(
+                lambda row: cls.apply_staleness_decay(
+                    row.get("instagram_confidence_score", 0.0), 
+                    row.get("signal_age_days")
+                ),
+                axis=1
+            )
+        return out