48 lines
1.6 KiB
Python
48 lines
1.6 KiB
Python
"""
|
|
run_pipeline.py
|
|
Single entry point executing the end-to-end multi-platform ranking pipeline:
|
|
1. Validates strict schema and uncorrupted nulls
|
|
2. Calculates HIFFI-XP composite scores & dynamic weight redistribution
|
|
3. Runs anomaly detection guardrails
|
|
4. Applies fail-closed publication eligibility & freshness gates
|
|
5. Regenerates provenance audit report
|
|
"""
|
|
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import pandas as pd
|
|
from core.ranking_engine import HIFFIRankingEngine
|
|
|
|
|
|
def main():
|
|
dataset_file = Path("data/hiffi_multiplatform_500.csv")
|
|
if not dataset_file.exists():
|
|
print(f"Error: Dataset {dataset_file} not found.", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
print("[1/5] Loading cohort dataset...")
|
|
df = pd.read_csv(dataset_file)
|
|
assert len(df) == 500, f"Expected 500 records, got {len(df)}"
|
|
|
|
print("[2/5] Calculating HIFFI-XP composite scores & effective weights...")
|
|
engine = HIFFIRankingEngine()
|
|
scored = engine.calculate_hiffi_xp(df)
|
|
|
|
print("[3/5] Running anomaly detection guardrails...")
|
|
flagged = engine.detect_metric_anomalies(scored)
|
|
|
|
print("[4/5] Applying fail-closed publication eligibility & freshness gates...")
|
|
gated, shortfall_audit = engine.apply_publication_eligibility_gates(flagged, return_metadata=True)
|
|
|
|
gated.to_csv(dataset_file, index=False)
|
|
print(" -> Saved refreshed dataset with all governance columns.")
|
|
|
|
print("[5/5] Generating provenance audit report...")
|
|
subprocess.run([sys.executable, "reports/provenance_summary.py"], check=True)
|
|
|
|
print("\n✓ Pipeline execution successful. All 500 records ranked and audited.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |