File size: 5,461 Bytes
6760346 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 | 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)"
|