ML: Построить рекомендательную систему

Решение

1. Необходимые данные

Для эффективной рекомендательной системы нужны:

Базовые данные:

• История покупок — user_id, product_id, timestamp, purchase_amount
• Взаимодействия пользователей — просмотры, клики, добавления в корзину, оценки/лайки
• Характеристики товаров — категория, описание, цена, теги, изображения
• Характеристики пользователей — возраст, пол, локация, предпочтения
• Контекст — сезонность, праздники, тренды

Метаданные:

• Временные метки для анализа поведения
• Информация о возвратах и жалобах
• Данные о конкурентах (цены, наличие)

2. Подходы к рекомендациям

2.1 Collaborative Filtering (Фильтрация по сотрудничеству)

Основана на предположении: если пользователи A и B похоже покупали раньше, им понравятся одни и те же товары.

from sklearn.metrics.pairwise import cosine_similarity
import numpy as np

# Матрица user-item (кто что купил)
user_item_matrix = np.array([
    [5, 0, 3, 0],  # user 1: купил товар 0 (оценка 5), товар 2 (оценка 3)
    [4, 0, 0, 2],  # user 2
    [0, 5, 5, 0],  # user 3
])

# Similarity между пользователями
user_similarity = cosine_similarity(user_item_matrix)
print("Similarity между пользователями:")
print(user_similarity)

# Рекомендация: товары от похожих пользователей
def get_collaborative_recommendation(user_id, user_item_matrix, similarity_matrix, n_recommendations=3):
    similar_users = np.argsort(similarity_matrix[user_id])[::-1][1:]  # топ похожих
    recommended_items = {}
    
    for similar_user in similar_users:
        items = np.where(user_item_matrix[similar_user] > 0)[0]  # товары похожего пользователя
        for item in items:
            if user_item_matrix[user_id, item] == 0:  # ещё не купил
                recommended_items[item] = recommended_items.get(item, 0) + similarity_matrix[user_id, similar_user]
    
    top_recommendations = sorted(recommended_items.items(), key=lambda x: x[1], reverse=True)[:n_recommendations]
    return [item[0] for item in top_recommendations]

print(f"\nРекомендации для user 0: {get_collaborative_recommendation(0, user_item_matrix, user_similarity)}")

Плюсы: Учитывает вкусы реальных пользователей, работает для новых товаров Минусы: Cold start проблема (новые пользователи/товары), разреженность матрицы

2.2 Content-Based Filtering (Фильтрация по содержанию)

Рекомендует товары похожие на те, которые пользователь уже оценил.

from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity

# Описания товаров
product_descriptions = [
    "красная рубашка хлопок спорт",
    "синяя рубашка хлопок офис",
    "чёрные кроссовки спорт кожа",
    "белые кроссовки спорт ткань",
]

# Векторизация описаний
vectorizer = TfidfVectorizer(analyzer='char', ngram_range=(2, 3))
product_vectors = vectorizer.fit_transform(product_descriptions)

# Похожесть между товарами
product_similarity = cosine_similarity(product_vectors)
print("Матрица похожести товаров:")
print(product_similarity)

# Рекомендация на основе товаров, которые нравились
def get_content_based_recommendation(user_purchased, product_similarity, n_recommendations=2):
    scores = np.zeros(product_similarity.shape[0])
    for product_id in user_purchased:
        scores += product_similarity[product_id]
    
    # Исключаем уже купленные
    for product_id in user_purchased:
        scores[product_id] = -1
    
    top_recommendations = np.argsort(scores)[::-1][:n_recommendations]
    return top_recommendations.tolist()

user_liked = [0, 1]  # пользователь купил товары 0 и 1
print(f"\nРекомендации на основе контента: {get_content_based_recommendation(user_liked, product_similarity)}")

Плюсы: Решает cold start, не нужны оценки других пользователей Минусы: Не открывает новое, ограничено качеством признаков

2.3 Hybrid подход (Гибридный)

Комбинирует collaborative и content-based:

def hybrid_recommendation(user_id, user_item_matrix, similarity_matrix, product_similarity, alpha=0.5, n_recommendations=3):
    # Collaborative score
    collab_items = get_collaborative_recommendation(user_id, user_item_matrix, similarity_matrix, n_recommendations=10)
    
    # Content-based score
    user_purchased = np.where(user_item_matrix[user_id] > 0)[0]
    content_items = get_content_based_recommendation(user_purchased, product_similarity, n_recommendations=10)
    
    # Комбинирование: берём пересечение + взвешенный скор
    combined_items = {}
    for item in collab_items:
        combined_items[item] = combined_items.get(item, 0) + alpha
    for item in content_items:
        combined_items[item] = combined_items.get(item, 0) + (1 - alpha)
    
    top = sorted(combined_items.items(), key=lambda x: x[1], reverse=True)[:n_recommendations]
    return [item[0] for item in top]

print(f"\nГибридные рекомендации: {hybrid_recommendation(0, user_item_matrix, user_similarity, product_similarity)}")

3. Оценка качества рекомендаций

Метрики для offline оценки:

from sklearn.metrics import mean_squared_error, mean_absolute_error

# Precision@K: доля релевантных в топ-K
def precision_at_k(recommendations, actual_purchases, k=5):
    rec_k = set(recommendations[:k])
    actual = set(actual_purchases)
    return len(rec_k.intersection(actual)) / k if k > 0 else 0

# Recall@K: доля найденных релевантных из всех
def recall_at_k(recommendations, actual_purchases, k=5):
    rec_k = set(recommendations[:k])
    actual = set(actual_purchases)
    return len(rec_k.intersection(actual)) / len(actual) if len(actual) > 0 else 0

# NDCG: Normalized Discounted Cumulative Gain
def ndcg_at_k(recommendations, actual_purchases, k=5):
    rec_k = recommendations[:k]
    gains = [1.0 if item in actual_purchases else 0.0 for item in rec_k]
    dcg = sum(gain / np.log2(i + 2) for i, gain in enumerate(gains))
    
    # Идеальный DCG
    ideal_gains = [1.0] * min(len(actual_purchases), k)
    idcg = sum(gain / np.log2(i + 2) for i, gain in enumerate(ideal_gains))
    
    return dcg / idcg if idcg > 0 else 0

# MAE/RMSE: для предсказания рейтингов
actual_ratings = [4, 3, 5]
predicted_ratings = [4.1, 2.8, 5.2]

print(f"MAE: {mean_absolute_error(actual_ratings, predicted_ratings):.3f}")
print(f"RMSE: {np.sqrt(mean_squared_error(actual_ratings, predicted_ratings)):.3f}")

# Тестовый пример
rec = [1, 3, 2, 0]
actual = [1, 2]
print(f"\nPrecision@5: {precision_at_k(rec, actual, 5):.2f}")
print(f"Recall@5: {recall_at_k(rec, actual, 5):.2f}")
print(f"NDCG@5: {ndcg_at_k(rec, actual, 5):.2f}")

Online метрики (A/B тестирование):

• CTR (Click-Through Rate) — доля кликов на рекомендации
• Conversion Rate — доля покупок от кликов
• Revenue per User — выручка на пользователя
• User Engagement — время на сайте, возвраты

4. Развертывание в Production

4.1 Архитектура системы

┌─────────────────┐
│  User Behavior  │  (история, взаимодействия)
└────────┬────────┘
         │
         ▼
┌──────────────────────┐
│  Feature Engineering │  (подготовка данных)
└────────┬─────────────┘
         │
         ▼
┌──────────────────────┐     ┌─────────────┐
│  Training Pipeline   │────▶│  Model      │
│  (weekly/monthly)    │     │  Registry   │
└────────┬─────────────┘     └─────────────┘
         │
         ▼
┌──────────────────────┐
│   API Service        │  (FastAPI/Flask)
│   (realtime reqs)    │
└────────┬─────────────┘
         │
         ▼
┌──────────────────────┐
│   Redis Cache        │  (кеширование рекомендаций)
└────────┬─────────────┘
         │
         ▼
┌──────────────────────┐
│   Frontend           │  (отображение)
└──────────────────────┘

4.2 Python код для Production API

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import redis
import json
import joblib
import logging

app = FastAPI()
redis_client = redis.Redis(host='localhost', port=6379, decode_responses=True)
logger = logging.getLogger(__name__)

# Загрузка обученной модели
model = joblib.load('recommendation_model.pkl')

class RecommendationRequest(BaseModel):
    user_id: int
    n_recommendations: int = 5

class RecommendationResponse(BaseModel):
    user_id: int
    products: list[int]
    scores: list[float]

@app.post('/recommend')
async def get_recommendations(request: RecommendationRequest):
    try:
        # Проверка кеша
        cache_key = f'recommendations:{request.user_id}'
        cached = redis_client.get(cache_key)
        if cached:
            return json.loads(cached)
        
        # Получение рекомендаций
        products, scores = model.predict(request.user_id, request.n_recommendations)
        
        response = RecommendationResponse(
            user_id=request.user_id,
            products=products,
            scores=scores
        )
        
        # Кеширование на 1 час
        redis_client.setex(cache_key, 3600, response.json())
        
        logger.info(f'Generated recommendations for user {request.user_id}')
        return response
        
    except Exception as e:
        logger.error(f'Error: {str(e)}')
        raise HTTPException(status_code=500, detail='Internal server error')

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

4.3 Развертывание (Docker + Kubernetes)

FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

4.4 Мониторинг и обновления

• Логирование: все рекомендации и клики для анализа
• Метрики: Prometheus для отслеживания performance
• Переобучение: weekly/monthly переобучение с новыми данными
• A/B тестирование: параллельные версии моделей
• Fallback: если модель недоступна, показываем популярные товары

Итоговая рекомендация

Для начальной реализации:

1. Начните с collaborative filtering (быстро и эффективно)
2. Добавьте content-based для новых товаров/пользователей
3. Используйте hybrid подход для финальной версии
4. Регулярно А/Б тестируйте и отслеживайте метрики
5. Развивайте через deep learning (Embeddings, LSTM) при необходимости