Recommendation System

Recommendation System

Production-ready architecture for scalable recommendation systems with feature stores, multi-tier caching, A/B testing, and comprehensive monitoring.

When to Use This Skill

Load this skill when:

• Building Recommendation APIs: Serving personalized recommendations at scale
• Implementing Caching: Multi-tier caching for sub-millisecond latency
• Running A/B Tests: Experimenting with recommendation algorithms
• Monitoring Quality: Tracking CTR, conversion, diversity, coverage
• Optimizing Performance: Reducing latency, increasing throughput
• Feature Engineering: Managing user/item features with feature stores

Quick Start: Recommendation API in 5 Steps

# 1. Install dependencies
pip install fastapi==0.109.0 redis==5.0.0 prometheus-client==0.19.0

# 2. Start Redis (for caching and feature store)
docker run -d -p 6379:6379 redis:alpine

# 3. Create recommendation service: app.py
cat > app.py << 'EOF'
from fastapi import FastAPI
from pydantic import BaseModel
from typing import List
import redis
import json

app = FastAPI()
cache = redis.Redis(host='localhost', port=6379, decode_responses=True)

class RecommendationResponse(BaseModel):
    user_id: str
    items: List[str]
    cached: bool

@app.post("/recommendations", response_model=RecommendationResponse)
async def get_recommendations(user_id: str, n: int = 10):
    # Check cache
    cache_key = f"recs:{user_id}:{n}"
    cached = cache.get(cache_key)

    if cached:
        return RecommendationResponse(
            user_id=user_id,
            items=json.loads(cached),
            cached=True
        )

    # Generate recommendations (simplified)
    items = [f"item_{i}" for i in range(n)]

    # Cache for 5 minutes
    cache.setex(cache_key, 300, json.dumps(items))

    return RecommendationResponse(
        user_id=user_id,
        items=items,
        cached=False
    )

@app.get("/health")
async def health():
    return {"status": "healthy"}
EOF

# 4. Run API
uvicorn app:app --host 0.0.0.0 --port 8000

# 5. Test
curl -X POST "http://localhost:8000/recommendations?user_id=user_123&n=10"

Result: Working recommendation API with caching in under 5 minutes.

System Architecture

┌─────────────┐     ┌─────────────┐     ┌─────────────┐
│ User Events │────▶│ Feature     │────▶│ Model       │
│ (clicks,    │     │ Store       │     │ Serving     │
│  purchases) │     │ (Redis)     │     │             │
└─────────────┘     └─────────────┘     └─────────────┘
                           │                    │
                           ▼                    ▼
                    ┌─────────────┐     ┌─────────────┐
                    │ Training    │     │ API         │
                    │ Pipeline    │     │ (FastAPI)   │
                    └─────────────┘     └─────────────┘
                                               │
                                               ▼
                                        ┌─────────────┐
                                        │ Monitoring  │
                                        │ (Prometheus)│
                                        └─────────────┘

Core Components

1. Feature Store

Centralized storage for user and item features:

import redis
import json

class FeatureStore:
    """Fast feature access with Redis caching."""

    def __init__(self, redis_client):
        self.redis = redis_client
        self.ttl = 3600  # 1 hour

    def get_user_features(self, user_id: str) -> dict:
        cache_key = f"user_features:{user_id}"
        cached = self.redis.get(cache_key)

        if cached:
            return json.loads(cached)

        # Fetch from database
        features = fetch_from_db(user_id)

        # Cache
        self.redis.setex(cache_key, self.ttl, json.dumps(features))
        return features

2. Model Serving

Serve multiple models for A/B testing:

class ModelServing:
    """Serve multiple recommendation models."""

    def __init__(self):
        self.models = {}

    def register_model(self, name: str, model, is_default: bool = False):
        self.models[name] = model
        if is_default:
            self.default_model = name

    def predict(self, user_features: dict, item_features: list, model_name: str = None):
        model = self.models.get(model_name or self.default_model)
        return model.predict(user_features, item_features)

3. Caching Layer

Multi-tier caching for low latency:

class TieredCache:
    """L1 (memory) -> L2 (Redis) -> L3 (database)."""

    def __init__(self, redis_client):
        self.l1_cache = {}  # In-memory
        self.redis = redis_client  # L2

    def get(self, key: str):
        # L1: In-memory (fastest)
        if key in self.l1_cache:
            return self.l1_cache[key]

        # L2: Redis
        cached = self.redis.get(key)
        if cached:
            value = json.loads(cached)
            self.l1_cache[key] = value  # Promote to L1
            return value

        # L3: Miss (fetch from database)
        return None

Key Metrics

Known Issues Prevention

1. Cold Start for New Users

Problem: No recommendations for users without history, poor initial experience.

Solution: Use popularity-based fallback:

def get_recommendations(user_id: str, n: int = 10):
    user_features = feature_store.get_user_features(user_id)

    # Check if new user (no purchase history)
    if user_features.get('total_purchases', 0) == 0:
        # Fallback to popular items
        return get_popular_items(n)

    # Personalized recommendations
    return generate_personalized_recs(user_id, n)

2. Cache Invalidation on User Actions

Problem: User makes purchase, cache still shows purchased item in recommendations.

Solution: Invalidate cache on relevant actions:

INVALIDATING_ACTIONS = {'purchase', 'rating', 'add_to_cart'}

def on_user_action(user_id: str, action: str):
    if action in INVALIDATING_ACTIONS:
        cache_key = f"recs:{user_id}:*"
        redis_client.delete(cache_key)
        logger.info(f"Invalidated cache for {user_id} due to {action}")

3. Thundering Herd on Cache Expiry

Problem: Many users' caches expire simultaneously, overload database/model.

Solution: Add random jitter to TTL:

import random

def set_cache(key: str, value: dict, base_ttl: int = 300):
    # Add ±10% jitter
    jitter = random.uniform(-0.1, 0.1) * base_ttl
    ttl = int(base_ttl + jitter)
    redis_client.setex(key, ttl, json.dumps(value))

4. Poor Diversity = Filter Bubble

Problem: Recommendations too similar, users only see same category.

Solution: Implement diversity constraint:

def rank_with_diversity(items: list, scores: list, n: int = 10):
    selected = []
    category_counts = {}

    for item, score in sorted(zip(items, scores), key=lambda x: -x[1]):
        category = item['category']

        # Limit 3 items per category
        if category_counts.get(category, 0) >= 3:
            continue

        selected.append(item)
        category_counts[category] = category_counts.get(category, 0) + 1

        if len(selected) >= n:
            break

    return selected

5. No Monitoring = Silent Degradation

Problem: Recommendation quality drops, nobody notices until users complain.

Solution: Continuous monitoring with alerts:

from prometheus_client import Counter, Histogram

recommendation_clicks = Counter('recommendation_clicks_total')
recommendation_latency = Histogram('recommendation_latency_seconds')

@app.post("/recommendations")
async def get_recommendations(user_id: str):
    start = time.time()

    recs = generate_recs(user_id)

    latency = time.time() - start
    recommendation_latency.observe(latency)

    return recs

@app.post("/track/click")
async def track_click(user_id: str, item_id: str):
    recommendation_clicks.inc()
    # Alert if CTR drops below 3%

6. Stale Features = Outdated Recommendations

Problem: User preferences change but features don't update, recommendations irrelevant.

Solution: Set appropriate TTLs and update triggers:

class FeatureStore:
    def __init__(self, redis_client):
        self.redis = redis_client
        # Shorter TTL for frequently changing features
        self.user_ttl = 300  # 5 minutes
        self.item_ttl = 3600  # 1 hour

    def update_on_event(self, user_id: str, event: str):
        # Invalidate on important events
        if event in ['purchase', 'rating']:
            self.redis.delete(f"user_features:{user_id}")
            logger.info(f"Refreshed features for {user_id}")

7. A/B Test Sample Size Too Small

Problem: Declare winner too early, results not statistically significant.

Solution: Calculate required sample size first:

def calculate_sample_size(
    baseline_rate: float,
    min_detectable_effect: float,
    alpha: float = 0.05,
    power: float = 0.8
) -> int:
    """Calculate required sample size per variant."""
    from scipy import stats

    z_alpha = stats.norm.ppf(1 - alpha/2)
    z_beta = stats.norm.ppf(power)

    p1 = baseline_rate
    p2 = baseline_rate * (1 + min_detectable_effect)
    p_avg = (p1 + p2) / 2

    n = (
        (z_alpha + z_beta)**2 * 2 * p_avg * (1 - p_avg) /
        (p2 - p1)**2
    )

    return int(n)

# Example: detect 10% lift with baseline CTR=5%
n_required = calculate_sample_size(
    baseline_rate=0.05,
    min_detectable_effect=0.10
)
print(f"Required sample size: {n_required} per variant")
# Wait until both variants reach this size before concluding

When to Load References

Load reference files for detailed production implementations:

• Production Architecture: Load references/production-architecture.md for complete FeatureStore, ModelServing, and RecommendationService implementations with batch fetching, caching integration, and FastAPI deployment patterns.

• Caching Strategies: Load references/caching-strategies.md when implementing multi-tier caching (L1/L2/L3), cache warming, invalidation strategies, probabilistic refresh, or thundering herd prevention.

• A/B Testing Framework: Load references/ab-testing-framework.md for deterministic variant assignment, Thompson sampling (multi-armed bandits), Bayesian and frequentist significance testing, and experiment tracking.

• Monitoring & Alerting: Load references/monitoring-alerting.md for Prometheus metrics integration, dashboard endpoints, alert rules, and quality monitoring (diversity, coverage).

Best Practices

1. Feature Precomputation: Compute features offline, serve from cache 2. Batch Fetching: Use Redis MGET for multiple users/items 3. Cache Aggressively: 5-15 minute TTL for user recommendations 4. Fail Gracefully: Return popular items if personalization fails 5. Monitor Everything: Track CTR, latency, diversity, coverage 6. A/B Test Continuously: Always be experimenting with new algorithms 7. Diversity Constraint: Ensure varied recommendations 8. Explain Recommendations: Provide reasons ("Highly rated", "Popular")

Common Patterns

Recommendation Service

class RecommendationService:
    def __init__(self, feature_store, model_serving, cache):
        self.feature_store = feature_store
        self.model_serving = model_serving
        self.cache = cache

    def get_recommendations(self, user_id: str, n: int = 10):
        # 1. Check cache
        cached = self.cache.get(f"recs:{user_id}:{n}")
        if cached:
            return cached

        # 2. Get features
        user_features = self.feature_store.get_user_features(user_id)
        candidates = self.get_candidates(user_id)

        # 3. Score candidates
        scores = self.model_serving.predict(user_features, candidates)

        # 4. Rank with diversity
        recommendations = self.rank_with_diversity(candidates, scores, n)

        # 5. Cache
        self.cache.set(f"recs:{user_id}:{n}", recommendations, ttl=300)

        return recommendations

A/B Testing

def assign_variant(user_id: str, experiment_id: str) -> str:
    """Deterministic assignment - same user always gets same variant."""
    import hashlib

    hash_input = f"{user_id}:{experiment_id}"
    hash_value = int(hashlib.md5(hash_input.encode()).hexdigest(), 16)

    # 50/50 split
    return 'control' if hash_value % 2 == 0 else 'treatment'

# Usage
variant = assign_variant('user_123', 'rec_algo_v2')
model_name = 'main' if variant == 'control' else 'experimental'
recs = get_recommendations(user_id, model_name=model_name)

Monitoring

from prometheus_client import Counter, Histogram

requests_total = Counter('recommendation_requests_total', ['status'])
latency_seconds = Histogram('recommendation_latency_seconds')

@app.post("/recommendations")
async def get_recommendations(user_id: str):
    with latency_seconds.time():
        try:
            recs = generate_recs(user_id)
            requests_total.labels(status='success').inc()
            return recs
        except Exception as e:
            requests_total.labels(status='error').inc()
            raise

A/B Testing Framework for Recommendations

Complete A/B testing implementation for recommendation systems including variant assignment, statistical significance testing, and experiment tracking.

Overview

A/B testing is critical for recommendation systems:

• Validate improvements: Test before rolling out
• Data-driven decisions: Replace intuition with metrics
• Continuous optimization: Always be experimenting

Variant Assignment

Deterministic User Assignment

import hashlib
from typing import List, Optional
from dataclasses import dataclass
from datetime import datetime
import logging

logger = logging.getLogger(__name__)


@dataclass
class Experiment:
    """Experiment configuration."""
    id: str
    name: str
    variants: List[str]
    traffic_allocation: List[float]  # Must sum to 1.0
    start_date: datetime
    end_date: Optional[datetime] = None
    active: bool = True


class ABTestingFramework:
    """
    A/B testing framework with deterministic assignment.

    Users are consistently assigned to same variant.
    """

    def __init__(self):
        self.experiments = {}

    def create_experiment(
        self,
        experiment_id: str,
        name: str,
        variants: List[str],
        traffic_allocation: Optional[List[float]] = None
    ) -> Experiment:
        """
        Create new experiment.

        Args:
            experiment_id: Unique experiment ID
            name: Human-readable name
            variants: List of variant names (e.g., ['control', 'treatment'])
            traffic_allocation: Traffic split (e.g., [0.5, 0.5])

        Returns:
            Experiment object
        """
        if not traffic_allocation:
            # Equal split
            traffic_allocation = [1.0 / len(variants)] * len(variants)

        if abs(sum(traffic_allocation) - 1.0) > 0.001:
            raise ValueError("Traffic allocation must sum to 1.0")

        experiment = Experiment(
            id=experiment_id,
            name=name,
            variants=variants,
            traffic_allocation=traffic_allocation,
            start_date=datetime.now()
        )

        self.experiments[experiment_id] = experiment
        logger.info(f"Created experiment: {name} ({experiment_id})")

        return experiment

    def assign_variant(
        self,
        user_id: str,
        experiment_id: str
    ) -> str:
        """
        Assign user to experiment variant.

        Uses deterministic hashing - same user always gets same variant.

        Args:
            user_id: User ID
            experiment_id: Experiment ID

        Returns:
            Variant name
        """
        if experiment_id not in self.experiments:
            raise ValueError(f"Experiment {experiment_id} not found")

        experiment = self.experiments[experiment_id]

        if not experiment.active:
            # Return control for inactive experiments
            return experiment.variants[0]

        # Hash user_id + experiment_id for deterministic assignment
        hash_input = f"{user_id}:{experiment_id}"
        hash_digest = hashlib.md5(hash_input.encode()).hexdigest()
        hash_value = int(hash_digest, 16)

        # Map hash to variant based on traffic allocation
        random_value = (hash_value % 10000) / 10000.0

        cumulative = 0.0
        for variant, allocation in zip(
            experiment.variants,
            experiment.traffic_allocation
        ):
            cumulative += allocation
            if random_value < cumulative:
                return variant

        # Fallback (should never happen)
        return experiment.variants[0]

    def get_variant_with_logging(
        self,
        user_id: str,
        experiment_id: str,
        logger_fn=None
    ) -> str:
        """
        Assign variant and log assignment.

        Args:
            user_id: User ID
            experiment_id: Experiment ID
            logger_fn: Function to log assignment (e.g., to database)

        Returns:
            Variant name
        """
        variant = self.assign_variant(user_id, experiment_id)

        # Log assignment for analysis
        if logger_fn:
            logger_fn({
                'user_id': user_id,
                'experiment_id': experiment_id,
                'variant': variant,
                'timestamp': datetime.now().isoformat()
            })

        return variant


# Example usage
ab_test = ABTestingFramework()

# Create experiment: test new recommendation algorithm
ab_test.create_experiment(
    experiment_id='rec_algo_v2',
    name='New Recommendation Algorithm Test',
    variants=['control', 'treatment'],
    traffic_allocation=[0.5, 0.5]
)

# Assign user
variant = ab_test.assign_variant(
    user_id='user_123',
    experiment_id='rec_algo_v2'
)
# variant = 'control' or 'treatment' (consistent for user_123)

Multi-Armed Bandit Alternative

Thompson Sampling

import numpy as np
from scipy import stats


class ThompsonSampling:
    """
    Thompson Sampling for dynamic traffic allocation.

    Automatically shifts traffic to better-performing variants.
    """

    def __init__(self, variant_names: List[str]):
        self.variants = variant_names

        # Beta distribution parameters (alpha, beta) for each variant
        # Start with uninformative prior: Beta(1, 1)
        self.successes = {v: 1 for v in variant_names}  # alpha
        self.failures = {v: 1 for v in variant_names}   # beta

    def select_variant(self) -> str:
        """
        Select variant using Thompson Sampling.

        Returns:
            Selected variant name
        """
        # Sample from each variant's Beta distribution
        samples = {
            variant: np.random.beta(
                self.successes[variant],
                self.failures[variant]
            )
            for variant in self.variants
        }

        # Return variant with highest sample
        return max(samples, key=samples.get)

    def update(self, variant: str, reward: float):
        """
        Update variant statistics.

        Args:
            variant: Variant name
            reward: 1.0 for success, 0.0 for failure
        """
        if reward > 0:
            self.successes[variant] += 1
        else:
            self.failures[variant] += 1

    def get_statistics(self) -> dict:
        """Get current statistics for all variants."""
        return {
            variant: {
                'successes': self.successes[variant],
                'failures': self.failures[variant],
                'estimated_ctr': self.successes[variant] / (
                    self.successes[variant] + self.failures[variant]
                )
            }
            for variant in self.variants
        }


# Example usage
bandit = ThompsonSampling(variants=['control', 'treatment_a', 'treatment_b'])

# For each user
for user_id in users:
    variant = bandit.select_variant()

    # Show recommendations from variant
    recommendations = get_recommendations(user_id, variant)

    # Track if user clicked/converted
    clicked = user_clicked(recommendations)
    reward = 1.0 if clicked else 0.0

    # Update bandit
    bandit.update(variant, reward)

# Check performance
print(bandit.get_statistics())

Statistical Significance Testing

Bayesian A/B Test Analysis

from typing import Tuple


class BayesianABTest:
    """Bayesian analysis of A/B test results."""

    def __init__(self):
        pass

    def analyze(
        self,
        control_successes: int,
        control_trials: int,
        treatment_successes: int,
        treatment_trials: int,
        prior_alpha: float = 1.0,
        prior_beta: float = 1.0
    ) -> dict:
        """
        Analyze A/B test using Bayesian inference.

        Args:
            control_successes: Number of conversions in control
            control_trials: Number of users in control
            treatment_successes: Number of conversions in treatment
            treatment_trials: Number of users in treatment
            prior_alpha: Prior alpha for Beta distribution
            prior_beta: Prior beta for Beta distribution

        Returns:
            Dictionary with analysis results
        """
        # Posterior distributions
        control_alpha = prior_alpha + control_successes
        control_beta = prior_beta + (control_trials - control_successes)

        treatment_alpha = prior_alpha + treatment_successes
        treatment_beta = prior_beta + (treatment_trials - treatment_successes)

        # Expected values (means)
        control_mean = control_alpha / (control_alpha + control_beta)
        treatment_mean = treatment_alpha / (treatment_alpha + treatment_beta)

        # Probability that treatment > control
        # Monte Carlo estimation
        n_samples = 100000
        control_samples = np.random.beta(control_alpha, control_beta, n_samples)
        treatment_samples = np.random.beta(treatment_alpha, treatment_beta, n_samples)

        prob_treatment_better = np.mean(treatment_samples > control_samples)

        # Relative lift
        relative_lift = (treatment_mean - control_mean) / control_mean

        # Credible intervals (95%)
        control_ci = stats.beta.interval(
            0.95,
            control_alpha,
            control_beta
        )
        treatment_ci = stats.beta.interval(
            0.95,
            treatment_alpha,
            treatment_beta
        )

        return {
            'control': {
                'mean': control_mean,
                'credible_interval': control_ci,
                'trials': control_trials,
                'successes': control_successes
            },
            'treatment': {
                'mean': treatment_mean,
                'credible_interval': treatment_ci,
                'trials': treatment_trials,
                'successes': treatment_successes
            },
            'probability_treatment_better': prob_treatment_better,
            'relative_lift': relative_lift,
            'significant': prob_treatment_better > 0.95  # 95% threshold
        }


# Example usage
analyzer = BayesianABTest()

results = analyzer.analyze(
    control_successes=850,
    control_trials=10000,
    treatment_successes=920,
    treatment_trials=10000
)

print(f"Control CTR: {results['control']['mean']:.2%}")
print(f"Treatment CTR: {results['treatment']['mean']:.2%}")
print(f"Probability treatment is better: {results['probability_treatment_better']:.2%}")
print(f"Relative lift: {results['relative_lift']:+.2%}")
print(f"Significant: {results['significant']}")

Frequentist Significance Test

from scipy import stats as scipy_stats


def ab_test_significance(
    control_conversions: int,
    control_total: int,
    treatment_conversions: int,
    treatment_total: int,
    alpha: float = 0.05
) -> dict:
    """
    Frequentist A/B test significance using two-proportion z-test.

    Args:
        control_conversions: Conversions in control
        control_total: Total users in control
        treatment_conversions: Conversions in treatment
        treatment_total: Total users in treatment
        alpha: Significance level (default 0.05)

    Returns:
        Dictionary with test results
    """
    # Conversion rates
    p_control = control_conversions / control_total
    p_treatment = treatment_conversions / treatment_total

    # Pooled proportion
    p_pooled = (control_conversions + treatment_conversions) / (
        control_total + treatment_total
    )

    # Standard error
    se = np.sqrt(
        p_pooled * (1 - p_pooled) * (
            1/control_total + 1/treatment_total
        )
    )

    # Z-score
    z_score = (p_treatment - p_control) / se

    # P-value (two-tailed)
    p_value = 2 * (1 - scipy_stats.norm.cdf(abs(z_score)))

    # Confidence interval for difference
    se_diff = np.sqrt(
        (p_control * (1 - p_control) / control_total) +
        (p_treatment * (1 - p_treatment) / treatment_total)
    )

    ci_lower = (p_treatment - p_control) - 1.96 * se_diff
    ci_upper = (p_treatment - p_control) + 1.96 * se_diff

    return {
        'control_rate': p_control,
        'treatment_rate': p_treatment,
        'difference': p_treatment - p_control,
        'relative_lift': (p_treatment - p_control) / p_control,
        'z_score': z_score,
        'p_value': p_value,
        'significant': p_value < alpha,
        'confidence_interval': (ci_lower, ci_upper),
        'sample_size': {
            'control': control_total,
            'treatment': treatment_total
        }
    }


# Example
result = ab_test_significance(
    control_conversions=450,
    control_total=5000,
    treatment_conversions=520,
    treatment_total=5000
)

print(f"Control Rate: {result['control_rate']:.2%}")
print(f"Treatment Rate: {result['treatment_rate']:.2%}")
print(f"Lift: {result['relative_lift']:+.2%}")
print(f"P-value: {result['p_value']:.4f}")
print(f"Significant (p<0.05): {result['significant']}")

Experiment Tracking

Complete Experiment Tracker

from typing import Dict, List
import sqlite3
import json


class ExperimentTracker:
    """
    Track experiment assignments and outcomes.

    Note: This implementation uses context managers for safe connection handling.
    For production use with high traffic, consider:
    1. Using a connection pool (e.g., sqlitepool for SQLite)
    2. Migrating to PostgreSQL with pgbouncer for connection pooling
    3. Implementing connection reuse for long-lived processes
    """

    def __init__(self, db_path: str = 'experiments.db'):
        self.db_path = db_path
        self._init_db()

    def _init_db(self):
        """Initialize database schema."""
        try:
            with sqlite3.connect(self.db_path) as conn:
                cursor = conn.cursor()

                # Assignments table
                cursor.execute("""
                    CREATE TABLE IF NOT EXISTS assignments (
                        user_id TEXT,
                        experiment_id TEXT,
                        variant TEXT,
                        timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
                        PRIMARY KEY (user_id, experiment_id)
                    )
                """)

                # Events table
                cursor.execute("""
                    CREATE TABLE IF NOT EXISTS events (
                        user_id TEXT,
                        experiment_id TEXT,
                        event_type TEXT,
                        event_value REAL,
                        timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
                    )
                """)

                conn.commit()
        except sqlite3.Error as e:
            logger.error(f"Database initialization failed: {e}")
            raise

    def log_assignment(
        self,
        user_id: str,
        experiment_id: str,
        variant: str
    ):
        """Log variant assignment."""
        try:
            with sqlite3.connect(self.db_path) as conn:
                cursor = conn.cursor()

                cursor.execute("""
                    INSERT OR REPLACE INTO assignments
                    (user_id, experiment_id, variant)
                    VALUES (?, ?, ?)
                """, (user_id, experiment_id, variant))

                conn.commit()
        except sqlite3.Error as e:
            logger.error(f"Failed to log assignment for user {user_id}: {e}")
            raise

    def log_event(
        self,
        user_id: str,
        experiment_id: str,
        event_type: str,
        event_value: float = 1.0
    ):
        """
        Log user event (click, conversion, etc.).

        Args:
            user_id: User ID
            experiment_id: Experiment ID
            event_type: Type of event ('click', 'conversion', etc.)
            event_value: Numeric value (e.g., revenue, 1.0 for binary)
        """
        try:
            with sqlite3.connect(self.db_path) as conn:
                cursor = conn.cursor()

                cursor.execute("""
                    INSERT INTO events
                    (user_id, experiment_id, event_type, event_value)
                    VALUES (?, ?, ?, ?)
                """, (user_id, experiment_id, event_type, event_value))

                conn.commit()
        except sqlite3.Error as e:
            logger.error(f"Failed to log event {event_type} for user {user_id}: {e}")
            raise

    def get_experiment_results(
        self,
        experiment_id: str,
        event_type: str = 'conversion'
    ) -> Dict:
        """
        Get experiment results by variant.

        Args:
            experiment_id: Experiment ID
            event_type: Event type to analyze

        Returns:
            Dictionary with results per variant
        """
        try:
            with sqlite3.connect(self.db_path) as conn:
                conn.row_factory = sqlite3.Row
                cursor = conn.cursor()

                # Get assignments and events by variant
                cursor.execute("""
                    SELECT
                        a.variant,
                        COUNT(DISTINCT a.user_id) as total_users,
                        COUNT(DISTINCT e.user_id) as converted_users,
                        SUM(COALESCE(e.event_value, 0)) as total_value
                    FROM assignments a
                    LEFT JOIN events e ON
                        a.user_id = e.user_id AND
                        a.experiment_id = e.experiment_id AND
                        e.event_type = ?
                    WHERE a.experiment_id = ?
                    GROUP BY a.variant
                """, (event_type, experiment_id))

                results = {}
                for row in cursor.fetchall():
                    variant = row['variant']
                    total = row['total_users']
                    conversions = row['converted_users'] or 0

                    results[variant] = {
                        'total_users': total,
                        'conversions': conversions,
                        'conversion_rate': conversions / total if total > 0 else 0,
                        'total_value': row['total_value'] or 0
                    }

                return results
        except sqlite3.Error as e:
            logger.error(f"Failed to get results for experiment {experiment_id}: {e}")
            raise


# Example usage
tracker = ExperimentTracker()

# Log assignment
tracker.log_assignment(
    user_id='user_123',
    experiment_id='rec_algo_v2',
    variant='treatment'
)

# Log conversion
tracker.log_event(
    user_id='user_123',
    experiment_id='rec_algo_v2',
    event_type='conversion',
    event_value=1.0
)

# Get results
results = tracker.get_experiment_results('rec_algo_v2')
print(results)
# {'control': {'total_users': 5000, 'conversions': 450, ...},
#  'treatment': {'total_users': 5000, 'conversions': 520, ...}}

Best Practices

1. Deterministic Assignment: Use hashing for consistent user experience 2. Sufficient Sample Size: Calculate required size before starting 3. Monitor Both Metrics: Track primary (conversion) and secondary (engagement) metrics 4. Avoid Peeking: Don't stop early just because one variant is winning 5. Consider Novelty Effect: Run experiments for 1-2 weeks minimum 6. Segment Analysis: Analyze by user segments (new vs returning, etc.) 7. Log Everything: Track assignments and outcomes for post-hoc analysis 8. Bayesian Approach: More intuitive interpretation than p-values

Caching Strategies for Recommendation Systems

Production caching patterns for low-latency recommendations with Redis, in-memory caching, and cache invalidation strategies.

Overview

Effective caching is critical for recommendation systems:

• Sub-millisecond latency: Serve from memory
• Cost reduction: Fewer model inferences
• Scalability: Handle traffic spikes

Redis Caching Patterns

Basic Recommendation Caching

import redis
import json
from typing import List, Dict, Optional
import logging
from datetime import timedelta

logger = logging.getLogger(__name__)


class RecommendationCache:
    """Redis-backed cache for recommendations."""

    def __init__(
        self,
        redis_client: redis.Redis,
        ttl_seconds: int = 300  # 5 minutes
    ):
        self.redis = redis_client
        self.ttl = ttl_seconds

    def get(self, user_id: str, n: int) -> Optional[List[Dict]]:
        """
        Get cached recommendations.

        Args:
            user_id: User ID
            n: Number of recommendations

        Returns:
            List of recommendations or None if not cached
        """
        cache_key = f"recs:{user_id}:{n}"

        try:
            cached = self.redis.get(cache_key)

            if cached:
                logger.debug(f"Cache hit: {cache_key}")
                return json.loads(cached)

            logger.debug(f"Cache miss: {cache_key}")
            return None

        except Exception as e:
            logger.error(f"Cache get failed: {e}")
            return None

    def set(
        self,
        user_id: str,
        n: int,
        recommendations: List[Dict],
        ttl: Optional[int] = None
    ):
        """
        Cache recommendations.

        Args:
            user_id: User ID
            n: Number of recommendations
            recommendations: List of recommendation dicts
            ttl: Optional custom TTL (seconds)
        """
        cache_key = f"recs:{user_id}:{n}"
        ttl = ttl or self.ttl

        try:
            self.redis.setex(
                cache_key,
                ttl,
                json.dumps(recommendations)
            )
            logger.debug(f"Cached: {cache_key} (TTL: {ttl}s)")

        except Exception as e:
            logger.error(f"Cache set failed: {e}")

    def invalidate(self, user_id: str):
        """
        Invalidate all cached recommendations for user.

        Args:
            user_id: User ID
        """
        pattern = f"recs:{user_id}:*"

        try:
            # Find keys matching pattern
            keys = self.redis.keys(pattern)

            if keys:
                self.redis.delete(*keys)
                logger.info(f"Invalidated {len(keys)} cache entries for {user_id}")

        except Exception as e:
            logger.error(f"Cache invalidation failed: {e}")

    def batch_get(
        self,
        user_ids: List[str],
        n: int
    ) -> Dict[str, Optional[List[Dict]]]:
        """
        Get recommendations for multiple users.

        Args:
            user_ids: List of user IDs
            n: Number of recommendations

        Returns:
            Dictionary mapping user_id to recommendations (or None)
        """
        cache_keys = [f"recs:{user_id}:{n}" for user_id in user_ids]

        try:
            # Batch fetch
            cached_values = self.redis.mget(cache_keys)

            results = {}
            for user_id, cached in zip(user_ids, cached_values):
                if cached:
                    results[user_id] = json.loads(cached)
                else:
                    results[user_id] = None

            hits = sum(1 for v in results.values() if v is not None)
            logger.info(f"Batch get: {hits}/{len(user_ids)} cache hits")

            return results

        except Exception as e:
            logger.error(f"Batch cache get failed: {e}")
            return {user_id: None for user_id in user_ids}


### Tiered Caching Strategy

from functools import lru_cache import time

class TieredCache: """ Multi-tier caching: L1 (in-memory) → L2 (Redis) → L3 (database).

L1: Fast, small capacity, short TTL L2: Medium speed, larger capacity, medium TTL L3: Slow, unlimited capacity, source of truth """

def __init__( self, redis_client: redis.Redis, l1_max_size: int = 1000, l1_ttl: int = 60, # 1 minute l2_ttl: int = 300 # 5 minutes ): self.redis = redis_client self.l1_ttl = l1_ttl self.l2_ttl = l2_ttl

L1: In-memory cache with LRU eviction

self.l1_cache = {} self.l1_max_size = l1_max_size self.l1_timestamps = {}

def get(self, key: str) -> Optional[Dict]: """Get value from tiered cache."""

L1: In-memory (fastest)

if key in self.l1_cache:

Check if expired

if time.time() - self.l1_timestamps[key] < self.l1_ttl: logger.debug(f"L1 hit: {key}") return self.l1_cache[key] else:

Expired, remove

del self.l1_cache[key] del self.l1_timestamps[key]

L2: Redis

try: cached = self.redis.get(key) if cached: logger.debug(f"L2 hit: {key}") value = json.loads(cached)

Promote to L1

self._set_l1(key, value)

return value

except Exception as e: logger.error(f"L2 cache error: {e}")

L3: Miss (caller will fetch from database)

logger.debug(f"Cache miss: {key}") return None

def set(self, key: str, value: Dict): """Set value in all cache tiers."""

L1: In-memory

self._set_l1(key, value)

L2: Redis

try: self.redis.setex( key, self.l2_ttl, json.dumps(value) ) except Exception as e: logger.error(f"L2 cache set error: {e}")

def _set_l1(self, key: str, value: Dict): """Set value in L1 cache with LRU eviction."""

Evict if at capacity

if len(self.l1_cache) >= self.l1_max_size and key not in self.l1_cache:

Find oldest entry

oldest_key = min( self.l1_timestamps.keys(), key=lambda k: self.l1_timestamps[k] ) del self.l1_cache[oldest_key] del self.l1_timestamps[oldest_key]

self.l1_cache[key] = value self.l1_timestamps[key] = time.time()

def invalidate(self, key: str): """Invalidate key in all tiers."""

L1

if key in self.l1_cache: del self.l1_cache[key] del self.l1_timestamps[key]

L2

try: self.redis.delete(key) except Exception as e: logger.error(f"L2 invalidation error: {e}")

## Cache Warming Strategies

### Precompute Popular User Recommendations

from concurrent.futures import ThreadPoolExecutor import schedule

class CacheWarmer: """Precompute and cache recommendations for popular users."""

def __init__( self, recommendation_service, cache: RecommendationCache, db_client ): self.rec_service = recommendation_service self.cache = cache self.db = db_client

def warm_top_users(self, top_n: int = 10000, workers: int = 10): """ Precompute recommendations for top N active users.

Args: top_n: Number of users to warm workers: Thread pool size """

Get most active users

user_ids = self._get_top_active_users(top_n)

logger.info(f"Warming cache for {len(user_ids)} users")

Parallel warm

with ThreadPoolExecutor(max_workers=workers) as executor: executor.map(self._warm_user, user_ids)

logger.info(f"Cache warming complete for {len(user_ids)} users")

def _warm_user(self, user_id: str): """Warm cache for single user.""" try:

Generate recommendations

recs = self.rec_service.get_recommendations( user_id=user_id, n=20 # Cache top 20 )

Cache is updated inside get_recommendations

logger.debug(f"Warmed cache for user {user_id}")

except Exception as e: logger.error(f"Failed to warm user {user_id}: {e}")

def _get_top_active_users(self, limit: int) -> List[str]: """Get most active users from database.""" query = """ SELECT user_id FROM user_activity ORDER BY last_active DESC LIMIT %s """

results = self.db.execute(query, (limit,)) return [row['user_id'] for row in results]

def schedule_warming(self, hour: int = 2): """ Schedule daily cache warming.

Args: hour: Hour to run (0-23) """ schedule.every().day.at(f"{hour:02d}:00").do( self.warm_top_users )

logger.info(f"Scheduled cache warming daily at {hour}:00")

Run scheduler loop (in production, use Celery/Airflow)

while True: schedule.run_pending() time.sleep(60)

## Cache Invalidation

### Event-Driven Invalidation

from typing import Set

class CacheInvalidator: """Invalidate cache based on user actions."""

def __init__(self, cache: RecommendationCache): self.cache = cache

Actions that should invalidate cache

self.invalidating_actions = { 'purchase', 'rating', 'add_to_cart', 'wishlist_add', 'profile_update' }

def on_user_action(self, user_id: str, action: str): """ Handle user action and invalidate if needed.

Args: user_id: User who performed action action: Action type """ if action in self.invalidating_actions: logger.info(f"Invalidating cache for {user_id} due to {action}") self.cache.invalidate(user_id)

def on_item_update(self, item_id: str): """ Invalidate caches that included this item.

Note: This is expensive! Only use for critical updates. """

In production, maintain an index: item_id -> [user_ids]

For now, invalidate all (not recommended for production)

logger.warning(f"Item {item_id} updated - full cache flush needed")

self.cache.redis.flushdb() # Use with caution!

## Probabilistic Cache Warming

### Thunder Herd Prevention

import random

class ProbabilisticCache: """ Cache with probabilistic early expiration to prevent thundering herd.

When many requests hit expired cache simultaneously, they all trigger expensive recomputation. This spreads out expirations. """

def __init__(self, redis_client: redis.Redis, base_ttl: int = 300): self.redis = redis_client self.base_ttl = base_ttl

def set(self, key: str, value: Dict): """Set with randomized TTL."""

Add jitter: ±10% of base TTL

jitter = random.uniform(-0.1, 0.1) * self.base_ttl ttl = int(self.base_ttl + jitter)

self.redis.setex(key, ttl, json.dumps(value))

def get_with_refresh( self, key: str, refresh_fn, refresh_probability: float = 0.1 ) -> Optional[Dict]: """ Get value, with probabilistic early refresh.

Args: key: Cache key refresh_fn: Function to recompute value refresh_probability: Chance to refresh before expiry

Returns: Cached or fresh value """ cached = self.redis.get(key)

if cached:

Probabilistically refresh even though cached

if random.random() < refresh_probability: logger.debug(f"Probabilistic refresh: {key}") value = refresh_fn() self.set(key, value) return value

return json.loads(cached)

Cache miss - recompute

logger.debug(f"Cache miss, recomputing: {key}") value = refresh_fn() self.set(key, value) return value

## Performance Monitoring

### Cache Metrics Tracker

from prometheus_client import Counter, Histogram, Gauge

cache_hits = Counter( 'recommendation_cache_hits_total', 'Total cache hits', ['cache_tier'] )

cache_misses = Counter( 'recommendation_cache_misses_total', 'Total cache misses', ['cache_tier'] )

cache_latency = Histogram( 'recommendation_cache_latency_seconds', 'Cache operation latency', ['operation'], buckets=[0.001, 0.005, 0.01, 0.05, 0.1] )

cache_size = Gauge( 'recommendation_cache_size_bytes', 'Current cache size', ['cache_tier'] )

class MonitoredCache(RecommendationCache): """Cache with Prometheus metrics."""

def get(self, user_id: str, n: int) -> Optional[List[Dict]]: import time start = time.time()

result = super().get(user_id, n)

latency = time.time() - start cache_latency.labels(operation='get').observe(latency)

if result: cache_hits.labels(cache_tier='redis').inc() else: cache_misses.labels(cache_tier='redis').inc()

return result

def set(self, user_id: str, n: int, recommendations: List[Dict], ttl=None): import time start = time.time()

super().set(user_id, n, recommendations, ttl)

latency = time.time() - start cache_latency.labels(operation='set').observe(latency)

## Best Practices

1. **Layer Your Caching**: Use L1 (memory) → L2 (Redis) → L3 (database)
2. **Set Appropriate TTLs**: Short TTL (1-5 min) for personalized, longer for popular
3. **Invalidate Strategically**: Only invalidate when user behavior changes
4. **Warm Popular Caches**: Precompute for active users during low-traffic hours
5. **Monitor Hit Rates**: Track cache effectiveness, optimize if <80% hit rate
6. **Handle Failures Gracefully**: Cache failures shouldn't break recommendations
7. **Add Jitter to TTLs**: Prevent thundering herd with randomized expiration
8. **Batch Operations**: Use MGET/MSET for multiple users

Monitoring and Alerting for Recommendation Systems

Production monitoring, metrics dashboards, alert configuration, and performance tracking for recommendation systems.

Key Metrics to Monitor

Business Metrics

• Click-Through Rate (CTR): % of recommendations clicked
• Conversion Rate: % of clicks that result in purchase/action
• Revenue Per User (RPU): Average revenue from recommendations
• Coverage: % of catalog being recommended
• Diversity: Variety in recommendations (opposite of concentration)

Technical Metrics

• Latency: P50, P95, P99 response times
• Throughput: Requests per second
• Cache Hit Rate: % of requests served from cache
• Error Rate: % of failed requests
• Model Staleness: Time since last model update

Prometheus Metrics Integration

Complete Metrics Setup

from prometheus_client import Counter, Histogram, Gauge, Summary
from prometheus_client import generate_latest, REGISTRY
from flask import Flask, Response
import time
import logging
from typing import Dict, List
from functools import wraps
import requests
from requests.exceptions import RequestException

logger = logging.getLogger(__name__)

# Define metrics

# Request metrics
recommendation_requests = Counter(
    'recommendation_requests_total',
    'Total recommendation requests',
    ['model_version', 'status']
)

recommendation_latency = Histogram(
    'recommendation_latency_seconds',
    'Recommendation generation latency',
    ['model_version'],
    buckets=[0.01, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0]
)

# Business metrics
recommendations_clicked = Counter(
    'recommendations_clicked_total',
    'Total clicked recommendations',
    ['model_version', 'position']  # position: 1-10
)

recommendations_converted = Counter(
    'recommendations_converted_total',
    'Total recommendations that converted',
    ['model_version']
)

recommendation_revenue = Summary(
    'recommendation_revenue_dollars',
    'Revenue from recommendations',
    ['model_version']
)

# Cache metrics
cache_hits = Counter(
    'recommendation_cache_hits_total',
    'Total cache hits'
)

cache_misses = Counter(
    'recommendation_cache_misses_total',
    'Total cache misses'
)

# Quality metrics
recommendation_diversity = Gauge(
    'recommendation_diversity_score',
    'Diversity score (0-1)',
    ['model_version']
)

catalog_coverage = Gauge(
    'recommendation_catalog_coverage',
    'Percentage of catalog recommended',
    ['model_version']
)

model_staleness_hours = Gauge(
    'recommendation_model_staleness_hours',
    'Hours since model was last updated',
    ['model_version']
)


class MetricsCollector:
    """Collect and expose recommendation metrics."""

    def __init__(self, model_version: str = 'v1'):
        self.model_version = model_version

    def track_request(self, success: bool = True):
        """Track recommendation request."""
        status = 'success' if success else 'error'
        recommendation_requests.labels(
            model_version=self.model_version,
            status=status
        ).inc()

    def track_latency(self, latency_seconds: float):
        """Track request latency."""
        recommendation_latency.labels(
            model_version=self.model_version
        ).observe(latency_seconds)

    def track_click(self, position: int):
        """
        Track recommendation click.

        Args:
            position: Position in recommendation list (1-10)
        """
        recommendations_clicked.labels(
            model_version=self.model_version,
            position=str(position)
        ).inc()

    def track_conversion(self, revenue: float = 0.0):
        """
        Track recommendation conversion.

        Args:
            revenue: Revenue generated (optional)
        """
        recommendations_converted.labels(
            model_version=self.model_version
        ).inc()

        if revenue > 0:
            recommendation_revenue.labels(
                model_version=self.model_version
            ).observe(revenue)

    def track_cache_hit(self, hit: bool):
        """Track cache hit/miss."""
        if hit:
            cache_hits.inc()
        else:
            cache_misses.inc()

    def update_diversity(self, score: float):
        """
        Update diversity score.

        Args:
            score: Diversity score between 0 and 1
        """
        recommendation_diversity.labels(
            model_version=self.model_version
        ).set(score)

    def update_coverage(self, coverage_pct: float):
        """
        Update catalog coverage.

        Args:
            coverage_pct: Percentage of catalog recommended (0-100)
        """
        catalog_coverage.labels(
            model_version=self.model_version
        ).set(coverage_pct)

    def update_model_staleness(self, hours: float):
        """
        Update model staleness.

        Args:
            hours: Hours since model was last updated
        """
        model_staleness_hours.labels(
            model_version=self.model_version
        ).set(hours)


def monitor_latency(metrics_collector: MetricsCollector):
    """Decorator to track function latency."""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            start_time = time.time()
            try:
                result = func(*args, **kwargs)
                metrics_collector.track_request(success=True)
                return result
            except Exception as e:
                metrics_collector.track_request(success=False)
                raise
            finally:
                latency = time.time() - start_time
                metrics_collector.track_latency(latency)
        return wrapper
    return decorator


# Example usage in FastAPI
from fastapi import FastAPI

app = FastAPI()
metrics = MetricsCollector(model_version='v2')


@app.post("/recommendations")
@monitor_latency(metrics)
async def get_recommendations(user_id: str, n: int = 10):
    """Generate recommendations with metrics tracking."""
    # Your recommendation logic here
    recommendations = generate_recs(user_id, n)

    return {'recommendations': recommendations}


@app.post("/track/click")
async def track_click(user_id: str, item_id: str, position: int):
    """Track recommendation click."""
    metrics.track_click(position)
    return {'status': 'tracked'}


@app.post("/track/conversion")
async def track_conversion(user_id: str, item_id: str, revenue: float):
    """Track recommendation conversion."""
    metrics.track_conversion(revenue)
    return {'status': 'tracked'}


@app.get("/metrics")
async def get_metrics():
    """Expose Prometheus metrics."""
    return Response(generate_latest(REGISTRY), media_type="text/plain")

Dashboard Monitoring

Metrics Dashboard Endpoint

from datetime import datetime, timedelta
from typing import Dict, List


class MetricsDashboard:
    """Aggregated metrics for dashboards."""

    def __init__(self, db_client):
        self.db = db_client

    def get_summary(
        self,
        start_date: datetime,
        end_date: datetime,
        model_version: str = 'v2'
    ) -> Dict:
        """
        Get aggregated metrics for time period.

        Args:
            start_date: Start of period
            end_date: End of period
            model_version: Model version to query

        Returns:
            Dictionary with aggregated metrics
        """
        # Query database for metrics in time range
        query = """
        SELECT
            COUNT(*) as total_requests,
            SUM(CASE WHEN clicked = 1 THEN 1 ELSE 0 END) as total_clicks,
            SUM(CASE WHEN converted = 1 THEN 1 ELSE 0 END) as total_conversions,
            SUM(revenue) as total_revenue,
            AVG(latency_ms) as avg_latency,
            PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY latency_ms) as p95_latency,
            PERCENTILE_CONT(0.99) WITHIN GROUP (ORDER BY latency_ms) as p99_latency
        FROM recommendation_events
        WHERE
            timestamp BETWEEN %s AND %s
            AND model_version = %s
        """

        result = self.db.execute(query, (start_date, end_date, model_version))
        row = result.fetchone()

        total_requests = row['total_requests']
        total_clicks = row['total_clicks']
        total_conversions = row['total_conversions']

        return {
            'period': {
                'start': start_date.isoformat(),
                'end': end_date.isoformat()
            },
            'requests': {
                'total': total_requests
            },
            'clicks': {
                'total': total_clicks,
                'ctr': total_clicks / total_requests if total_requests > 0 else 0
            },
            'conversions': {
                'total': total_conversions,
                'conversion_rate': total_conversions / total_clicks if total_clicks > 0 else 0
            },
            'revenue': {
                'total': float(row['total_revenue'] or 0),
                'per_user': float(row['total_revenue'] or 0) / total_requests if total_requests > 0 else 0
            },
            'latency': {
                'avg_ms': float(row['avg_latency'] or 0),
                'p95_ms': float(row['p95_latency'] or 0),
                'p99_ms': float(row['p99_latency'] or 0)
            }
        }

    def get_hourly_metrics(
        self,
        date: datetime,
        model_version: str = 'v2'
    ) -> List[Dict]:
        """
        Get metrics broken down by hour.

        Args:
            date: Date to query
            model_version: Model version

        Returns:
            List of hourly metric dictionaries
        """
        query = """
        SELECT
            DATE_TRUNC('hour', timestamp) as hour,
            COUNT(*) as requests,
            SUM(CASE WHEN clicked = 1 THEN 1 ELSE 0 END) as clicks,
            SUM(CASE WHEN converted = 1 THEN 1 ELSE 0 END) as conversions,
            AVG(latency_ms) as avg_latency
        FROM recommendation_events
        WHERE
            DATE(timestamp) = %s
            AND model_version = %s
        GROUP BY hour
        ORDER BY hour
        """

        results = self.db.execute(query, (date.date(), model_version))

        return [
            {
                'hour': row['hour'].isoformat(),
                'requests': row['requests'],
                'clicks': row['clicks'],
                'ctr': row['clicks'] / row['requests'] if row['requests'] > 0 else 0,
                'conversions': row['conversions'],
                'avg_latency_ms': float(row['avg_latency'] or 0)
            }
            for row in results
        ]


# FastAPI endpoints
@app.get("/dashboard/summary")
async def dashboard_summary(
    start_date: str,
    end_date: str,
    model_version: str = 'v2'
):
    """Get dashboard summary metrics."""
    dashboard = MetricsDashboard(db_client)

    summary = dashboard.get_summary(
        start_date=datetime.fromisoformat(start_date),
        end_date=datetime.fromisoformat(end_date),
        model_version=model_version
    )

    return summary


@app.get("/dashboard/hourly")
async def dashboard_hourly(date: str, model_version: str = 'v2'):
    """Get hourly metrics for date."""
    dashboard = MetricsDashboard(db_client)

    hourly = dashboard.get_hourly_metrics(
        date=datetime.fromisoformat(date),
        model_version=model_version
    )

    return {'metrics': hourly}

Alert Configuration

Alert Rules

from typing import Callable, Dict
import logging

logger = logging.getLogger(__name__)


class AlertRule:
    """Define alert threshold and action."""

    def __init__(
        self,
        name: str,
        metric_fn: Callable[[], float],
        threshold: float,
        comparison: str = '>',  # '>', '<', '>=', '<=', '=='
        severity: str = 'warning'  # 'info', 'warning', 'critical'
    ):
        self.name = name
        self.metric_fn = metric_fn
        self.threshold = threshold
        self.comparison = comparison
        self.severity = severity

    def check(self) -> bool:
        """
        Check if alert should trigger.

        Returns:
            True if alert condition met
        """
        current_value = self.metric_fn()

        if self.comparison == '>':
            triggered = current_value > self.threshold
        elif self.comparison == '<':
            triggered = current_value < self.threshold
        elif self.comparison == '>=':
            triggered = current_value >= self.threshold
        elif self.comparison == '<=':
            triggered = current_value <= self.threshold
        elif self.comparison == '==':
            triggered = current_value == self.threshold
        else:
            raise ValueError(f"Unknown comparison: {self.comparison}")

        if triggered:
            logger.warning(
                f"Alert triggered: {self.name} "
                f"(value={current_value:.4f}, threshold={self.threshold})"
            )

        return triggered


class AlertManager:
    """Manage alerts and notifications."""

    def __init__(self, slack_webhook: str = None):
        self.slack_webhook = slack_webhook
        self.rules = []

    def add_rule(self, rule: AlertRule):
        """Add alert rule."""
        self.rules.append(rule)
        logger.info(f"Added alert rule: {rule.name}")

    def check_all(self):
        """Check all alert rules."""
        triggered = []

        for rule in self.rules:
            if rule.check():
                triggered.append(rule)

        if triggered:
            self._send_alerts(triggered)

        return triggered

    def _send_alerts(self, rules: List[AlertRule]):
        """Send alerts for triggered rules."""
        if not self.slack_webhook:
            return

        # Group by severity
        critical = [r for r in rules if r.severity == 'critical']
        warnings = [r for r in rules if r.severity == 'warning']

        message = "🚨 *Recommendation System Alerts*\n\n"

        if critical:
            message += "*Critical:*\n"
            for rule in critical:
                value = rule.metric_fn()
                message += f"• {rule.name}: {value:.4f} (threshold: {rule.threshold})\n"
            message += "\n"

        if warnings:
            message += "*Warnings:*\n"
            for rule in warnings:
                value = rule.metric_fn()
                message += f"• {rule.name}: {value:.4f} (threshold: {rule.threshold})\n"

        # Send to Slack with timeout and error handling
        try:
            response = requests.post(
                self.slack_webhook,
                json={'text': message},
                timeout=5  # Prevent hanging
            )
            response.raise_for_status()  # Raise exception for 4xx/5xx
            logger.info(f"Successfully sent {len(rules)} alerts to Slack")
        except RequestException as e:
            logger.error(f"Failed to send Slack alert: {e}")
            # Don't re-raise - alert system should continue monitoring


# Example alert configuration
alert_manager = AlertManager(slack_webhook=os.getenv('SLACK_WEBHOOK'))

# Alert: CTR drops below 5%
alert_manager.add_rule(AlertRule(
    name='Low CTR',
    metric_fn=lambda: get_current_ctr(),  # Function returning current CTR
    threshold=0.05,
    comparison='<',
    severity='warning'
))

# Alert: P95 latency exceeds 500ms
alert_manager.add_rule(AlertRule(
    name='High Latency',
    metric_fn=lambda: get_p95_latency_ms(),
    threshold=500,
    comparison='>',
    severity='critical'
))

# Alert: Error rate above 1%
alert_manager.add_rule(AlertRule(
    name='High Error Rate',
    metric_fn=lambda: get_error_rate(),
    threshold=0.01,
    comparison='>',
    severity='critical'
))

# Check alerts every 5 minutes
import schedule

schedule.every(5).minutes.do(alert_manager.check_all)

while True:
    schedule.run_pending()
    time.sleep(60)

Quality Monitoring

Diversity and Coverage Tracking

import numpy as np
from collections import Counter


class QualityMonitor:
    """Monitor recommendation quality metrics."""

    def __init__(self):
        self.recommendations_history = []

    def track_recommendations(
        self,
        user_id: str,
        item_ids: List[str],
        categories: List[str]
    ):
        """Track recommendations for quality analysis."""
        self.recommendations_history.append({
            'user_id': user_id,
            'item_ids': item_ids,
            'categories': categories
        })

    def calculate_diversity(self) -> float:
        """
        Calculate diversity score (0-1).

        Higher = more diverse recommendations.
        Uses entropy of category distribution.
        """
        all_categories = []

        for rec in self.recommendations_history[-1000:]:  # Last 1000
            all_categories.extend(rec['categories'])

        if not all_categories:
            return 0.0

        # Calculate category distribution
        category_counts = Counter(all_categories)
        total = len(all_categories)

        # Entropy
        entropy = 0
        for count in category_counts.values():
            p = count / total
            entropy -= p * np.log2(p)

        # Normalize (max entropy = log2(n_categories))
        max_entropy = np.log2(len(category_counts))
        diversity = entropy / max_entropy if max_entropy > 0 else 0

        return diversity

    def calculate_coverage(self, catalog_size: int) -> float:
        """
        Calculate catalog coverage (0-100).

        % of catalog items recommended.
        """
        recommended_items = set()

        for rec in self.recommendations_history[-10000:]:  # Last 10k
            recommended_items.update(rec['item_ids'])

        coverage = len(recommended_items) / catalog_size * 100

        return coverage


# Track quality metrics in Prometheus
quality_monitor = QualityMonitor()

def update_quality_metrics():
    """Update quality metrics periodically."""
    diversity = quality_monitor.calculate_diversity()
    coverage = quality_monitor.calculate_coverage(catalog_size=100000)

    # Update Prometheus gauges
    recommendation_diversity.labels(model_version='v2').set(diversity)
    catalog_coverage.labels(model_version='v2').set(coverage)

    logger.info(f"Quality metrics: diversity={diversity:.3f}, coverage={coverage:.1f}%")


# Run every hour
schedule.every().hour.do(update_quality_metrics)

Best Practices

1. Monitor Business Metrics: CTR, conversion rate, revenue (not just technical) 2. Set Baseline Alerts: Alert when metrics deviate significantly from baseline 3. Track Latency Percentiles: P95/P99 more important than average 4. Log Everything: Comprehensive logging enables debugging 5. Dashboard for Stakeholders: Non-technical metrics (CTR, revenue) 6. A/B Test Monitoring: Compare variants in real-time 7. Quality Metrics: Track diversity and coverage, not just accuracy 8. Model Staleness: Alert if model hasn't been updated recently

Production Recommendation System Architecture

Complete architecture patterns for scalable, production-ready recommendation systems with feature stores, model serving, and monitoring.

Overview

A production recommendation system requires:

• Feature Store: Centralized feature management
• Model Serving: Low-latency predictions
• Caching Layer: Sub-millisecond responses
• A/B Testing: Continuous experimentation
• Monitoring: Track quality and performance

Complete Recommendation Service

from typing import List, Dict, Optional
import numpy as np
from dataclasses import dataclass
import logging
from datetime import datetime, timedelta
import hashlib
import json

logger = logging.getLogger(__name__)


@dataclass
class Recommendation:
    """Single recommendation result."""
    item_id: str
    score: float
    reason: str
    metadata: Dict


class FeatureStore:
    """
    Centralized feature storage for users and items.

    Provides low-latency access to precomputed features.
    """

    def __init__(self, redis_client=None, db_client=None):
        self.redis = redis_client  # Fast cache
        self.db = db_client  # Persistent storage
        self.cache_ttl = 3600  # 1 hour

    def get_user_features(self, user_id: str) -> Dict:
        """
        Get user features with caching.

        Returns:
            Dictionary of user features
        """
        # Try cache first
        cache_key = f"user_features:{user_id}"

        if self.redis:
            cached = self.redis.get(cache_key)
            if cached:
                logger.debug(f"Cache hit for user {user_id}")
                return json.loads(cached)

        # Fetch from database
        features = self._fetch_user_features_from_db(user_id)

        # Cache for next time
        if self.redis and features:
            self.redis.setex(
                cache_key,
                self.cache_ttl,
                json.dumps(features)
            )

        return features

    def get_item_features(self, item_id: str) -> Dict:
        """Get item features with caching."""
        cache_key = f"item_features:{item_id}"

        if self.redis:
            cached = self.redis.get(cache_key)
            if cached:
                return json.loads(cached)

        features = self._fetch_item_features_from_db(item_id)

        if self.redis and features:
            self.redis.setex(
                cache_key,
                self.cache_ttl,
                json.dumps(features)
            )

        return features

    def get_batch_item_features(self, item_ids: List[str]) -> Dict[str, Dict]:
        """
        Efficiently fetch features for multiple items.

        Uses Redis MGET for batching.
        """
        if not self.redis:
            return {
                item_id: self.get_item_features(item_id)
                for item_id in item_ids
            }

        # Batch fetch from cache
        cache_keys = [f"item_features:{item_id}" for item_id in item_ids]
        cached_values = self.redis.mget(cache_keys)

        results = {}
        missing_ids = []

        for item_id, cached in zip(item_ids, cached_values):
            if cached:
                results[item_id] = json.loads(cached)
            else:
                missing_ids.append(item_id)

        # Fetch missing from database
        if missing_ids:
            missing_features = self._fetch_batch_items_from_db(missing_ids)

            # Cache missing features
            pipe = self.redis.pipeline()
            for item_id, features in missing_features.items():
                cache_key = f"item_features:{item_id}"
                pipe.setex(
                    cache_key,
                    self.cache_ttl,
                    json.dumps(features)
                )
                results[item_id] = features
            pipe.execute()

        return results

    def _fetch_user_features_from_db(self, user_id: str) -> Dict:
        """Fetch user features from database."""
        if not self.db:
            # Fallback to defaults
            return {
                'user_id': user_id,
                'total_purchases': 0,
                'avg_rating': 0.0,
                'favorite_categories': []
            }

        # Actual database query
        query = """
        SELECT
            total_purchases,
            avg_rating,
            favorite_categories,
            recent_views,
            created_at
        FROM user_features
        WHERE user_id = %s
        """

        result = self.db.execute(query, (user_id,))
        if not result:
            return {}

        row = result.fetchone()
        if not row:
            return {}  # No matching user found

        return dict(row)

    def _fetch_item_features_from_db(self, item_id: str) -> Dict:
        """Fetch item features from database."""
        if not self.db:
            return {
                'item_id': item_id,
                'category': 'unknown',
                'price': 0.0,
                'popularity': 0.0
            }

        query = """
        SELECT
            category,
            price,
            popularity_score,
            avg_rating,
            num_ratings,
            tags
        FROM item_features
        WHERE item_id = %s
        """

        result = self.db.execute(query, (item_id,))
        if not result:
            return {}

        row = result.fetchone()
        if not row:
            return {}  # No matching item found

        return dict(row)

    def _fetch_batch_items_from_db(self, item_ids: List[str]) -> Dict[str, Dict]:
        """Fetch multiple items efficiently."""
        if not self.db:
            return {}

        placeholders = ','.join(['%s'] * len(item_ids))
        query = f"""
        SELECT
            item_id,
            category,
            price,
            popularity_score,
            avg_rating,
            tags
        FROM item_features
        WHERE item_id IN ({placeholders})
        """

        results = self.db.execute_many(query, item_ids)
        return {row['item_id']: dict(row) for row in results}


class ModelServing:
    """
    Model serving layer for recommendations.

    Supports multiple models and A/B testing.
    """

    def __init__(self):
        self.models = {}
        self.default_model = None

    def register_model(
        self,
        name: str,
        model,
        is_default: bool = False
    ):
        """Register a model for serving."""
        self.models[name] = model

        if is_default or not self.default_model:
            self.default_model = name

        logger.info(f"Registered model: {name}")

    def predict(
        self,
        user_features: Dict,
        item_features: List[Dict],
        model_name: Optional[str] = None
    ) -> np.ndarray:
        """
        Generate predictions using specified model.

        Args:
            user_features: User feature dictionary
            item_features: List of item feature dictionaries
            model_name: Model to use (defaults to primary)

        Returns:
            Array of scores for each item
        """
        model_name = model_name or self.default_model

        if model_name not in self.models:
            raise ValueError(f"Model {model_name} not found")

        model = self.models[model_name]

        # Transform features to model input format
        X = self._prepare_features(user_features, item_features)

        # Predict
        scores = model.predict(X)

        return scores

    def _prepare_features(
        self,
        user_features: Dict,
        item_features: List[Dict]
    ) -> np.ndarray:
        """
        Prepare features for model input.

        Combines user and item features into format expected by model.
        """
        # This depends on your model's feature format
        # Example: concatenate user features with each item
        n_items = len(item_features)

        # Extract user vector
        user_vec = np.array([
            user_features.get('total_purchases', 0),
            user_features.get('avg_rating', 0.0),
            len(user_features.get('favorite_categories', []))
        ])

        # Extract item vectors
        item_vecs = np.array([
            [
                item.get('popularity_score', 0.0),
                item.get('price', 0.0),
                item.get('avg_rating', 0.0)
            ]
            for item in item_features
        ])

        # Combine: repeat user vector for each item
        user_repeated = np.tile(user_vec, (n_items, 1))
        features = np.hstack([user_repeated, item_vecs])

        return features


class RecommendationService:
    """
    Complete recommendation service.

    Orchestrates feature fetching, scoring, ranking, and filtering.
    """

    def __init__(
        self,
        feature_store: FeatureStore,
        model_serving: ModelServing,
        cache=None
    ):
        self.feature_store = feature_store
        self.model_serving = model_serving
        self.cache = cache
        self.cache_ttl = 300  # 5 minutes

    def get_recommendations(
        self,
        user_id: str,
        n: int = 10,
        filters: Optional[Dict] = None,
        model_name: Optional[str] = None
    ) -> List[Recommendation]:
        """
        Generate personalized recommendations.

        Args:
            user_id: User to recommend for
            n: Number of recommendations
            filters: Optional filters (category, price range, etc.)
            model_name: Model variant to use

        Returns:
            List of Recommendation objects
        """
        # Check cache
        cache_key = f"recs:{user_id}:{n}:{model_name}"

        if self.cache:
            cached = self.cache.get(cache_key)
            if cached:
                logger.info(f"Cache hit for user {user_id}")
                return [
                    Recommendation(**rec)
                    for rec in json.loads(cached)
                ]

        # Get user features
        user_features = self.feature_store.get_user_features(user_id)

        # Get candidate items
        candidates = self._get_candidates(user_id, filters)

        if not candidates:
            logger.warning(f"No candidates for user {user_id}")
            return []

        # Get item features (batch)
        item_features = self.feature_store.get_batch_item_features(candidates)

        # Prepare features for scoring
        item_feature_list = [
            item_features.get(item_id, {})
            for item_id in candidates
        ]

        # Score candidates
        scores = self.model_serving.predict(
            user_features,
            item_feature_list,
            model_name
        )

        # Rank and select top N
        recommendations = self._rank_and_diversify(
            candidates,
            scores,
            item_features,
            n
        )

        # Cache result
        if self.cache:
            serialized = [
                {
                    'item_id': rec.item_id,
                    'score': rec.score,
                    'reason': rec.reason,
                    'metadata': rec.metadata
                }
                for rec in recommendations
            ]
            self.cache.setex(
                cache_key,
                self.cache_ttl,
                json.dumps(serialized)
            )

        return recommendations

    def _get_candidates(
        self,
        user_id: str,
        filters: Optional[Dict] = None
    ) -> List[str]:
        """
        Get candidate items to score.

        Uses multiple strategies:
        1. Collaborative filtering candidates
        2. Popular items in user's categories
        3. New items (serendipity)
        """
        candidates = set()

        # Strategy 1: Similar users' purchases
        # (Simplified - in production, use precomputed similar users)
        similar_user_items = self._get_similar_user_items(user_id, limit=50)
        candidates.update(similar_user_items)

        # Strategy 2: Popular in favorite categories
        user_features = self.feature_store.get_user_features(user_id)
        for category in user_features.get('favorite_categories', [])[:3]:
            popular = self._get_popular_in_category(category, limit=20)
            candidates.update(popular)

        # Strategy 3: Recently added (exploration)
        new_items = self._get_new_items(limit=10)
        candidates.update(new_items)

        # Apply filters
        if filters:
            candidates = self._apply_filters(candidates, filters)

        # Remove already purchased
        purchased = set(user_features.get('purchased_items', []))
        candidates = candidates - purchased

        return list(candidates)

    def _rank_and_diversify(
        self,
        candidates: List[str],
        scores: np.ndarray,
        item_features: Dict[str, Dict],
        n: int
    ) -> List[Recommendation]:
        """
        Rank candidates and ensure diversity.

        Uses MMR (Maximal Marginal Relevance) for diversity.
        """
        # Sort by score
        sorted_indices = np.argsort(scores)[::-1]

        # Select top N with diversity
        selected = []
        selected_categories = set()

        for idx in sorted_indices:
            if len(selected) >= n:
                break

            item_id = candidates[idx]
            score = float(scores[idx])
            features = item_features.get(item_id, {})
            category = features.get('category', 'unknown')

            # Diversity: limit items per category
            if selected_categories.count(category) >= 3:
                continue

            selected.append(
                Recommendation(
                    item_id=item_id,
                    score=score,
                    reason=self._generate_reason(features),
                    metadata=features
                )
            )
            selected_categories.add(category)

        return selected

    def _generate_reason(self, features: Dict) -> str:
        """Generate human-readable reason for recommendation."""
        category = features.get('category', 'item')
        avg_rating = features.get('avg_rating', 0.0)

        if avg_rating >= 4.5:
            return f"Highly rated {category}"
        elif features.get('popularity_score', 0) > 0.8:
            return f"Popular {category}"
        else:
            return f"Recommended {category}"

    def _get_similar_user_items(self, user_id: str, limit: int) -> List[str]:
        """Get items purchased by similar users."""
        # Simplified - in production, precompute similar users
        return []

    def _get_popular_in_category(self, category: str, limit: int) -> List[str]:
        """Get popular items in category."""
        # Query from database or cache
        return []

    def _get_new_items(self, limit: int) -> List[str]:
        """Get recently added items."""
        return []

    def _apply_filters(
        self,
        candidates: List[str],
        filters: Dict
    ) -> List[str]:
        """Apply user-specified filters."""
        # Filter by price, category, etc.
        return candidates


## Deployment Pattern

from fastapi import FastAPI, HTTPException, Depends from pydantic import BaseModel, Field from typing import List, Optional import redis import psycopg2

app = FastAPI()

Initialize components

import os

redis_client = redis.Redis( host=os.getenv('REDIS_HOST', 'localhost'), port=int(os.getenv('REDIS_PORT', 6379)), decode_responses=True )

db_conn = psycopg2.connect( host=os.getenv('DB_HOST', 'localhost'), database=os.getenv('DB_NAME', 'recdb'), user=os.getenv('DB_USER', 'user'), password=os.getenv('DB_PASSWORD'), port=int(os.getenv('DB_PORT', 5432)) )

feature_store = FeatureStore(redis_client=redis_client, db_client=db_conn) model_serving = ModelServing()

Load models

import joblib main_model = joblib.load('/models/main_recommender.pkl') model_serving.register_model('main', main_model, is_default=True)

Initialize service

rec_service = RecommendationService( feature_store=feature_store, model_serving=model_serving, cache=redis_client )

class RecommendationRequest(BaseModel): user_id: str = Field(..., description="User ID") n: int = Field(10, ge=1, le=100, description="Number of recommendations") filters: Optional[Dict] = Field(None, description="Optional filters") model_name: Optional[str] = Field(None, description="Model variant")

class RecommendationResponse(BaseModel): user_id: str recommendations: List[Dict] model_used: str cached: bool

@app.post("/recommendations", response_model=RecommendationResponse) async def get_recommendations(request: RecommendationRequest): """Generate personalized recommendations.""" try: recommendations = rec_service.get_recommendations( user_id=request.user_id, n=request.n, filters=request.filters, model_name=request.model_name )

return RecommendationResponse( user_id=request.user_id, recommendations=[ { 'item_id': rec.item_id, 'score': rec.score, 'reason': rec.reason } for rec in recommendations ], model_used=request.model_name or 'main', cached=False # Would check if from cache )

except Exception as e: logger.error(f"Recommendation failed for {request.user_id}: {e}") raise HTTPException(500, f"Recommendation failed: {str(e)}")

@app.get("/health") async def health(): """Health check endpoint.""" return {"status": "healthy", "models": list(model_serving.models.keys())}

## Best Practices

1. **Feature Precomputation**: Compute features offline, serve online
2. **Batch Fetching**: Use Redis MGET for multiple items
3. **Cache Aggressively**: Cache user recommendations (5-15 min TTL)
4. **Fail Gracefully**: Return popular items if personalization fails
5. **Monitor Latency**: Track P95/P99 latency, optimize slow paths
6. **Version Models**: Support multiple model versions for A/B testing
7. **Diversity**: Ensure recommendations aren't too similar
8. **Explanation**: Provide reasons for recommendations