diff --git a/api.py b/api.py new file mode 100644 index 0000000..5ea875a --- /dev/null +++ b/api.py @@ -0,0 +1,143 @@ +""" +api.py +HIFFI Intelligence REST API Service. +Serves multi-platform composite rankings, verified sales telemetry, +and isolated commercial simulation models. +""" + +from enum import Enum +from pathlib import Path +from typing import Optional +from fastapi import FastAPI, HTTPException, Query +from pydantic import BaseModel +import pandas as pd + +from core.cpi_engine import CommercialPowerEngine + +# Module-level FastAPI declaration +app = FastAPI( + title="HIFFI Artist Intelligence API", + description="Multi-platform analytics, sales audit telemetry, and simulation engine.", + version="1.0.0", +) + +DATA_PATH = Path("data/hiffi_multiplatform_500.csv") +engine = CommercialPowerEngine() + + +class ScopeChoice(str, Enum): + all = "all" + verified = "verified" + + +def get_dataset() -> pd.DataFrame: + if not DATA_PATH.exists(): + raise HTTPException( + status_code=500, + detail="Master dataset data/hiffi_multiplatform_500.csv not found.", + ) + return pd.read_csv(DATA_PATH) + + +class SimulationRequest(BaseModel): + commercial_weight: float = 0.15 + mode: str = "global_redistribution" # 'matched_only' or 'global_redistribution' + + +@app.get("/health", tags=["System"]) +def health_check(): + """Returns engine heartbeat and dataset count.""" + try: + df = get_dataset() + return { + "status": "healthy", + "total_tracked_artists": len(df), + "engine": "CommercialPowerEngine v1.0", + } + except Exception as e: + return {"status": "unhealthy", "error": str(e)} + + +@app.get("/telemetry/coverage", tags=["Governance"]) +def telemetry_coverage(): + """Returns data provenance tier distribution.""" + df = get_dataset() + cspc_count = int((df.get("sales_data_status") == "VERIFIED_CSPC").sum()) + riaa_count = int((df.get("sales_data_status") == "VERIFIED_RIAA").sum()) + sea_count = int((df.get("sales_data_status") == "ESTIMATED_SEA").sum()) + unranked_count = len(df) - cspc_count - riaa_count - sea_count + + return { + "total_monitored": len(df), + "verified_cspc": cspc_count, + "verified_riaa": riaa_count, + "estimated_sea": sea_count, + "unranked": unranked_count, + "active_coverage_percentage": round( + ((cspc_count + riaa_count + sea_count) / len(df)) * 100.0, 2 + ), + } + + +@app.get("/artists", tags=["Rankings"]) +def list_artists( + scope: ScopeChoice = ScopeChoice.all, + limit: int = Query(50, ge=1, le=500), + vertical: Optional[str] = None, +): + """Returns ranked artist list respecting data governance scopes.""" + df = get_dataset() + + if scope == ScopeChoice.verified: + df = df[df["sales_data_status"].isin(["VERIFIED_CSPC", "VERIFIED_RIAA"])].copy() + + if vertical and "subgenre" in df.columns: + df = df[df["subgenre"].str.lower() == vertical.lower()].copy() + + sort_col = "hiffi_xp_published_rank" if "hiffi_xp_published_rank" in df.columns else "hiffi_rank" + df = df.sort_values(by=sort_col, ascending=True).head(limit) + + return { + "count": len(df), + "scope": scope.value, + "records": df.to_dict(orient="records"), + } + + +@app.post("/simulate", tags=["Simulation"]) +def run_simulation(req: SimulationRequest): + """Calculates rank shifts in-memory without mutating published rankings.""" + df = get_dataset() + + if req.mode == "matched_only": + df = df[df["sales_data_status"] == "VERIFIED_CSPC"].copy() + + simulated_df = engine.simulate_ranking( + df, commercial_weight=req.commercial_weight, mode=req.mode + ) + metrics = engine.compute_simulation_metrics(simulated_df) + + output_cols = [ + "name", + "hiffi_xp_published_rank", + "simulated_rank", + "simulation_rank_delta", + "hiffi_xp_score", + "simulated_hiffi_score", + "sales_data_status", + ] + render_cols = [c for c in output_cols if c in simulated_df.columns] + + return { + "parameters": { + "commercial_weight": req.commercial_weight, + "mode": req.mode, + }, + "metrics": metrics, + "rankings": simulated_df[render_cols].to_dict(orient="records"), + } + + +if __name__ == "__main__": + import uvicorn + uvicorn.run("api:app", host="0.0.0.0", port=8000, reload=True) \ No newline at end of file diff --git a/app.py b/app.py index 051c46c..9fd4a57 100644 --- a/app.py +++ b/app.py @@ -2,16 +2,19 @@ app.py HIFFI Artist Intelligence - Multi-Platform Executive Dashboard Clean dark-themed design with multi-platform evidence tiers, Spotify & Instagram telemetry, -ChartMasters record sales (CSPC/EAS), breakout velocity index, and deep-dive scouting dossier. +ChartMasters record sales (CSPC/EAS), breakout velocity index, deep-dive scouting dossier, +and experimental Commercial Power Index (CPI) simulation engine with sensitivity analysis. """ -import io from pathlib import Path import pandas as pd import plotly.express as px import plotly.graph_objects as go import streamlit as st +from core.cpi_engine import CommercialPowerEngine +from exporters.excel_workbook import generate_executive_sales_workbook + st.set_page_config( page_title="HIFFI | Multi-Platform Intelligence", page_icon="⚑", @@ -22,7 +25,7 @@ st.set_page_config( DATA_PATH = Path("data/hiffi_multiplatform_500.csv") FALLBACK_DATA_PATH = Path("enriched_artists.csv") -# Global CSS Polish +# Global CSS Styling st.markdown( """