Create algorithm.py

algorithm.py

@@ -0,0 +1,281 @@
+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"