| import numpy as np |
| from typing import Dict, Tuple |
|
|
| |
| |
| |
| 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 |
|
|
| |
| |
| |
| @staticmethod |
| def _clamp(value, low=0.0, high=1.0): |
| return float(np.clip(value, low, high)) |
|
|
| |
| |
| |
| 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 |
|
|
| |
| |
| |
| 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} |
|
|
| |
| |
| |
| def predict_affinity(self, content: Dict) -> float: |
| return content["prediction_score_p_hva"] * 100 |
|
|
| |
| |
| |
| 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 |
|
|
| |
| |
| |
| 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)) |
|
|
| |
| |
| |
| 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)" |
|
|
