Scoring Engine

Statistical scoring for ranking and comparing items across categories.

When to Use This Skill

Ranking content by performance (views, engagement)
Comparing items across categories with different baselines
Need freshness decay for time-sensitive content
Want confidence scores based on sample size

Core Concepts

Use percentiles over mean/std for skewed data. Apply freshness decay for older content. Calculate confidence based on sample size. Normalize across categories for fair comparison.

Implementation

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)))

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

Usage Examples

engine = ScoringEngine(redis_client)

# Build category stats
videos = await fetch_category_videos("gaming")
stats = await engine.build_category_stats("gaming", videos)

# Score individual items
for video in videos:
    hours_old = (datetime.now() - video["created_at"]).total_seconds() / 3600
    score, confidence = engine.score_item(
        views=video["view_count"],
        hours_old=hours_old,
        stats=stats,
    )
    print(f"{video['title']}: {score:.1f} (confidence: {confidence}%)")

Best Practices

Remove outliers before calculating statistics
Use percentiles over mean/std for skewed data
Apply freshness decay for time-sensitive content
Calculate confidence based on sample size
Cache category statistics (expensive to compute)

Common Mistakes

Using mean/std for highly skewed data
Not removing outliers (extreme values dominate)
Forgetting freshness decay (old content ranks too high)
Ignoring confidence (treating all scores equally)

Related Patterns

analytics-pipeline (data collection)
community-feed (applying scores to feeds)

scoring-engine