hiffi-artist-intelligence/core/cpi_engine.py

"""
core/cpi_engine.py
Commercial Power Index (CPI) & Multi-Scenario Simulation Engine.
"""

from typing import Dict, List, Literal
import numpy as np
import pandas as pd


class CommercialPowerEngine:
    GLOBAL_MIN_LOG_EAS: float = 5.000000  # log10(100,000 EAS)
    GLOBAL_MAX_LOG_EAS: float = 8.397940  # log10(250,000,000 EAS)

    def __init__(
        self,
        min_log_eas: float = GLOBAL_MIN_LOG_EAS,
        max_log_eas: float = GLOBAL_MAX_LOG_EAS,
    ):
        self.min_log_eas = min_log_eas
        self.max_log_eas = max_log_eas

    def calculate_cpi(self, df: pd.DataFrame, eas_col: str = "total_eas_cspc") -> pd.Series:
        if eas_col not in df.columns:
            return pd.Series(index=df.index, dtype=float)

        eas_series = pd.to_numeric(df[eas_col], errors="coerce")
        valid_mask = eas_series.notna() & (eas_series > 0)

        cpi_series = pd.Series(index=df.index, dtype=float)
        if not valid_mask.any():
            return cpi_series

        log_eas = np.log10(np.maximum(eas_series[valid_mask], 1.0))
        normalized = ((log_eas - self.min_log_eas) / (self.max_log_eas - self.min_log_eas)) * 100.0
        cpi_series[valid_mask] = np.clip(normalized, 0.0, 100.0)

        return cpi_series.round(2)

    def simulate_ranking(
        self,
        df: pd.DataFrame,
        commercial_weight: float = 0.15,
        mode: Literal["matched_only", "global_redistribution"] = "matched_only",
        score_col: str = "hiffi_xp_score",
        cpi_col: str = "commercial_power_index",
        base_rank_col: str = "hiffi_xp_published_rank",
    ) -> pd.DataFrame:
        sim_df = df.copy()

        if cpi_col not in sim_df.columns:
            sim_df[cpi_col] = self.calculate_cpi(sim_df)

        if mode == "matched_only":
            sim_df = sim_df[sim_df[cpi_col].notna()].copy()
            if sim_df.empty:
                return sim_df
            sim_df["baseline_matched_rank"] = (
                pd.to_numeric(sim_df[score_col], errors="coerce")
                .rank(ascending=False, method="min")
                .astype(int)
            )
            comparison_rank_col = "baseline_matched_rank"
        else:
            comparison_rank_col = base_rank_col if base_rank_col in sim_df.columns else "hiffi_xp_rank"

        base_scores = pd.to_numeric(sim_df[score_col], errors="coerce").fillna(0.0)
        cpi_scores = pd.to_numeric(sim_df[cpi_col], errors="coerce")

        has_cpi = cpi_scores.notna()
        w_sales = max(0.0, min(commercial_weight, 0.50))
        w_base = 1.0 - w_sales

        sim_scores = pd.Series(index=sim_df.index, dtype=float)
        sim_scores[has_cpi] = (w_base * base_scores[has_cpi]) + (w_sales * cpi_scores[has_cpi])
        sim_scores[~has_cpi] = base_scores[~has_cpi]

        sim_df["simulated_hiffi_score"] = sim_scores.round(2)
        sim_df["simulated_rank"] = (
            sim_df["simulated_hiffi_score"].rank(ascending=False, method="min").astype(int)
        )

        if comparison_rank_col in sim_df.columns:
            sim_df["simulation_rank_delta"] = (
                pd.to_numeric(sim_df[comparison_rank_col], errors="coerce") - sim_df["simulated_rank"]
            )

        return sim_df

    # Backward-compatible alias
    simulate_composite_ranking = simulate_ranking

    def compute_simulation_metrics(self, sim_df: pd.DataFrame) -> Dict[str, any]:
        delta_series = sim_df.get("simulation_rank_delta", pd.Series(dtype=float)).dropna()
        if delta_series.empty:
            return {
                "top_gainer_name": "None", "top_gainer_delta": 0,
                "top_loser_name": "None", "top_loser_delta": 0,
                "mean_abs_delta": 0.0, "matched_coverage": 0, "total_cohort": len(sim_df)
            }

        max_delta = delta_series.max()
        min_delta = delta_series.min()

        top_gainer_rows = sim_df[sim_df["simulation_rank_delta"] == max_delta]
        top_loser_rows = sim_df[sim_df["simulation_rank_delta"] == min_delta]

        top_gainer_name = top_gainer_rows["name"].iloc[0] if not top_gainer_rows.empty else "None"
        top_loser_name = top_loser_rows["name"].iloc[0] if not top_loser_rows.empty else "None"

        matched_count = int(sim_df["commercial_power_index"].notna().sum()) if "commercial_power_index" in sim_df.columns else 0

        return {
            "top_gainer_name": top_gainer_name,
            "top_gainer_delta": int(max_delta),
            "top_loser_name": top_loser_name,
            "top_loser_delta": int(min_delta),
            "mean_abs_delta": round(float(delta_series.abs().mean()), 2),
            "matched_coverage": matched_count,
            "total_cohort": len(sim_df),
        }

    def run_sensitivity_matrix(
        self,
        df: pd.DataFrame,
        weights: List[float] = [0.0, 0.05, 0.10, 0.15, 0.20],
        mode: Literal["matched_only", "global_redistribution"] = "matched_only",
    ) -> pd.DataFrame:
        base_sim = self.simulate_ranking(df, commercial_weight=0.0, mode=mode)
        matrix_df = pd.DataFrame({"name": base_sim["name"]})

        for w in weights:
            run_df = self.simulate_ranking(df, commercial_weight=w, mode=mode)
            w_label = f"rank_{int(w*100)}pct"
            name_to_rank = dict(zip(run_df["name"], run_df["simulated_rank"]))
            matrix_df[w_label] = matrix_df["name"].map(name_to_rank)

        rank_cols = [c for c in matrix_df.columns if c.startswith("rank_")]
        matrix_df["rank_volatility_spread"] = matrix_df[rank_cols].max(axis=1) - matrix_df[rank_cols].min(axis=1)
        return matrix_df.sort_values(by="rank_volatility_spread", ascending=False)