algorithm.py

import numpy as np
from typing import Dict, Tuple

# =========================
# CONFIGURATION CONSTANTS
# =========================
W1_PREDICTION = 0.50
W2_TALENT = 0.15
W3_PROGRESS = 0.15
W4_QUALITY = 0.15

MAX_DYNAMIC_BOOST = 0.05
SUPPRESSION_PENALTY = 0.60
FOLLOW_THRESHOLD = 1000

PS_FLAT_PROGRESS = 50.0
NETWORK_AVG_ESR = 0.10


class UpnisoAlgorithmPipeline:
    """
    Production-grade ranking pipeline.
    Stateless per request. Safe for concurrent execution.
    """

    def __init__(self, avg_esr: float = NETWORK_AVG_ESR, avg_ts: float = 50.0, avg_wtpu: float = 0.50):
        self.avg_esr = avg_esr
        self.avg_ts = avg_ts
        self.avg_wtpu = avg_wtpu

    # -------------------------
    # INTERNAL SAFE NORMALIZERS
    # -------------------------
    @staticmethod
    def _clamp(value, low=0.0, high=1.0):
        return float(np.clip(value, low, high))

    # -------------------------
    # STAGE 1 — ANALYZE
    # -------------------------
    def analyze(self, creator: Dict, content: Dict) -> Tuple[Dict, Dict]:
        creator = dict(creator)
        content = dict(content)

        creator["ts_90_days_ago"] = max(creator.get("ts_90_days_ago", self.avg_ts), 1.0)
        creator["ts_14_days_ago"] = max(creator.get("ts_14_days_ago", creator["ts_90_days_ago"]), 1.0)
        creator["avg_wtpu_month1"] = max(creator.get("avg_wtpu_month1", self.avg_wtpu), 0.01)
        creator["stdev_upload_days"] = creator.get("stdev_upload_days", 4.0)

        content["wtpu_current"] = self._clamp(content.get("wtpu_current", 0.0))
        content["nvr"] = self._clamp(content.get("nvr", 0.0))
        content["mis_score"] = 1.0 if content.get("mis_compliance") else 0.5
        content["esr_current"] = self._clamp(content.get("esr_current", self.avg_esr))
        content["prediction_score_p_hva"] = self._clamp(content.get("prediction_score_p_hva", 0.5))

        return creator, content

    # -------------------------
    # STAGE 2 — SCORE CREATOR
    # -------------------------
    def score_creator(self, creator: Dict) -> Dict[str, float]:
        wtpu = creator.get("wtpu_last_10_avg", 0.0)
        hva = creator.get("avg_hva_rate", 0.05)
        consistency = 1.0 / (creator["stdev_upload_days"] + 1.0)

        ts = 100 * (0.5 * wtpu + 0.3 * hva + 0.2 * consistency)
        ts = float(np.clip(ts, 0, 100))

        ts_delta = ((ts / creator["ts_90_days_ago"]) - 1.0) * 50
        wtpu_delta = ((creator.get("avg_wtpu_month3", creator["avg_wtpu_month1"]) / creator["avg_wtpu_month1"]) - 1.0) * 50
        ps = PS_FLAT_PROGRESS + ts_delta + wtpu_delta

        if (ts - creator["ts_14_days_ago"]) / creator["ts_14_days_ago"] >= 0.20:
            ps += 15

        ps = float(np.clip(ps, 0, 100))
        return {"TS": ts, "PS": ps}

    # -------------------------
    # STAGE 3 — PREDICT
    # -------------------------
    def predict_affinity(self, content: Dict) -> float:
        return content["prediction_score_p_hva"] * 100

    # -------------------------
    # STAGE 4 — QUALITY & BOOST
    # -------------------------
    def quality_and_boost(self, creator: Dict, content: Dict, merit: Dict) -> Tuple[float, float, bool]:
        qs = 100 * (
            0.6 * content["wtpu_current"] +
            0.3 * content["nvr"] +
            0.1 * content["mis_score"]
        )
        qs = float(np.clip(qs, 0, 100))

        dynamic_boost = 0.0
        if creator.get("follower_count", 0) < FOLLOW_THRESHOLD:
            merit_norm = (
                W2_TALENT * merit["TS"] +
                W3_PROGRESS * merit["PS"] +
                W4_QUALITY * qs
            ) / 100
            dynamic_boost = float(np.clip(merit_norm * 0.5, 0, MAX_DYNAMIC_BOOST))

        penalty = (content["esr_current"] / self.avg_esr) >= 5.0 if self.avg_esr > 0 else False
        return qs, dynamic_boost, penalty

    # -------------------------
    # STAGE 5 — FINAL RANKING
    # -------------------------
    def final_score(self, merit, pred, qs, boost, penalty) -> float:
        cs = (
            W1_PREDICTION * (pred / 100) +
            W2_TALENT * (merit["TS"] / 100) +
            W3_PROGRESS * (merit["PS"] / 100) +
            W4_QUALITY * (qs / 100)
        ) + boost

        if penalty:
            cs *= SUPPRESSION_PENALTY

        return float(np.clip(cs, 0, 1))

    # =========================
    # 🚀 PRODUCTION ENTRYPOINT
    # =========================
    def rank_feed(self, creators: Dict[str, Dict], contents: Dict[str, Dict]):
        ranked = []

        for cid, content in contents.items():
            creator = creators.get(content["creator_id"])
            if not creator:
                continue

            creator, content = self.analyze(creator, content)
            merit = self.score_creator(creator)
            pred = self.predict_affinity(content)
            qs, boost, penalty = self.quality_and_boost(creator, content, merit)
            cs = self.final_score(merit, pred, qs, boost, penalty)

            ranked.append({
                "content_id": cid,
                "creator_id": content["creator_id"],
                "score": cs,
                "TS": merit["TS"],
                "PS": merit["PS"],
                "QS": qs,
                "boost": boost,
                "penalty": penalty
            })

        ranked.sort(key=lambda x: x["score"], reverse=True)
        return ranked


def run_demo():
    return "Upniso algorithm loaded (production mode)"