import numpy as np import random from typing import List, Dict, Union, Tuple # --- Configuration Constants (Normalized to 0-100 Scale) --- # Weighting for the Final Content Score (CS) W1_PREDICTION = 0.50 W2_TALENT = 0.15 W3_PROGRESS = 0.15 W4_QUALITY = 0.15 # Dynamic Allocation MAX_DYNAMIC_BOOST = 0.05 # Max 5% of CS allowed via dynamic boosts/penalties SUPPRESSION_PENALTY = 0.60 # Penalty multiplier for high ESR content FOLLOW_THRESHOLD = 1000 # Max followers for Small-Creator Advantage # --- Score Scaling and Clamping Constants --- PS_FLAT_PROGRESS = 50.0 # PS score when a creator has 0% improvement NETWORK_AVG_ESR = 0.10 # Network average Early Skip Rate (0-1) class UpnisoAlgorithmPipeline: """ Upniso Merit-Based Recommendation Algorithm Pipeline (V2.0). Encapsulates the 6-stage architecture: Analyze, Score, Predict, Test, Distribute, and Learn. All core scores are normalized to a 0-100 scale. """ def __init__(self, avg_esr: float = NETWORK_AVG_ESR, avg_ts: float = 50.0, avg_wtpu: float = 0.50): """Initializes the pipeline with network baseline averages for comparison.""" self.avg_esr = avg_esr self.avg_ts = avg_ts self.avg_wtpu = avg_wtpu self.creators = {} self.content_queue = {} # --- Stage 1: Analyze (Data Ingestion and Cleanup) --- def analyze(self, creator_id: str, creator_data: dict, content_id: str, content_data: dict): """ Ingests, sanitizes, and verifies raw data from the database/API. Adds fallbacks for missing historical data. """ # Guard against division by zero and provide sensible fallbacks (Task B.3) creator_data['ts_90_days_ago'] = max(creator_data.get('ts_90_days_ago', self.avg_ts), 1.0) creator_data['avg_wtpu_month1'] = max(creator_data.get('avg_wtpu_month1', self.avg_wtpu), 0.01) creator_data['ts_14_days_ago'] = creator_data.get('ts_14_days_ago', creator_data['ts_90_days_ago']) creator_data['stdev_upload_days'] = creator_data.get('stdev_upload_days', 4.0) content_data['wtpu_current'] = np.clip(content_data.get('wtpu_current', 0.0), 0.0, 1.0).item() content_data['nvr'] = np.clip(content_data.get('nvr', 0.0), 0.0, 1.0).item() # MIS is 1.0 for compliance (True), 0.5 otherwise (False/Missing) content_data['mis_score'] = 1.0 if content_data.get('mis_compliance', False) else 0.5 content_data['esr_current'] = np.clip(content_data.get('esr_current', 0.5), 0.0, 1.0).item() content_data['prediction_score_p_hva'] = content_data.get('prediction_score_p_hva', 0.5) self.creators[creator_id] = creator_data self.content_queue[content_id] = content_data self.content_queue[content_id]['creator_id'] = creator_id # --- Stage 2: Score (Creator Profile Merit Calculation) --- def score(self, creator_id: str) -> Dict[str, float]: """ Calculates the creator's persistent Talent Score (TS) and Progress Score (PS). Both scores are normalized to 0-100. """ creator_data = self.creators.get(creator_id) if not creator_data: return {'TS': self.avg_ts, 'PS': PS_FLAT_PROGRESS} # 1. Talent Score (TS) Calculation (0-100, Normalized) wtpu_norm = creator_data['wtpu_last_10_avg'] hva_norm = creator_data.get('avg_hva_rate', 0.05) # Consistency Factor: Closer to 1.0 is better (lower stdev) consistency = 1.0 / (creator_data['stdev_upload_days'] + 1.0) # TS = 100 * (0.5 * WTPU + 0.3 * HVA + 0.2 * Consistency) ts_merit = 100 * (0.5 * wtpu_norm + 0.3 * hva_norm + 0.2 * consistency) ts_today = np.clip(ts_merit, 0, 100).item() self.creators[creator_id]['ts_today'] = ts_today # 2. Progress Score (PS) Calculation (0-100, 50=Flat) ts_90_days_ago = creator_data['ts_90_days_ago'] wtpu_m1 = creator_data['avg_wtpu_month1'] wtpu_m3 = creator_data.get('avg_wtpu_month3', wtpu_m1) # PS_Raw = 50 + [Delta TS * 50] + [Delta WTPU * 50] ts_delta_contribution = ((ts_today / ts_90_days_ago) - 1.0) * 50 wtpu_delta_contribution = ((wtpu_m3 / wtpu_m1) - 1.0) * 50 ps_raw = PS_FLAT_PROGRESS + ts_delta_contribution + wtpu_delta_contribution # Forgiveness Boost Check (Task A.10: If TS improved >= 20% in 14 days) ts_14_days_ago = creator_data['ts_14_days_ago'] ts_14_day_improvement = (ts_today - ts_14_days_ago) / ts_14_days_ago forgiveness_boost = 0 if ts_14_day_improvement >= 0.20: forgiveness_boost = 15 # +15 score for demonstrated momentum # Clamping PS score (0-100) ps_score = np.clip(ps_raw + forgiveness_boost, 0, 100).item() return {'TS': ts_today, 'PS': ps_score} # --- Stage 3: Predict (Affinity Model Simulation) --- def predict(self, content_id: str) -> float: """ Simulates the Prediction Score (P(HVA)) based on user affinity. Returns score (0-100). """ content_data = self.content_queue.get(content_id) if not content_data: return PS_FLAT_PROGRESS prediction_raw = content_data['prediction_score_p_hva'] * 100 return np.clip(prediction_raw, 0, 100).item() # --- Stage 4: Test (Fair Exposure Engine - FEE) --- def test(self, content_id: str, merit_scores: Dict[str, float]) -> Tuple[float, float]: """ Calculates the Quality Score (QS) and the dynamic boost/penalty flag based on initial T0/T1 micro-test results. Returns QS (0-100) and Dynamic Factor (0 to 5%). """ content_data = self.content_queue.get(content_id) creator_data = self.creators.get(content_data['creator_id']) if not content_data or not creator_data: return PS_FLAT_PROGRESS, 0.0 # 1. Quality Score (QS) Calculation (0-100, Normalized) wtpu = content_data['wtpu_current'] nvr = content_data['nvr'] mis = content_data['mis_score'] # QS = 100 * (0.6 * WTPU + 0.3 * NVR + 0.1 * MIS) qs_score = 100 * (0.6 * wtpu + 0.3 * nvr + 0.1 * mis) qs_score = np.clip(qs_score, 0, 100).item() # 2. Dynamic Boost Calculation (5% Max) dynamic_factor = 0.0 # Small-Creator Advantage (1.5x Multiplier to Merit in Test Phase) if creator_data['follower_count'] < FOLLOW_THRESHOLD: # Merit component calculation (normalized to 0-1) merit_value_norm = (W2_TALENT * merit_scores['TS'] + W3_PROGRESS * merit_scores['PS'] + W4_QUALITY * qs_score) / 100 # 1.5x multiplier means adding 50% of the calculated merit value (45% weight) boost_value = merit_value_norm * 0.5 # Clamp the boost to the allocated 5% maximum dynamic weight dynamic_factor += np.clip(boost_value, 0, MAX_DYNAMIC_BOOST).item() # 3. Negative Signal Suppression Flag (ESR >= 5x Avg) esr_ratio = content_data['esr_current'] / self.avg_esr if self.avg_esr > 0 else 1.0 penalty_flag = False if esr_ratio >= 5.0: penalty_flag = True self.content_queue[content_id]['penalty_flag'] = penalty_flag self.content_queue[content_id]['QS'] = qs_score return qs_score, dynamic_factor # --- Stage 5: Distribute (Final Content Score Ranking) --- def distribute(self, content_id: str, merit_scores: Dict[str, float], pred_score: float, qs_score: float, dynamic_factor: float) -> float: """ Calculates the Final Content Score (CS, 0-1.0) used for feed ranking. """ # Convert 0-100 scores to 0-1 for weighting ts_norm = merit_scores['TS'] / 100 ps_norm = merit_scores['PS'] / 100 qs_norm = qs_score / 100 pred_norm = pred_score / 100 # Baseline CS (95% weight) = (50% Pred) + (45% Merit) cs_baseline = (W1_PREDICTION * pred_norm) + \ (W2_TALENT * ts_norm) + \ (W3_PROGRESS * ps_norm) + \ (W4_QUALITY * qs_norm) # Final Content Score (Max 1.0) final_cs = cs_baseline + dynamic_factor # Apply Suppression Penalty if flagged if self.content_queue.get(content_id, {}).get('penalty_flag', False): final_cs *= SUPPRESSION_PENALTY return np.clip(final_cs, 0, 1).item() # --- Stage 6: Learn (Feedback Loop Simulation) --- def learn(self, creator_id: str, content_id: str, final_cs: float): """ Simulates updating the creator's profile data based on content performance (Offline Job). """ # In a real system, this would run daily to update ts_90_days_ago, ts_14_days_ago, etc. # For simulation, we just record results. self.content_queue[content_id]['final_cs'] = final_cs self.content_queue[content_id]['PS'] = self.content_queue[content_id].get('PS', 0) self.content_queue[content_id]['TS'] = self.content_queue[content_id].get('TS', 0) # --- Simulation Mode (Task B.6 / D) --- def run_simulation(self, creators_data: Dict[str, Dict], contents_data: Dict[str, Dict]): """Runs a competition simulation for all provided content and returns the ranked list.""" results = [] print("="*80) print("UPNISO ALGORITHM SIMULATION (V2.0)") print("="*80) for content_id, content_input in contents_data.items(): creator_id = content_input['creator_id'] creator_input = creators_data[creator_id] # --- Pipeline Execution --- self.analyze(creator_id, creator_input, content_id, content_input) merit = self.score(creator_id) pred = self.predict(content_id) qs, dynamic_boost = self.test(content_id, merit) # Store scores for reporting self.content_queue[content_id]['QS'] = qs self.content_queue[content_id]['PS'] = merit['PS'] self.content_queue[content_id]['TS'] = merit['TS'] self.content_queue[content_id]['Pred'] = pred self.content_queue[content_id]['dynamic_boost'] = dynamic_boost final_cs = self.distribute(content_id, merit, pred, qs, dynamic_boost) self.learn(creator_id, content_id, final_cs) # --- Reporting --- penalty = "APPLIED" if self.content_queue[content_id].get('penalty_flag') else "None" results.append({ 'content_id': content_id, 'creator_id': creator_id, 'CS': final_cs, 'TS': merit['TS'], 'PS': merit['PS'], 'QS': qs, 'Pred': pred, 'Dynamic_Boost': dynamic_boost, 'Penalty': penalty }) # Final Ranking results.sort(key=lambda x: x['CS'], reverse=True) print("\n" + "="*80) print("FINAL RANKED ORDER (Rank #1 is highest CS)") print("="*80) for i, r in enumerate(results): print(f"RANK {i+1:2}: CS={r['CS']:.4f} | TS={r['TS']:.1f} | PS={r['PS']:.1f} | QS={r['QS']:.1f} | Pred={r['Pred']:.1f} | ID={r['content_id']} ({r['creator_id']})") return results # --- Simulation Data and Execution (Task D) --- # 1. Creator Data (Pre-calculated and Denormalized inputs) CREATORS_DATA = { 'C1': {'follower_count': 500, 'ts_90_days_ago': 20.0, 'ts_14_days_ago': 28.0, 'wtpu_last_10_avg': 0.85, 'avg_hva_rate': 0.15, 'stdev_upload_days': 0.5, 'avg_wtpu_month1': 0.2, 'avg_wtpu_month3': 0.4}, # Small/High-Skill: TS up 40% in 14 days 'C2': {'follower_count': 15000, 'ts_90_days_ago': 70.0, 'ts_14_days_ago': 70.0, 'wtpu_last_10_avg': 0.65, 'avg_hva_rate': 0.05, 'stdev_upload_days': 4.0, 'avg_wtpu_month1': 0.6, 'avg_wtpu_month3': 0.6}, # Luck-Focused: Stagnant, inconsistent 'C3': {'follower_count': 5000, 'ts_90_days_ago': 50.0, 'ts_14_days_ago': 45.0, 'wtpu_last_10_avg': 0.50, 'avg_hva_rate': 0.08, 'stdev_upload_days': 2.5, 'avg_wtpu_month1': 0.5, 'avg_wtpu_month3': 0.4}, # Mid-Stagnant: Regressing 'C4': {'follower_count': 100000, 'ts_90_days_ago': 85.0, 'ts_14_days_ago': 88.0, 'wtpu_last_10_avg': 0.80, 'avg_hva_rate': 0.20, 'stdev_upload_days': 1.0, 'avg_wtpu_month1': 0.7, 'avg_wtpu_month3': 0.77}, # Large/Consistent: Elite skill, steady improvement 'C5': {'follower_count': 50, 'ts_90_days_ago': 10.0, 'ts_14_days_ago': 11.0, 'wtpu_last_10_avg': 0.70, 'avg_hva_rate': 0.10, 'stdev_upload_days': 3.0, 'avg_wtpu_month1': 0.1, 'avg_wtpu_month3': 0.3}, # Micro/Improving: Huge WTPU progress, TS up 25% in 14 days } # 2. Content Data (Micro-Test results and Model Predictions) CONTENTS_DATA = { 'CN1': {'creator_id': 'C1', 'wtpu_current': 0.95, 'nvr': 0.20, 'mis_compliance': True, 'esr_current': 0.08, 'prediction_score_p_hva': 0.30}, # Niche, High Quality 'CN2': {'creator_id': 'C2', 'wtpu_current': 0.15, 'nvr': 0.05, 'mis_compliance': False, 'esr_current': 0.55, 'prediction_score_p_hva': 0.90}, # Clickbait, Low Quality (ESR 5.5x Avg) 'CN3': {'creator_id': 'C3', 'wtpu_current': 0.50, 'nvr': 0.10, 'mis_compliance': True, 'esr_current': 0.10, 'prediction_score_p_hva': 0.50}, # Average 'CN4': {'creator_id': 'C4', 'wtpu_current': 0.80, 'nvr': 0.30, 'mis_compliance': True, 'esr_current': 0.12, 'prediction_score_p_hva': 0.85}, # Elite, High Affinity 'CN5': {'creator_id': 'C5', 'wtpu_current': 0.75, 'nvr': 0.15, 'mis_compliance': True, 'esr_current': 0.05, 'prediction_score_p_hva': 0.20}, # High Progress, Unknown Affinity } if __name__ == '__main__': pipeline = UpnisoAlgorithmPipeline() pipeline.run_simulation(CREATORS_DATA, CONTENTS_DATA) def run_demo(): return "Algorithm connected successfully"