Внедрение и оценка качества рекомендательной системы

Введение

Вы построили модель. Но будет ли она работать в реальности? Как узнать, что рекомендации действительно увеличивают продажи? В этом уроке мы разберём, как оценить качество рекомендательной системы до и после внедрения.

Две категории метрик

Офлайн-метрики (без реальных пользователей)
    ↓
Оцениваем на исторических данных
    ↓
[Precision@K, Recall@K, MAP@K, NDCG@K]
    ↓
Онлайн-метрики (с реальными пользователями)
    ↓
A/B тестирование
    ↓
[CTR, CR, AOV, GMV, Retention]

Офлайн-метрики: как оценить до запуска

1. Precision@K

Доля релевантных товаров среди топ-K рекомендаций.

def precision_at_k(actual, predicted, k=5):
    """
    actual: список реально купленных/понравившихся товаров
    predicted: список рекомендованных товаров
    k: сколько топ-рекомендаций рассматриваем
    """
    predicted_k = predicted[:k]
    relevant = set(actual) & set(predicted_k)
    return len(relevant) / k

# Пример
actual = ['A', 'B', 'C', 'D']
predicted = ['A', 'E', 'B', 'F', 'C']

print(f"Precision@3: {precision_at_k(actual, predicted, k=3):.2f}")
# Precision@3: 2/3 = 0.67 (A и B релевантны, E нет)

2. Recall@K

Доля всех релевантных товаров, попавших в рекомендации.

def recall_at_k(actual, predicted, k=5):
    predicted_k = predicted[:k]
    relevant = set(actual) & set(predicted_k)
    return len(relevant) / len(actual) if len(actual) > 0 else 0

print(f"Recall@3: {recall_at_k(actual, predicted, k=3):.2f}")
# Recall@3: 2/4 = 0.50 (из 4 релевантных нашли только 2)

3. MAP@K (Mean Average Precision)

Учитывает порядок рекомендаций.

def average_precision_at_k(actual, predicted, k=5):
    predicted_k = predicted[:k]
    score = 0.0
    num_hits = 0.0
    
    for i, p in enumerate(predicted_k):
        if p in actual:
            num_hits += 1.0
            score += num_hits / (i + 1.0)
    
    return score / min(len(actual), k) if len(actual) > 0 else 0

def map_at_k(actual_list, predicted_list, k=5):
    """Среднее по всем пользователям"""
    return np.mean([average_precision_at_k(a, p, k) 
                    for a, p in zip(actual_list, predicted_list)])

# Пример для одного пользователя
actual = ['A', 'B', 'C']
predicted = ['A', 'C', 'D', 'B', 'E']

print(f"AP@3: {average_precision_at_k(actual, predicted, k=3):.2f}")
# AP@3: (1/1 + 2/3) / 3 = (1 + 0.67) / 3 = 0.56

4. NDCG@K (Normalized Discounted Cumulative Gain)

Учитывает не только релевантность, но и вес релевантности (например, оценку 5 важнее, чем оценку 3).

def dcg_at_k(scores, k=5):
    """Discounted Cumulative Gain"""
    scores_k = scores[:k]
    return sum((2**rel - 1) / np.log2(i + 2) for i, rel in enumerate(scores_k))

def ndcg_at_k(actual_scores, predicted, k=5):
    """
    actual_scores: словарь {товар: релевантность}
    predicted: список рекомендованных товаров
    """
    # Идеальный порядок (по убыванию релевантности)
    ideal_scores = sorted(actual_scores.values(), reverse=True)
    
    # Фактические оценки в порядке рекомендаций
    actual_scores_ordered = [actual_scores.get(p, 0) for p in predicted[:k]]
    
    dcg = dcg_at_k(actual_scores_ordered, k)
    idcg = dcg_at_k(ideal_scores, k)
    
    return dcg / idcg if idcg > 0 else 0

# Пример
actual_scores = {'A': 5, 'B': 4, 'C': 3, 'D': 2, 'E': 1}
predicted = ['A', 'C', 'B', 'D', 'E']

print(f"NDCG@3: {ndcg_at_k(actual_scores, predicted, k=3):.2f}")

Онлайн-метрики: A/B тестирование

Даже если офлайн-метрики хороши, только A/B тест покажет реальный эффект на бизнес.

Схема A/B теста:

Все пользователи
       |
   ┌───┴───┐
   ↓       ↓
Группа A   Группа B
(контроль)  (эксперимент)
Без реком.  С рекомендациями
   ↓           ↓
Сравниваем метрики

Что измерять в A/B тесте:

Пример результатов A/B теста:

Вывод: рекомендации увеличивают выручку на 46%, A/B тест успешен.

Внедрение в продакшен: создание API

from flask import Flask, request, jsonify
import pickle
import pandas as pd

app = Flask(__name__)

# Загружаем обученную модель
with open('svd_model.pkl', 'rb') as f:
    model = pickle.load(f)

# Загружаем список всех товаров
all_items = pd.read_csv('products.csv')['product_id'].tolist()

# Кэш популярных рекомендаций (для холодного старта)
popular_items = pd.read_csv('popular_items.csv')['product_id'].tolist()[:10]

@app.route('/recommend', methods=['GET'])
def recommend():
    user_id = request.args.get('user_id', type=int)
    top_n = request.args.get('top_n', 5, type=int)
    
    # Получаем историю пользователя (из базы данных)
    user_history = get_user_history(user_id)
    
    # Если нет истории → холодный старт
    if not user_history:
        recommendations = popular_items[:top_n]
        source = 'popular'
    else:
        # Используем SVD модель
        predictions = []
        for item in all_items:
            if item not in user_history:
                pred = model.predict(user_id, item).est
                predictions.append((item, pred))
        
        predictions.sort(key=lambda x: x[1], reverse=True)
        recommendations = [item for item, _ in predictions[:top_n]]
        source = 'personalized'
    
    # Логируем для аналитики
    log_recommendation(user_id, recommendations, source)
    
    return jsonify({
        'user_id': user_id,
        'recommendations': recommendations,
        'source': source
    })

@app.route('/event', methods=['POST'])
def track_event():
    """Отслеживание кликов и покупок"""
    data = request.json
    user_id = data['user_id']
    item_id = data['item_id']
    event_type = data['event_type']  # 'click', 'purchase'
    
    # Сохраняем в базу данных для будущего обучения
    save_event(user_id, item_id, event_type)
    
    return jsonify({'status': 'ok'})

if __name__ == '__main__':
    app.run(port=5000)

Архитектура продакшен-системы

┌─────────────────────────────────────────────────────────────┐
│                      ПОЛЬЗОВАТЕЛЬ                            │
└─────────────────────────────────────────────────────────────┘
                              │
                              ↓
┌─────────────────────────────────────────────────────────────┐
│                   API (Flask/FastAPI)                        │
│              GET /recommend?user_id=123                      │
└─────────────────────────────────────────────────────────────┘
                              │
            ┌─────────────────┼─────────────────┐
            ↓                 ↓                 ↓
    ┌──────────────┐  ┌──────────────┐  ┌──────────────┐
    │   Redis      │  │  PostgreSQL  │  │   S3/Модель  │
    │   (кэш)      │  │ (история)    │  │   (веса)     │
    └──────────────┘  └──────────────┘  └──────────────┘
                              │
                              ↓
┌─────────────────────────────────────────────────────────────┐
│                   Batch job (ежедневно)                      │
│            Переобучение модели на новых данных               │
└─────────────────────────────────────────────────────────────┘

Мониторинг в продакшене

Что отслеживать:

Пример дашборда мониторинга (Prometheus + Grafana):

from prometheus_client import Counter, Histogram, Gauge

# Метрики
recommendation_requests = Counter('recommendation_requests_total', 'Total requests')
recommendation_latency = Histogram('recommendation_latency_seconds', 'Request latency')
model_confidence = Gauge('model_confidence', 'Average prediction confidence')

# Пример использования
@recommendation_latency.time()
def get_recommendations(user_id):
    recommendation_requests.inc()
    # ... логика ...
    model_confidence.set(0.85)
    return recommendations

Чеклист внедрения рекомендательной системы

Этап 1. Подготовка (1–2 дня)

• Собрать историю действий пользователей

• Очистить данные от аномалий

• Выбрать базовую модель (item-item или популярные)

Этап 2. Разработка (3–5 дней)

• Реализовать модель на Python

• Вычислить офлайн-метрики (Precision@K, Recall@K)

• Сравнить несколько подходов

Этап 3. Тестирование (2–3 дня)

• Запустить A/B тест на 5–10% трафика

• Собрать статистику за 3–7 дней

• Проверить статистическую значимость

Этап 4. Продакшен (2–3 дня)

• Создать API для рекомендаций

• Настроить кэширование (Redis)

• Добавить мониторинг и алерты

Этап 5. Итерации (постоянно)

• Собирать обратную связь (клики, покупки)

• Ежедневно переобучать модель

• Экспериментировать с новыми алгоритмами

Резюме урока и всего курса

Поздравляю! Вы прошли полный курс по рекомендательным системам.