878 lines
35 KiB
Python
878 lines
35 KiB
Python
"""
|
|
test_multiplatform_ranking.py
|
|
Comprehensive QA & Data Governance Test Suite for HIFFI Multi-Platform Pipeline.
|
|
Enforces zero-imputation policy, score bounding, identity resolution,
|
|
fallbacks, strict null retention, publication eligibility gates,
|
|
and ChartMasters record sales governance.
|
|
"""
|
|
|
|
import inspect
|
|
from pathlib import Path
|
|
import sys
|
|
import numpy as np
|
|
import pandas as pd
|
|
import pytest
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parent
|
|
if str(PROJECT_ROOT) not in sys.path:
|
|
sys.path.insert(0, str(PROJECT_ROOT))
|
|
|
|
from core.ranking_engine import HIFFIRankingEngine
|
|
|
|
DATASET_PATH = PROJECT_ROOT / "data" / "hiffi_multiplatform_500.csv"
|
|
DATASET_ANCHOR_TIME = pd.Timestamp("2026-10-26T12:00:00Z")
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def dataset():
|
|
if not DATASET_PATH.exists():
|
|
pytest.fail(f"Cohort dataset not found at {DATASET_PATH}")
|
|
return pd.read_csv(DATASET_PATH)
|
|
|
|
|
|
def test_cohort_size(dataset):
|
|
"""Ensures exact 500-artist cohort size."""
|
|
assert len(dataset) == 500, f"Expected 500 artists, found {len(dataset)}"
|
|
|
|
|
|
def test_spotify_streaming_governance(dataset):
|
|
"""Zero-Imputation Policy: spotify_monthly_listeners must not contain imputed zeroes."""
|
|
if "spotify_monthly_listeners" in dataset.columns:
|
|
zero_count = (dataset["spotify_monthly_listeners"] == 0).sum()
|
|
assert zero_count == 0, f"Found {zero_count} imputed 0 values in spotify_monthly_listeners."
|
|
|
|
|
|
def test_missing_metrics_remain_nan(dataset):
|
|
"""Missing metrics must strictly be preserved as NaN."""
|
|
for col in ["spotify_monthly_listeners", "instagram_followers"]:
|
|
if col in dataset.columns:
|
|
assert dataset[col].isna().sum() > 0, f"Expected missing values in {col} to be NaN."
|
|
|
|
|
|
def test_production_ranking_preserved(dataset):
|
|
"""Production ranking baseline must remain unmodified."""
|
|
assert "production_hiffi_rank" in dataset.columns
|
|
assert "production_hiffi_score" in dataset.columns
|
|
valid_prod = dataset["production_hiffi_score"].dropna()
|
|
assert (valid_prod >= 0.0).all() and (valid_prod <= 100.0).all()
|
|
|
|
|
|
def test_score_normalization_bounds(dataset):
|
|
"""HIFFI-XP composite scores must be strictly bounded in [0.0, 100.0]."""
|
|
if "hiffi_xp_score" in dataset.columns:
|
|
valid_scores = dataset["hiffi_xp_score"].dropna()
|
|
assert (valid_scores >= 0.0).all() and (valid_scores <= 100.0).all(), (
|
|
f"XP Scores outside [0, 100] bounds: min={valid_scores.min()}, max={valid_scores.max()}"
|
|
)
|
|
|
|
|
|
def test_rank_monotonicity(dataset):
|
|
"""Higher HIFFI-XP score must yield better (numerically lower) rank."""
|
|
if "hiffi_xp_score" in dataset.columns and "hiffi_xp_rank" in dataset.columns:
|
|
sorted_df = dataset.dropna(subset=["hiffi_xp_score", "hiffi_xp_rank"]).sort_values(by="hiffi_xp_rank")
|
|
scores = sorted_df["hiffi_xp_score"].to_numpy()
|
|
assert (np.diff(scores) <= 1e-5).all(), "Rank monotonicity violated."
|
|
|
|
|
|
def test_schema_provenance_columns(dataset):
|
|
"""All required audit and provenance columns must exist in schema."""
|
|
required = [
|
|
"name", "production_hiffi_score", "production_hiffi_rank",
|
|
"spotify_monthly_listeners", "spotify_daily_change",
|
|
"instagram_followers", "instagram_verification_status",
|
|
"hiffi_xp_score", "hiffi_xp_rank", "rank_delta"
|
|
]
|
|
for col in required:
|
|
assert col in dataset.columns, f"Missing required column: {col}"
|
|
|
|
|
|
def _invoke_instagram_verifier(verifier, artist_name, candidate_handle, profile_data):
|
|
"""Dynamically adapts arguments to match evaluate_candidate or verify_candidate signature."""
|
|
method = None
|
|
for attr in ["evaluate_candidate", "verify_candidate", "verify_artist", "verify"]:
|
|
if hasattr(verifier, attr):
|
|
method = getattr(verifier, attr)
|
|
break
|
|
|
|
if method is None:
|
|
raise AttributeError("Could not find candidate verification method on verifier.")
|
|
|
|
sig = inspect.signature(method)
|
|
params = sig.parameters
|
|
|
|
kwargs = {}
|
|
for p in ["artist_name", "artist", "name"]:
|
|
if p in params:
|
|
kwargs[p] = artist_name
|
|
break
|
|
|
|
for p in ["candidate_handle", "handle", "instagram_handle"]:
|
|
if p in params:
|
|
kwargs[p] = candidate_handle
|
|
break
|
|
|
|
for p in ["profile_data", "profile", "data"]:
|
|
if p in params:
|
|
kwargs[p] = profile_data
|
|
break
|
|
|
|
for p in ["follower_count", "followers"]:
|
|
if p in params:
|
|
kwargs[p] = profile_data.get("followers", 0)
|
|
break
|
|
|
|
for p in ["bio", "biography", "candidate_bio"]:
|
|
if p in params:
|
|
kwargs[p] = profile_data.get("biography", "")
|
|
break
|
|
|
|
for p in ["is_verified", "verified"]:
|
|
if p in params:
|
|
kwargs[p] = profile_data.get("is_verified", False)
|
|
break
|
|
|
|
return method(**kwargs)
|
|
|
|
|
|
def test_instagram_verifier_candidate_matching():
|
|
"""Test algorithmic confidence score calculation for Instagram handle matching."""
|
|
from collectors.instagram_verifier import InstagramCandidateVerifier
|
|
|
|
verifier = InstagramCandidateVerifier()
|
|
res = _invoke_instagram_verifier(
|
|
verifier,
|
|
artist_name="Kendrick Lamar",
|
|
candidate_handle="kendricklamar",
|
|
profile_data={"followers": 15000000, "biography": "pgLang", "full_name": "Kendrick Lamar"}
|
|
)
|
|
status = res.get("status") or res.get("verification_status") or res.get("instagram_verification_status")
|
|
conf = res.get("confidence_score") or res.get("instagram_confidence_score", 0.0)
|
|
evidence = str(res.get("evidence") or res.get("audit_evidence") or res.get("instagram_audit_evidence", ""))
|
|
|
|
assert status == "AUTOMATED_VERIFIED"
|
|
assert conf >= 0.70
|
|
assert "kendricklamar" in evidence.lower() or "exact" in evidence.lower() or len(evidence) > 0
|
|
|
|
|
|
def test_instagram_verifier_suspicious_follower_count():
|
|
"""Ensure handles with fewer than 10,000 followers are quarantined."""
|
|
from collectors.instagram_verifier import InstagramCandidateVerifier
|
|
|
|
verifier = InstagramCandidateVerifier()
|
|
res = _invoke_instagram_verifier(
|
|
verifier,
|
|
artist_name="Drake",
|
|
candidate_handle="drake",
|
|
profile_data={"followers": 2500, "biography": "Fan club", "full_name": "Aubrey Drake"}
|
|
)
|
|
status = res.get("status") or res.get("verification_status") or res.get("instagram_verification_status")
|
|
followers = res.get("followers") or res.get("instagram_followers")
|
|
|
|
assert status in ("UNCERTAIN_NEEDS_REVIEW", "REJECTED_FAN_PAGE", "QUARANTINE_LOW_FOLLOWERS")
|
|
assert followers is None or status != "AUTOMATED_VERIFIED"
|
|
|
|
|
|
def test_spotify_fallback_null_safety(monkeypatch):
|
|
"""Verify Spotify embed fallback safely returns structured null dictionary on unindexed items."""
|
|
from collectors.spotify_fallback_collector import SpotifyFallbackCollector
|
|
|
|
class MockResponse:
|
|
status_code = 404
|
|
text = "Not Found"
|
|
|
|
monkeypatch.setattr("requests.Session.get", lambda *args, **kwargs: MockResponse())
|
|
|
|
fallback = SpotifyFallbackCollector()
|
|
res = fallback.fetch_artist_fallback("Unknown Artist", "spotify:artist:invalid_or_missing_id_999999")
|
|
assert res["spotify_monthly_listeners"] is None
|
|
assert res["spotify_daily_change"] is None
|
|
assert "NOT_FOUND" in str(res["spotify_status"]) or "UNAVAILABLE" in str(res["spotify_status"])
|
|
|
|
|
|
def test_spotify_fallback_mocked_network_error(monkeypatch):
|
|
"""Verify Spotify fallback circuit breaker / exception handling gracefully returns null values."""
|
|
from collectors.spotify_fallback_collector import SpotifyFallbackCollector
|
|
import requests
|
|
|
|
def mock_get(*args, **kwargs):
|
|
raise requests.exceptions.ConnectionError("Mocked network failure")
|
|
|
|
monkeypatch.setattr(requests.Session, "get", mock_get)
|
|
|
|
fallback = SpotifyFallbackCollector()
|
|
res = fallback.fetch_artist_fallback("Kendrick Lamar", "spotify:artist:2YZyLo08N0Wb9xBt1NhZWg")
|
|
assert res["spotify_monthly_listeners"] is None
|
|
assert res["spotify_daily_change"] is None
|
|
|
|
|
|
def test_duplicate_records_handled_safely():
|
|
"""Ranking engine handles duplicate artist monikers without crashing."""
|
|
engine = HIFFIRankingEngine()
|
|
test_df = pd.DataFrame([
|
|
{"name": "Baby", "production_hiffi_score": 75.0, "production_hiffi_rank": 10, "spotify_monthly_listeners": 15000000, "spotify_daily_change": 10000, "instagram_followers": 5000000},
|
|
{"name": "Baby", "production_hiffi_score": 40.0, "production_hiffi_rank": 150, "spotify_monthly_listeners": 200000, "spotify_daily_change": 100, "instagram_followers": 50000}
|
|
])
|
|
scored = engine.calculate_hiffi_xp(test_df)
|
|
assert len(scored) == 2
|
|
assert scored.loc[0, "hiffi_xp_score"] > scored.loc[1, "hiffi_xp_score"]
|
|
|
|
|
|
def test_spotify_api_collector_unconfigured(monkeypatch):
|
|
"""Verify that unconfigured credentials safely return null values under zero-imputation."""
|
|
from collectors.spotify_api_collector import SpotifyAPICollector
|
|
|
|
monkeypatch.delenv("SPOTIFY_CLIENT_ID", raising=False)
|
|
monkeypatch.delenv("SPOTIFY_CLIENT_SECRET", raising=False)
|
|
collector = SpotifyAPICollector(client_id="", client_secret="")
|
|
res = collector.fetch_artist("test_id")
|
|
assert res["spotify_status"] == "API_NOT_CONFIGURED"
|
|
assert res["spotify_followers"] is None
|
|
assert res["spotify_popularity"] is None
|
|
|
|
|
|
def test_spotify_api_collector_empty_id():
|
|
"""Verify that empty or invalid artist IDs safely yield None without throwing errors."""
|
|
from collectors.spotify_api_collector import SpotifyAPICollector
|
|
|
|
collector = SpotifyAPICollector(client_id="dummy", client_secret="dummy")
|
|
res = collector.fetch_artist("")
|
|
assert res["spotify_followers"] is None
|
|
assert res["spotify_status"] in ("INVALID_ID", "API_NOT_CONFIGURED")
|
|
|
|
|
|
def test_dynamic_weight_redistribution_fairness():
|
|
"""Ensure artists with missing platforms receive fair proportional scores bounded in [0, 100]."""
|
|
engine = HIFFIRankingEngine()
|
|
test_df = pd.DataFrame([
|
|
{
|
|
"name": "Artist All Signals",
|
|
"production_hiffi_score": 80.0,
|
|
"production_hiffi_rank": 10,
|
|
"spotify_monthly_listeners": 20_000_000,
|
|
"spotify_daily_change": 50_000,
|
|
"instagram_followers": 5_000_000,
|
|
},
|
|
{
|
|
"name": "Artist Missing IG",
|
|
"production_hiffi_score": 80.0,
|
|
"production_hiffi_rank": 10,
|
|
"spotify_monthly_listeners": 20_000_000,
|
|
"spotify_daily_change": 50_000,
|
|
"instagram_followers": np.nan,
|
|
},
|
|
{
|
|
"name": "Artist YouTube Only",
|
|
"production_hiffi_score": 80.0,
|
|
"production_hiffi_rank": 10,
|
|
"spotify_monthly_listeners": np.nan,
|
|
"spotify_daily_change": np.nan,
|
|
"instagram_followers": np.nan,
|
|
}
|
|
])
|
|
|
|
ranked = engine.calculate_hiffi_xp(test_df)
|
|
assert ranked["hiffi_xp_score"].notna().all(), "Redistribution produced unexpected NaN scores"
|
|
assert (ranked["hiffi_xp_score"] >= 0.0).all() and (ranked["hiffi_xp_score"] <= 100.0).all()
|
|
assert ranked.loc[1, "hiffi_xp_score"] > 0.0
|
|
|
|
|
|
def test_score_dominance(dataset):
|
|
"""If Artist A >= Artist B across all present metrics, Score(A) must be >= Score(B)."""
|
|
clean_sub = dataset.dropna(subset=[
|
|
"production_hiffi_score",
|
|
"spotify_monthly_listeners",
|
|
"instagram_followers",
|
|
"hiffi_xp_score"
|
|
]).copy()
|
|
|
|
if len(clean_sub) < 5:
|
|
pytest.skip("Not enough fully populated records to run dominance sample.")
|
|
|
|
sample = clean_sub.head(25).to_dict("records")
|
|
for a in sample:
|
|
for b in sample:
|
|
if (
|
|
a["production_hiffi_score"] >= b["production_hiffi_score"] and
|
|
a["spotify_monthly_listeners"] >= b["spotify_monthly_listeners"] and
|
|
a["instagram_followers"] >= b["instagram_followers"]
|
|
):
|
|
assert a["hiffi_xp_score"] >= (b["hiffi_xp_score"] - 1e-4), (
|
|
f"Dominance violated: {a['name']} ({a['hiffi_xp_score']}) < {b['name']} ({b['hiffi_xp_score']})"
|
|
)
|
|
|
|
|
|
def test_kendall_tau_diagnostic(dataset):
|
|
"""Evaluates rank correlation between baseline and composite rank as a diagnostic."""
|
|
from scipy.stats import kendalltau
|
|
|
|
valid = dataset.dropna(subset=["production_hiffi_rank", "hiffi_xp_rank"]).copy()
|
|
if len(valid) < 10:
|
|
pytest.skip("Insufficient ranked rows for Kendall Tau calculation.")
|
|
|
|
tau, p_val = kendalltau(valid["production_hiffi_rank"], valid["hiffi_xp_rank"])
|
|
assert tau > 0.10, f"Rank correlation abnormally low or inverted: tau={tau:.3f}"
|
|
|
|
|
|
def test_score_quantile_spread(dataset):
|
|
"""Ensures composite scores do not clump into a single degenerate value."""
|
|
if "hiffi_xp_score" in dataset.columns:
|
|
valid_scores = dataset["hiffi_xp_score"].dropna()
|
|
if len(valid_scores) > 10:
|
|
assert valid_scores.std() > 5.0, f"Scores exhibit degenerate clumping: std={valid_scores.std():.2f}"
|
|
|
|
|
|
def test_raw_dataset_strict_nan_retention(dataset):
|
|
"""Verifies that unindexed secondary fields preserve genuine nulls and guard against synthetic zeroes."""
|
|
secondary_metric_cols = [
|
|
"spotify_monthly_listeners",
|
|
"spotify_daily_change",
|
|
"instagram_followers"
|
|
]
|
|
|
|
for col in secondary_metric_cols:
|
|
if col in dataset.columns:
|
|
series = dataset[col]
|
|
assert series.isna().any(), f"Column '{col}' contains zero null values across cohort."
|
|
|
|
string_nulls = series.dropna().astype(str).str.strip().str.lower().isin(
|
|
["nan", "none", "null", "undefined", "n/a"]
|
|
)
|
|
assert not string_nulls.any(), f"Column '{col}' contains string-serialized null tokens."
|
|
|
|
numeric_vals = pd.to_numeric(series.dropna(), errors="coerce")
|
|
if col in ["spotify_monthly_listeners", "instagram_followers"]:
|
|
zero_records = (numeric_vals == 0).sum()
|
|
assert zero_records == 0, (
|
|
f"Column '{col}' has {zero_records} entries with literal zero (0). "
|
|
"Unindexed platforms must remain NaN."
|
|
)
|
|
|
|
|
|
def test_publication_eligibility_gates(dataset):
|
|
"""Verifies that publication gates enforce evidence tiers using anchored timestamps without mutating raw scores."""
|
|
engine = HIFFIRankingEngine()
|
|
gated = engine.apply_publication_eligibility_gates(dataset, reference_time=DATASET_ANCHOR_TIME)
|
|
|
|
assert (gated["hiffi_xp_score"] == dataset["hiffi_xp_score"]).all(), (
|
|
"Publication gates altered the underlying raw composite scores."
|
|
)
|
|
|
|
top_10 = gated[gated["hiffi_xp_published_rank"] <= 10]
|
|
assert len(top_10) == 10
|
|
assert (top_10["platform_evidence_tier"] == 3).all(), (
|
|
f"Found artists with < 3 platforms in Top 10 published ranks: "
|
|
f"{top_10[top_10['platform_evidence_tier'] < 3][['name', 'platform_evidence_tier']]}"
|
|
)
|
|
|
|
top_50 = gated[gated["hiffi_xp_published_rank"] <= 50]
|
|
assert len(top_50) == 50
|
|
assert (top_50["platform_evidence_tier"] >= 2).all(), (
|
|
f"Found single-platform artists in Top 50 published ranks: "
|
|
f"{top_50[top_50['platform_evidence_tier'] < 2][['name', 'platform_evidence_tier']]}"
|
|
)
|
|
|
|
ranks = gated["hiffi_xp_published_rank"].sort_values().tolist()
|
|
expected_ranks = list(range(1, len(gated) + 1))
|
|
assert ranks == expected_ranks, "Published ranks contain gaps or duplicates."
|
|
|
|
|
|
def test_publication_gate_blocks_failed_youtube_lookup():
|
|
"""Confirms artists with status_flag == 'FAILED_LOOKUP' are disqualified from YouTube evidence credit."""
|
|
engine = HIFFIRankingEngine()
|
|
now_iso = DATASET_ANCHOR_TIME.isoformat()
|
|
|
|
test_df = pd.DataFrame([
|
|
{
|
|
"name": "Failed Lookup Artist",
|
|
"status_flag": "FAILED_LOOKUP",
|
|
"production_hiffi_score": 7.98,
|
|
"youtube_updated_at": now_iso,
|
|
"spotify_monthly_listeners": 25_000_000,
|
|
"spotify_updated_at": now_iso,
|
|
"instagram_followers": 10_000_000,
|
|
"instagram_verification_status": "AUTOMATED_VERIFIED",
|
|
"instagram_updated_at": now_iso,
|
|
"hiffi_xp_score": 85.0,
|
|
"hiffi_xp_rank": 1
|
|
}
|
|
])
|
|
|
|
gated = engine.apply_publication_eligibility_gates(test_df, reference_time=DATASET_ANCHOR_TIME)
|
|
row = gated.iloc[0]
|
|
|
|
assert row["platform_evidence_tier"] == 2
|
|
assert "FAILED_YOUTUBE_LOOKUP" in row["evidence_exclusions"]
|
|
|
|
|
|
def test_publication_gate_blocks_unverified_identity():
|
|
"""Confirms that UNCERTAIN_NEEDS_REVIEW status does not count toward evidence tiers."""
|
|
engine = HIFFIRankingEngine()
|
|
now_iso = DATASET_ANCHOR_TIME.isoformat()
|
|
|
|
test_df = pd.DataFrame([
|
|
{
|
|
"name": "Artist Pending Verification",
|
|
"status_flag": "COMPLETE",
|
|
"production_hiffi_score": 95.0,
|
|
"youtube_updated_at": now_iso,
|
|
"spotify_monthly_listeners": 40_000_000,
|
|
"spotify_updated_at": now_iso,
|
|
"instagram_followers": 20_000_000,
|
|
"instagram_verification_status": "UNCERTAIN_NEEDS_REVIEW",
|
|
"instagram_updated_at": now_iso,
|
|
"hiffi_xp_score": 96.0,
|
|
"hiffi_xp_rank": 1
|
|
},
|
|
{
|
|
"name": "Artist Fully Verified",
|
|
"status_flag": "COMPLETE",
|
|
"production_hiffi_score": 90.0,
|
|
"youtube_updated_at": now_iso,
|
|
"spotify_monthly_listeners": 35_000_000,
|
|
"spotify_updated_at": now_iso,
|
|
"instagram_followers": 15_000_000,
|
|
"instagram_verification_status": "AUTOMATED_VERIFIED",
|
|
"instagram_updated_at": now_iso,
|
|
"hiffi_xp_score": 91.0,
|
|
"hiffi_xp_rank": 2
|
|
}
|
|
])
|
|
|
|
gated = engine.apply_publication_eligibility_gates(test_df, reference_time=DATASET_ANCHOR_TIME)
|
|
pending_row = gated[gated["name"] == "Artist Pending Verification"].iloc[0]
|
|
verified_row = gated[gated["name"] == "Artist Fully Verified"].iloc[0]
|
|
|
|
assert pending_row["platform_evidence_tier"] == 2
|
|
assert verified_row["platform_evidence_tier"] == 3
|
|
|
|
|
|
def test_publication_gate_missing_column_fails_closed():
|
|
"""Confirms that omitting the verification column fails closed (gives 0 IG credit)."""
|
|
engine = HIFFIRankingEngine()
|
|
now_iso = DATASET_ANCHOR_TIME.isoformat()
|
|
|
|
test_df = pd.DataFrame([
|
|
{
|
|
"name": "Artist No Status Column",
|
|
"status_flag": "COMPLETE",
|
|
"production_hiffi_score": 95.0,
|
|
"youtube_updated_at": now_iso,
|
|
"spotify_monthly_listeners": 40_000_000,
|
|
"spotify_updated_at": now_iso,
|
|
"instagram_followers": 20_000_000,
|
|
"instagram_updated_at": now_iso,
|
|
"hiffi_xp_score": 96.0,
|
|
"hiffi_xp_rank": 1
|
|
}
|
|
])
|
|
|
|
gated = engine.apply_publication_eligibility_gates(test_df, reference_time=DATASET_ANCHOR_TIME)
|
|
assert gated.loc[0, "platform_evidence_tier"] == 2
|
|
|
|
|
|
def test_publication_gate_zero_multiplatform_cohort():
|
|
"""Confirms an entirely Tier 1 cohort produces consecutive ranks and flags an audited shortfall."""
|
|
engine = HIFFIRankingEngine()
|
|
now_iso = DATASET_ANCHOR_TIME.isoformat()
|
|
|
|
records = [
|
|
{
|
|
"name": f"Solo Artist {i+1}",
|
|
"status_flag": "COMPLETE",
|
|
"production_hiffi_score": 85.0 - i,
|
|
"youtube_updated_at": now_iso,
|
|
"spotify_monthly_listeners": np.nan,
|
|
"spotify_updated_at": np.nan,
|
|
"instagram_followers": np.nan,
|
|
"instagram_verification_status": "UNVERIFIED",
|
|
"instagram_updated_at": np.nan,
|
|
"hiffi_xp_score": 80.0 - i,
|
|
"hiffi_xp_rank": i + 1
|
|
}
|
|
for i in range(25)
|
|
]
|
|
|
|
tier1_df = pd.DataFrame(records)
|
|
gated, audit = engine.apply_publication_eligibility_gates(tier1_df, reference_time=DATASET_ANCHOR_TIME, return_metadata=True)
|
|
|
|
assert gated["hiffi_xp_published_rank"].tolist() == list(range(1, 26))
|
|
assert (gated["platform_evidence_tier"] == 1).all()
|
|
assert audit["top_10_shortfall"] is True
|
|
assert audit["top_50_shortfall"] is True
|
|
|
|
|
|
def test_freshness_gate_stale_timestamps_demote_tier():
|
|
"""Validates that metrics older than threshold lose evidence tier credit while retaining raw score."""
|
|
engine = HIFFIRankingEngine()
|
|
ref_time = DATASET_ANCHOR_TIME
|
|
|
|
test_df = pd.DataFrame([{
|
|
"name": "Artist Stale Telemetry",
|
|
"status_flag": "COMPLETE",
|
|
"production_hiffi_score": 90.0,
|
|
"youtube_updated_at": "2026-10-25T12:00:00Z",
|
|
"spotify_monthly_listeners": 25_000_000,
|
|
"spotify_updated_at": "2026-10-15T12:00:00Z",
|
|
"instagram_followers": 10_000_000,
|
|
"instagram_verification_status": "AUTOMATED_VERIFIED",
|
|
"instagram_updated_at": "2026-10-01T12:00:00Z",
|
|
"hiffi_xp_score": 88.5,
|
|
"hiffi_xp_rank": 5
|
|
}])
|
|
|
|
gated = engine.apply_publication_eligibility_gates(test_df, reference_time=ref_time)
|
|
row = gated.iloc[0]
|
|
|
|
assert row["platform_evidence_tier"] == 1
|
|
assert row["hiffi_xp_score"] == 88.5
|
|
assert "STALE_OR_MISSING_SPOTIFY_TIMESTAMP" in row["evidence_exclusions"]
|
|
assert "STALE_OR_MISSING_INSTAGRAM_TIMESTAMP" in row["evidence_exclusions"]
|
|
|
|
|
|
def test_freshness_gate_missing_or_malformed_timestamp():
|
|
"""Fails closed when timestamp string is empty, null, or malformed."""
|
|
engine = HIFFIRankingEngine()
|
|
ref_time = DATASET_ANCHOR_TIME
|
|
|
|
test_df = pd.DataFrame([{
|
|
"name": "Artist Malformed Timestamp",
|
|
"status_flag": "COMPLETE",
|
|
"production_hiffi_score": 85.0,
|
|
"youtube_updated_at": "invalid_date_format",
|
|
"spotify_monthly_listeners": 15_000_000,
|
|
"spotify_updated_at": None,
|
|
"instagram_followers": 5_000_000,
|
|
"instagram_verification_status": "AUTOMATED_VERIFIED",
|
|
"instagram_updated_at": "2026-10-25T12:00:00Z",
|
|
"hiffi_xp_score": 82.0,
|
|
"hiffi_xp_rank": 10
|
|
}])
|
|
|
|
gated = engine.apply_publication_eligibility_gates(test_df, reference_time=ref_time)
|
|
row = gated.iloc[0]
|
|
|
|
assert row["platform_evidence_tier"] == 1
|
|
assert "STALE_OR_MISSING_YOUTUBE_TIMESTAMP" in row["evidence_exclusions"]
|
|
assert "STALE_OR_MISSING_SPOTIFY_TIMESTAMP" in row["evidence_exclusions"]
|
|
|
|
|
|
def test_freshness_gate_missing_timestamp_columns_fail_closed():
|
|
"""Confirms that omitting timestamp columns fails closed to 0 tier credit."""
|
|
engine = HIFFIRankingEngine()
|
|
|
|
test_df = pd.DataFrame([{
|
|
"name": "Artist No Timestamps",
|
|
"status_flag": "COMPLETE",
|
|
"production_hiffi_score": 90.0,
|
|
"spotify_monthly_listeners": 20_000_000,
|
|
"instagram_followers": 10_000_000,
|
|
"instagram_verification_status": "AUTOMATED_VERIFIED",
|
|
"hiffi_xp_score": 88.0,
|
|
"hiffi_xp_rank": 1
|
|
}])
|
|
|
|
gated = engine.apply_publication_eligibility_gates(test_df, reference_time=DATASET_ANCHOR_TIME)
|
|
row = gated.iloc[0]
|
|
|
|
assert row["platform_evidence_tier"] == 0
|
|
assert "STALE_OR_MISSING_YOUTUBE_TIMESTAMP" in row["evidence_exclusions"]
|
|
assert "STALE_OR_MISSING_SPOTIFY_TIMESTAMP" in row["evidence_exclusions"]
|
|
assert "STALE_OR_MISSING_INSTAGRAM_TIMESTAMP" in row["evidence_exclusions"]
|
|
|
|
|
|
def test_anomaly_detection_flags_suspicious_patterns():
|
|
"""Validates that ratio anomalies are flagged into anomaly_flags without mutating scores."""
|
|
engine = HIFFIRankingEngine()
|
|
|
|
test_df = pd.DataFrame([
|
|
{
|
|
"name": "Organic Artist",
|
|
"lifetime_views": 100_000_000,
|
|
"instagram_followers": 5_000_000,
|
|
"spotify_monthly_listeners": 10_000_000,
|
|
"spotify_daily_change": 50_000,
|
|
"hiffi_xp_score": 85.0
|
|
},
|
|
{
|
|
"name": "Inorganic IG Artist",
|
|
"lifetime_views": 100_000,
|
|
"instagram_followers": 5_000_000,
|
|
"spotify_monthly_listeners": 500_000,
|
|
"spotify_daily_change": 1_000,
|
|
"hiffi_xp_score": 70.0
|
|
},
|
|
{
|
|
"name": "Spike Velocity Artist",
|
|
"lifetime_views": 50_000_000,
|
|
"instagram_followers": 1_000_000,
|
|
"spotify_monthly_listeners": 100_000,
|
|
"spotify_daily_change": 40_000,
|
|
"hiffi_xp_score": 65.0
|
|
}
|
|
])
|
|
|
|
flagged = engine.detect_metric_anomalies(test_df)
|
|
|
|
assert flagged.loc[0, "anomaly_flags"] == "CLEAN"
|
|
assert "SUSPICIOUS_IG_TO_YT_RATIO" in flagged.loc[1, "anomaly_flags"]
|
|
assert "EXTREME_SPOTIFY_VELOCITY_RATIO" in flagged.loc[2, "anomaly_flags"]
|
|
assert flagged.loc[1, "hiffi_xp_score"] == 70.0
|
|
|
|
|
|
# ==============================================================================
|
|
# CHARTMASTERS RECORD SALES GOVERNANCE & INTEGRITY TESTS
|
|
# ==============================================================================
|
|
|
|
MOCK_CHARTMASTERS_HTML = """
|
|
<html>
|
|
<body>
|
|
<table>
|
|
<thead>
|
|
<tr>
|
|
<th>Rank</th>
|
|
<th>Artist</th>
|
|
<th>Album Sales</th>
|
|
<th>Physical Singles</th>
|
|
<th>Digital Singles</th>
|
|
<th>Streaming EAS</th>
|
|
<th>Total EAS</th>
|
|
</tr>
|
|
</thead>
|
|
<tbody>
|
|
<tr>
|
|
<td>1</td>
|
|
<td>The Beatles</td>
|
|
<td>162,000,000</td>
|
|
<td>33,000,000</td>
|
|
<td>28,000,000</td>
|
|
<td>10,000,000</td>
|
|
<td>233,000,000</td>
|
|
</tr>
|
|
<tr>
|
|
<td>2</td>
|
|
<td>Michael Jackson</td>
|
|
<td>135,000,000</td>
|
|
<td>26,000,000</td>
|
|
<td>24,000,000</td>
|
|
<td>12,000,000</td>
|
|
<td>197,000,000</td>
|
|
</tr>
|
|
<tr>
|
|
<td>3</td>
|
|
<td>Eminem</td>
|
|
<td>63,000,000</td>
|
|
<td>9,000,000</td>
|
|
<td>45,000,000</td>
|
|
<td>60,000,000</td>
|
|
<td>177,000,000</td>
|
|
</tr>
|
|
<tr>
|
|
<td>4</td>
|
|
<td>A$AP Rocky</td>
|
|
<td>2,500,000</td>
|
|
<td>100,000</td>
|
|
<td>15,000,000</td>
|
|
<td>20,000,000</td>
|
|
<td>37,600,000</td>
|
|
</tr>
|
|
</tbody>
|
|
</table>
|
|
</body>
|
|
</html>
|
|
"""
|
|
|
|
|
|
def test_chartmasters_parser_schema_and_types():
|
|
"""Validates HTML table parsing, column typing, and metrics conversion."""
|
|
from collectors.chartmasters_collector import ChartMastersSalesCollector
|
|
|
|
collector = ChartMastersSalesCollector()
|
|
count = collector.load_sales_table(html_content=MOCK_CHARTMASTERS_HTML)
|
|
assert count == 4
|
|
|
|
beatles = collector.get_artist_sales_metrics("The Beatles")
|
|
assert beatles["sales_data_status"] == "VERIFIED_CSPC"
|
|
assert beatles["chartmasters_sales_rank"] == 1
|
|
assert beatles["total_eas_cspc"] == 233_000_000
|
|
assert beatles["pure_album_sales"] == 162_000_000
|
|
assert isinstance(beatles["streaming_eas"], int)
|
|
|
|
|
|
def test_chartmasters_stage_alias_resolution():
|
|
"""Validates stylized name alias resolution (e.g. A$AP Rocky -> asap rocky)."""
|
|
from collectors.chartmasters_collector import ChartMastersSalesCollector
|
|
|
|
collector = ChartMastersSalesCollector()
|
|
collector.load_sales_table(html_content=MOCK_CHARTMASTERS_HTML)
|
|
|
|
asap = collector.get_artist_sales_metrics("A$AP Rocky")
|
|
assert asap["sales_data_status"] == "VERIFIED_CSPC"
|
|
assert asap["chartmasters_sales_rank"] == 4
|
|
assert asap["total_eas_cspc"] == 37_600_000
|
|
|
|
|
|
def test_chartmasters_zero_imputation_null_retention():
|
|
"""Ensures uncataloged artists strictly return None and are never imputed with 0."""
|
|
from collectors.chartmasters_collector import ChartMastersSalesCollector
|
|
|
|
collector = ChartMastersSalesCollector()
|
|
collector.load_sales_table(html_content=MOCK_CHARTMASTERS_HTML)
|
|
|
|
unindexed = collector.get_artist_sales_metrics("Emerging Underground Artist 123")
|
|
assert unindexed["sales_data_status"] == "NOT_CATALOGED_YET"
|
|
assert unindexed["chartmasters_sales_rank"] is None
|
|
assert unindexed["total_eas_cspc"] is None
|
|
assert unindexed["pure_album_sales"] is None
|
|
|
|
|
|
def test_chartmasters_rank_monotonicity():
|
|
"""Higher sales rank (1 vs 2) must strictly correspond to higher or equal Total EAS."""
|
|
from collectors.chartmasters_collector import ChartMastersSalesCollector
|
|
|
|
collector = ChartMastersSalesCollector()
|
|
collector.load_sales_table(html_content=MOCK_CHARTMASTERS_HTML)
|
|
|
|
beatles = collector.get_artist_sales_metrics("The Beatles")
|
|
mj = collector.get_artist_sales_metrics("Michael Jackson")
|
|
eminem = collector.get_artist_sales_metrics("Eminem")
|
|
|
|
assert beatles["chartmasters_sales_rank"] < mj["chartmasters_sales_rank"] < eminem["chartmasters_sales_rank"]
|
|
assert beatles["total_eas_cspc"] >= mj["total_eas_cspc"] >= eminem["total_eas_cspc"]
|
|
|
|
|
|
def test_chartmasters_enrich_cohort_dataframe():
|
|
"""Tests zero-imputation cohort enrichment over a pandas DataFrame."""
|
|
from collectors.chartmasters_collector import ChartMastersSalesCollector
|
|
|
|
collector = ChartMastersSalesCollector()
|
|
collector.load_sales_table(html_content=MOCK_CHARTMASTERS_HTML)
|
|
|
|
cohort_df = pd.DataFrame([
|
|
{"name": "Eminem", "production_hiffi_score": 92.0},
|
|
{"name": "Non-Indexed Indie Artist", "production_hiffi_score": 45.0}
|
|
])
|
|
|
|
enriched = collector.enrich_cohort(cohort_df, name_column="name")
|
|
|
|
assert "total_eas_cspc" in enriched.columns
|
|
assert "chartmasters_sales_rank" in enriched.columns
|
|
|
|
# Row 0 (Eminem) has valid numeric metrics
|
|
assert enriched.loc[0, "total_eas_cspc"] == 177_000_000
|
|
assert enriched.loc[0, "chartmasters_sales_rank"] == 3
|
|
|
|
# Row 1 (Non-indexed) must be None/NaN, not 0
|
|
assert pd.isna(enriched.loc[1, "total_eas_cspc"])
|
|
assert pd.isna(enriched.loc[1, "chartmasters_sales_rank"])
|
|
assert enriched.loc[1, "sales_data_status"] == "NOT_CATALOGED_YET"
|
|
# ---------------------------------------------------------------------------
|
|
# Phase 3: Commercial Power Index (CPI) & Simulation Tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_cpi_mathematical_bounds_and_null_governance():
|
|
"""
|
|
Verifies that Commercial Power Index scales within [0.0, 100.0] anchored
|
|
to the global industry benchmarks and uncataloged artists strictly retain NaN.
|
|
"""
|
|
from core.cpi_engine import CommercialPowerEngine
|
|
|
|
engine = CommercialPowerEngine()
|
|
df = pd.DataFrame([
|
|
{"name": "Eminem", "total_eas_cspc": 177_000_000},
|
|
{"name": "Drake", "total_eas_cspc": 142_700_000},
|
|
{"name": "Snoop Dogg", "total_eas_cspc": 49_500_000},
|
|
{"name": "Emerging Act", "total_eas_cspc": None},
|
|
])
|
|
|
|
cpi = engine.calculate_cpi(df)
|
|
|
|
# Eminem (177M EAS) scales to ~95.59 on the 100k to 250M industry reference scale
|
|
assert cpi.iloc[0] == 95.59, f"Expected Eminem to be 95.59, got {cpi.iloc[0]}"
|
|
assert 85.0 < cpi.iloc[1] < cpi.iloc[0], "Drake must rank just below Eminem"
|
|
assert 70.0 < cpi.iloc[2] < cpi.iloc[1], "Snoop Dogg must be positive and bounded"
|
|
assert pd.isna(cpi.iloc[3]), "Uncataloged artist must strictly retain NaN for CPI"
|
|
|
|
|
|
def test_cpi_simulation_weight_redistribution_fairness():
|
|
"""
|
|
Verifies that simulated ranking dynamically redistributes 100% of the weight
|
|
back to the baseline score for uncataloged artists without synthetic dilution.
|
|
"""
|
|
from core.cpi_engine import CommercialPowerEngine
|
|
|
|
engine = CommercialPowerEngine()
|
|
df = pd.DataFrame([
|
|
{"name": "Artist A (Cataloged)", "hiffi_xp_score": 80.0, "hiffi_xp_published_rank": 2, "total_eas_cspc": 100_000_000},
|
|
{"name": "Artist B (Uncataloged)", "hiffi_xp_score": 85.0, "hiffi_xp_published_rank": 1, "total_eas_cspc": None},
|
|
])
|
|
|
|
sim = engine.simulate_composite_ranking(df, commercial_weight=0.20, mode="global_redistribution")
|
|
|
|
# Uncataloged artist must preserve baseline score without 20% penalty
|
|
uncataloged_score = sim.loc[sim["name"] == "Artist B (Uncataloged)", "simulated_hiffi_score"].values[0]
|
|
assert uncataloged_score == 85.0, f"Expected 85.0, got {uncataloged_score}"
|
|
|
|
# Cataloged artist must reflect 80% baseline + 20% CPI (100M EAS yields CPI of 88.29)
|
|
cataloged_score = sim.loc[sim["name"] == "Artist A (Cataloged)", "simulated_hiffi_score"].values[0]
|
|
expected_cpi = engine.calculate_cpi(df).iloc[0]
|
|
expected_cataloged = round((0.80 * 80.0) + (0.20 * expected_cpi), 2)
|
|
assert cataloged_score == expected_cataloged, f"Expected {expected_cataloged}, got {cataloged_score}"
|
|
def test_cpi_simulation_rank_monotonicity():
|
|
"""
|
|
Verifies that simulated ranks are strictly monotonic with respect to simulated scores.
|
|
"""
|
|
from core.cpi_engine import CommercialPowerEngine
|
|
|
|
engine = CommercialPowerEngine()
|
|
df = pd.DataFrame([
|
|
{"name": "Artist A", "hiffi_xp_score": 90.0, "hiffi_xp_published_rank": 1, "total_eas_cspc": 150_000_000},
|
|
{"name": "Artist B", "hiffi_xp_score": 80.0, "hiffi_xp_published_rank": 2, "total_eas_cspc": 50_000_000},
|
|
{"name": "Artist C", "hiffi_xp_score": 70.0, "hiffi_xp_published_rank": 3, "total_eas_cspc": None},
|
|
])
|
|
|
|
sim = engine.simulate_composite_ranking(df, commercial_weight=0.15, mode="global_redistribution")
|
|
sorted_sim = sim.sort_values(by="simulated_rank", ascending=True)
|
|
|
|
scores = sorted_sim["simulated_hiffi_score"].tolist()
|
|
assert scores == sorted(scores, reverse=True), "Simulated scores must decrease monotonically as simulated rank increases"
|
|
|
|
|
|
def test_production_outputs_strictly_immutable_under_simulation():
|
|
"""
|
|
CRITICAL REGRESSION TEST:
|
|
Verifies that running simulations across any commercial weights (0%, 10%, 20%)
|
|
and modes never alters or corrupts production baseline scores or published ranks.
|
|
"""
|
|
from core.cpi_engine import CommercialPowerEngine
|
|
|
|
engine = CommercialPowerEngine()
|
|
df_master = pd.read_csv("data/hiffi_multiplatform_500.csv")
|
|
|
|
original_production_scores = df_master["hiffi_xp_score"].copy()
|
|
original_published_ranks = df_master["hiffi_xp_published_rank"].copy()
|
|
|
|
for mode in ["matched_only", "global_redistribution"]:
|
|
for weight in [0.0, 0.05, 0.15, 0.25]:
|
|
sim_df = engine.simulate_ranking(df_master, commercial_weight=weight, mode=mode)
|
|
|
|
pd.testing.assert_series_equal(
|
|
df_master["hiffi_xp_score"],
|
|
original_production_scores,
|
|
check_names=True,
|
|
obj="Master production scores must not mutate"
|
|
)
|
|
pd.testing.assert_series_equal(
|
|
df_master["hiffi_xp_published_rank"],
|
|
original_published_ranks,
|
|
check_names=True,
|
|
obj="Master published ranks must not mutate"
|
|
)
|
|
|
|
if mode == "global_redistribution":
|
|
unmatched_mask = df_master["sales_data_status"] == "NOT_CATALOGED_YET"
|
|
pd.testing.assert_series_equal(
|
|
sim_df.loc[unmatched_mask, "simulated_hiffi_score"],
|
|
sim_df.loc[unmatched_mask, "hiffi_xp_score"],
|
|
check_names=False,
|
|
obj="Uncataloged artists must retain 100% baseline score"
|
|
) |