Python
```python
from dataclasses import dataclass
from typing import List, Dict, Optional, Tuple
import statistics
import math
@dataclass
class CategoryStats:
"""Statistical summary for a category."""
category_key: str
sample_count: int
view_mean: float
view_std: float
view_p25: float
view_p50: float
view_p75: float
view_p90: float
view_min: float
view_max: float
outliers_removed: int = 0
@classmethod
def from_videos(cls, category_key: str, videos: List[Dict], remove_outliers: bool = True) -> "CategoryStats":
if not videos:
return cls._empty(category_key)
views = [v.get("view_count", 0) for v in videos if v.get("view_count", 0) > 0]
if not views:
return cls._empty(category_key)
outliers_removed = 0
if remove_outliers and len(views) > 10:
views, outliers_removed = cls._remove_outliers(views)
sorted_views = sorted(views)
n = len(sorted_views)
return cls(
category_key=category_key,
sample_count=len(views),
view_mean=statistics.mean(views),
view_std=statistics.stdev(views) if len(views) > 1 else 0,
view_p25=sorted_views[int(n * 0.25)],
view_p50=sorted_views[int(n * 0.50)],
view_p75=sorted_views[int(n * 0.75)],
view_p90=sorted_views[int(n * 0.90)],
view_min=min(views),
view_max=max(views),
outliers_removed=outliers_removed,
)
@staticmethod
def _remove_outliers(values: List[float]) -> Tuple[List[float], int]:
"""Remove outliers using IQR method."""
sorted_vals = sorted(values)
n = len(sorted_vals)
q1, q3 = sorted_vals[int(n 0.25)], sorted_vals[int(n 0.75)]
iqr = q3 - q1
lower, upper = q1 - 1.5 iqr, q3 + 1.5 iqr
filtered = [v for v in values if lower <= v <= upper]
return filtered, len(values) - len(filtered)
@classmethod
def _empty(cls, category_key: str) -> "CategoryStats":
return cls(category_key=category_key, sample_count=0, view_mean=0, view_std=1,
view_p25=0, view_p50=0, view_p75=0, view_p90=0, view_min=0, view_max=0)
@dataclass
class PercentileThresholds:
p25: float
p50: float
p75: float
p90: float
def calculate_percentile_score(value: float, thresholds: PercentileThresholds) -> float:
"""Map value to 0-100 score based on percentile thresholds."""
if value <= 0:
return 0.0
if value <= thresholds.p25:
return 25 * (value / thresholds.p25) if thresholds.p25 > 0 else 0
elif value <= thresholds.p50:
return 25 + 25 * ((value - thresholds.p25) / (thresholds.p50 - thresholds.p25))
elif value <= thresholds.p75:
return 50 + 25 * ((value - thresholds.p50) / (thresholds.p75 - thresholds.p50))
elif value <= thresholds.p90:
return 75 + 15 * ((value - thresholds.p75) / (thresholds.p90 - thresholds.p75))
else:
excess = min(value - thresholds.p90, thresholds.p90 * 2)
return 90 + 10 (excess / (thresholds.p90 2))
def freshness_decay(hours_old: float, half_life: float = 24.0) -> float:
"""Exponential decay: factor = 0.5^(age/half_life)"""
if hours_old <= 0:
return 1.0
return math.pow(0.5, hours_old / half_life)
def recency_boost(hours_old: float, boost_window: float = 6.0) -> float:
"""Extra boost for very fresh content (1.0-1.5)."""
if hours_old >= boost_window:
return 1.0
return 1.5 - (0.5 * hours_old / boost_window)
def calculate_confidence(sample_size: int, score_variance: float = 0.0) -> int:
"""Confidence score (0-100) based on sample size and variance."""
if sample_size <= 0:
return 0
sample_confidence = min(100, 25 * math.log10(sample_size + 1))
variance_penalty = min(30, score_variance * 10)
return max(0, min(100, int(sample_confidence - variance_penalty)))
```
```python
def combine_scores(
scores: Dict[str, float],
weights: Dict[str, float],
) -> Tuple[float, int]:
"""Combine multiple scores with weights."""
if not scores:
return 0.0, 0
total_weight = 0.0
weighted_sum = 0.0
for name, score in scores.items():
weight = weights.get(name, 1.0)
weighted_sum += score * weight
total_weight += weight
if total_weight == 0:
return 0.0, 0
combined = weighted_sum / total_weight
confidence = calculate_confidence(len(scores) * 10)
return combined, confidence
class ScoringEngine:
"""Enterprise-grade scoring engine."""
def __init__(self, redis_client):
self.redis = redis_client
self._stats_cache: Dict[str, CategoryStats] = {}
async def build_category_stats(self, category_key: str, videos: List[Dict]) -> CategoryStats:
stats = CategoryStats.from_videos(category_key, videos, remove_outliers=True)
self._stats_cache[category_key] = stats
return stats
def score_item(
self,
views: int,
hours_old: float,
stats: CategoryStats,
) -> Tuple[float, int]:
thresholds = PercentileThresholds(
p25=stats.view_p25, p50=stats.view_p50,
p75=stats.view_p75, p90=stats.view_p90,
)
view_score = calculate_percentile_score(views, thresholds)
freshness = freshness_decay(hours_old)
recency = recency_boost(hours_old)
# Velocity score
velocity = views / max(hours_old, 1.0)
velocity_thresholds = PercentileThresholds(
p25=stats.view_p25/24, p50=stats.view_p50/24,
p75=stats.view_p75/24, p90=stats.view_p90/24,
)
velocity_score = calculate_percentile_score(velocity, velocity_thresholds)
# Combine
scores = {"views": view_score, "velocity": velocity_score}
weights = {"views": 0.6, "velocity": 0.4}
combined, confidence = combine_scores(scores, weights)
final_score = min(100, combined freshness recency)
return final_score, confidence
```
|