Files

167 lines
6.4 KiB
Python
Raw Permalink Normal View History

"""
core/health_detector.py
Automated pipeline health checks, anomaly detection, and scraper circuit breaker.
Enforces zero-imputation validation, metric range checks, and quarantine governance.
"""
from datetime import datetime, timezone
from pathlib import Path
import sys
import time
from typing import Dict, List, Optional
import numpy as np
import pandas as pd
PROJECT_ROOT = Path(__file__).resolve().parent.parent
if str(PROJECT_ROOT) not in sys.path:
sys.path.insert(0, str(PROJECT_ROOT))
DEFAULT_DATASET = PROJECT_ROOT / "data" / "hiffi_multiplatform_500.csv"
class ScraperCircuitBreaker:
"""
Tracks successive scraping or network failures.
Opens circuit when failure threshold is exceeded to prevent cascading bans.
"""
def __init__(self, failure_threshold: int = 5, cooldown_seconds: int = 30):
self.failure_threshold = failure_threshold
self.cooldown_seconds = cooldown_seconds
self.failure_count = 0
self.last_failure_time: Optional[float] = None
self.state = "CLOSED" # CLOSED, OPEN, HALF-OPEN
def record_failure(self, error_msg: str = ""):
self.failure_count += 1
self.last_failure_time = time.time()
print(f"[CircuitBreaker] Warning: Failure #{self.failure_count}/{self.failure_threshold} ({error_msg})")
if self.failure_count >= self.failure_threshold:
self.state = "OPEN"
print(f"[CircuitBreaker] Circuit TRIPPED to OPEN. Entering {self.cooldown_seconds}s cooldown.")
def record_success(self):
self.failure_count = 0
self.state = "CLOSED"
def is_tripped(self) -> bool:
if self.state == "OPEN":
if self.last_failure_time and (time.time() - self.last_failure_time > self.cooldown_seconds):
self.state = "HALF-OPEN"
return False
return True
return False
class HealthAnomalyDetector:
"""
Evaluates cohort datasets for schema compliance, zero-imputation violations,
unexpected metric shifts, and missing critical platform handles.
"""
def __init__(self, dataset_path: Optional[Path] = None):
self.dataset_path = dataset_path or DEFAULT_DATASET
def run_audit(self) -> pd.DataFrame:
if not self.dataset_path.exists():
print(f"[HealthDetector] Error: Dataset not found at {self.dataset_path}")
return pd.DataFrame()
df = pd.read_csv(self.dataset_path)
anomalies: List[Dict[str, object]] = []
# 1. Zero-Imputation Governance Checks: Detect fabricated zero-values
if "spotify_monthly_listeners" in df.columns:
zero_listeners = df[df["spotify_monthly_listeners"] == 0]
for _, r in zero_listeners.iterrows():
anomalies.append({
"name": r["name"],
"type": "IMPUTED_ZERO_STREAMING",
"production_rank": r.get("production_hiffi_rank"),
"value": 0,
"handle": r.get("spotify_id"),
"severity": "CRITICAL"
})
# 2. Instagram Top-50 Missing Verification Flags
if "production_hiffi_rank" in df.columns and "instagram_verification_status" in df.columns:
top_50 = df[df["production_hiffi_rank"] <= 50]
for _, r in top_50.iterrows():
status = str(r.get("instagram_verification_status", ""))
followers = r.get("instagram_followers")
if status not in ("MANUAL_VERIFIED", "AUTOMATED_VERIFIED") or pd.isna(followers):
anomalies.append({
"name": r["name"],
"type": "TOP_50_MISSING_INSTAGRAM",
"production_rank": r["production_hiffi_rank"],
"value": followers,
"handle": r.get("instagram_handle"),
"severity": "MEDIUM"
})
# 3. Suspicious Follower Quarantine Violations (< 10k followers marked verified)
if "instagram_followers" in df.columns and "instagram_verification_status" in df.columns:
low_vol_verified = df[
(df["instagram_verification_status"] == "AUTOMATED_VERIFIED") &
(df["instagram_followers"] < 10000)
]
for _, r in low_vol_verified.iterrows():
anomalies.append({
"name": r["name"],
"type": "QUARANTINE_VIOLATION_LOW_FOLLOWERS",
"production_rank": r.get("production_hiffi_rank"),
"value": r["instagram_followers"],
"handle": r.get("instagram_handle"),
"severity": "HIGH"
})
# 4. Normalization Range Boundaries Checks ([0.0, 100.0])
if "hiffi_xp_score" in df.columns:
out_of_bounds = df[(df["hiffi_xp_score"] < 0.0) | (df["hiffi_xp_score"] > 100.0)]
for _, r in out_of_bounds.iterrows():
anomalies.append({
"name": r["name"],
"type": "SCORE_OUT_OF_BOUNDS",
"production_rank": r.get("production_hiffi_rank"),
"value": r["hiffi_xp_score"],
"handle": None,
"severity": "CRITICAL"
})
return pd.DataFrame(anomalies)
def main():
print("=" * 85)
print("HIFFI PIPELINE AUTOMATED HEALTH & ANOMALY DETECTOR")
print("=" * 85)
detector = HealthAnomalyDetector()
report = detector.run_audit()
if not DEFAULT_DATASET.exists():
print(f"[HealthDetector] Target dataset missing: {DEFAULT_DATASET}")
return
df = pd.read_csv(DEFAULT_DATASET)
total_rows = len(df)
total_anomalies = len(report)
critical_count = len(report[report["severity"] == "CRITICAL"]) if not report.empty else 0
print(f"Total Cohort Rows Evaluated: {total_rows}")
print(f"Total Anomalies Flagged: {total_anomalies}")
print(f"Actionable Unresolved (P0/P1): {critical_count}")
print(f"Data Governance Breaches: {critical_count}")
print("-" * 85)
if not report.empty:
display_cols = ["name", "type", "production_rank", "value", "handle", "severity"]
cols = [c for c in display_cols if c in report.columns]
print(report[cols].to_string(index=False))
else:
print("[HealthDetector] All governance checks passed. Zero anomalies detected.")
print("=" * 85)
if __name__ == "__main__":
main()