Hybrid Recommendation System Development

Реализация гибридной рекомендательной системы

Гибридные системы объединяют collaborative filtering, content-based и другие сигналы. Ни один подход не работает хорошо во всех ситуациях: CF плохо справляется с холодным стартом, CB ограничен метаданными, popularity-based не персонализирует. Гибридизация решает несколько проблем одновременно и улучшает метрики на 15-30% против лучшего одиночного метода.

Архитектуры гибридизации

Weighted Hybrid — взвешенное усреднение скоров разных моделей. Простейший подход, хорошо работает когда компоненты независимы.

Cascade Hybrid — retrieval → scoring → re-ranking. Каждый уровень фильтрует и улучшает предыдущий.

Feature Augmentation — эмбеддинги одной модели как признаки для другой.

Mixed — разные алгоритмы для разных сегментов пользователей или контекстов.

Ensemble с динамическими весами

import numpy as np
from sklearn.linear_model import LogisticRegression
import pandas as pd

class HybridRecommender:
    def __init__(self, collaborative_model, content_model, popular_model):
        self.cf_model = collaborative_model
        self.cb_model = content_model
        self.popular_model = popular_model
        self.weight_model = None  # Meta-learner для весов

    def train_ensemble_weights(self, val_interactions: pd.DataFrame,
                                user_features: pd.DataFrame) -> None:
        """Обучение meta-learner для динамических весов"""
        X_meta = []
        y_meta = []

        for _, row in val_interactions.iterrows():
            user_id = row['user_id']
            item_id = row['item_id']
            label = row['purchased']

            user_feats = user_features[user_features['user_id'] == user_id].iloc[0]
            history_len = user_feats.get('interaction_count', 0)
            item_popularity = user_feats.get('item_popularity', 0.5)
            has_content = user_feats.get('has_rich_content', True)

            # CF score
            cf_score = self._get_cf_score(user_id, item_id)
            # CB score
            cb_score = self._get_cb_score(user_id, item_id)
            # Popular score
            pop_score = self._get_popular_score(item_id)

            meta_features = [
                cf_score, cb_score, pop_score,
                np.log1p(history_len),
                item_popularity,
                int(has_content),
                cf_score - cb_score,  # разность сигналов
                cf_score * np.log1p(history_len)  # взаимодействие
            ]
            X_meta.append(meta_features)
            y_meta.append(label)

        self.weight_model = LogisticRegression(C=1.0, max_iter=200)
        self.weight_model.fit(np.array(X_meta), np.array(y_meta))

    def recommend(self, user_id: str, n: int = 10,
                   user_context: dict = None) -> list[tuple]:
        """Гибридные рекомендации с автоматическим выбором стратегии"""
        history_len = user_context.get('interaction_count', 0) if user_context else 0

        # Стратегия зависит от данных о пользователе
        if history_len == 0:
            # Новый пользователь: только популярное + content если есть входные данные
            return self._cold_start_recommend(user_id, user_context, n)
        elif history_len < 10:
            # Мало данных: 30% CF + 50% CB + 20% popular
            return self._sparse_user_recommend(user_id, n)
        else:
            # Достаточно данных: weighted ensemble
            return self._full_ensemble_recommend(user_id, n)

    def _full_ensemble_recommend(self, user_id: str, n: int) -> list[tuple]:
        """Полный ансамбль для пользователей с историей"""
        # Получаем топ кандидатов от каждой модели
        cf_candidates = dict(self.cf_model.recommend(user_id, n=n*3))
        cb_candidates = dict(self.cb_model.recommend(user_id, n=n*3))
        pop_candidates = dict(self.popular_model.get_popular(n=n*2))

        all_items = set(cf_candidates) | set(cb_candidates) | set(pop_candidates)

        scored = []
        for item_id in all_items:
            cf_score = cf_candidates.get(item_id, 0)
            cb_score = cb_candidates.get(item_id, 0)
            pop_score = pop_candidates.get(item_id, 0)

            if self.weight_model is not None:
                meta_features = np.array([[cf_score, cb_score, pop_score, 0, 0, 1,
                                          cf_score - cb_score, 0]])
                final_score = self.weight_model.predict_proba(meta_features)[0][1]
            else:
                # Статические веса как fallback
                final_score = 0.5 * cf_score + 0.3 * cb_score + 0.2 * pop_score

            scored.append((item_id, final_score))

        scored.sort(key=lambda x: x[1], reverse=True)
        return scored[:n]

    def _cold_start_recommend(self, user_id: str,
                               context: dict, n: int) -> list[tuple]:
        """Рекомендации для нового пользователя"""
        if context and context.get('onboarding_preferences'):
            # Пользователь указал предпочтения при регистрации
            return self.cb_model.recommend_by_preferences(
                context['onboarding_preferences'], n=n
            )
        # Иначе — популярное в категории
        category = context.get('browsed_category') if context else None
        return self.popular_model.get_popular_in_category(category, n=n)

    def _sparse_user_recommend(self, user_id: str, n: int) -> list[tuple]:
        """Мало данных: больше content-based"""
        cf = dict(self.cf_model.recommend(user_id, n=n*2) or [])
        cb = dict(self.cb_model.recommend(user_id, n=n*2) or [])
        pop = dict(self.popular_model.get_popular(n=n) or [])

        all_items = set(cf) | set(cb) | set(pop)
        scored = []
        for item_id in all_items:
            score = (0.2 * cf.get(item_id, 0) +
                     0.6 * cb.get(item_id, 0) +
                     0.2 * pop.get(item_id, 0))
            scored.append((item_id, score))

        scored.sort(key=lambda x: x[1], reverse=True)
        return scored[:n]

    def _get_cf_score(self, user_id, item_id) -> float:
        try:
            recs = dict(self.cf_model.recommend(user_id, n=100))
            return recs.get(item_id, 0.0)
        except Exception:
            return 0.0

    def _get_cb_score(self, user_id, item_id) -> float:
        try:
            profile = self.cb_model.get_user_profile(user_id)
            if profile is None:
                return 0.0
            recs = dict(self.cb_model.recommend(profile, n=100))
            return recs.get(item_id, 0.0)
        except Exception:
            return 0.0

    def _get_popular_score(self, item_id) -> float:
        popularity = getattr(self.popular_model, 'item_popularity', {})
        return popularity.get(item_id, 0.0)

Результаты гибридизации

Динамические веса через meta-learner дают +6-8% к метрикам против статических. Обучение meta-learner: 1-2 часа на валидационной выборке. Ключевые сигналы для определения весов: количество взаимодействий пользователя, давность последнего действия, наличие контентных метаданных.