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

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

Новостные рекомендации — баланс между персонализацией и информационным разнообразием. Проблема эхо-камеры реальна: если только рекомендовать то, что пользователь уже читает, формируем информационный пузырь. Плюс новости быстро устаревают: статья 3-часовой давности ценнее статьи вчерашней.

Time-Aware рекомендации

import numpy as np
import pandas as pd
from sentence_transformers import SentenceTransformer
from sklearn.metrics.pairwise import cosine_similarity
from datetime import datetime, timedelta

class NewsRecommender:
    def __init__(self):
        self.encoder = SentenceTransformer('paraphrase-multilingual-mpnet-base-v2')
        self.articles = {}
        self.article_embeddings = {}

    def add_article(self, article_id: str, title: str, text: str,
                     category: str, published_at: datetime,
                     tags: list = None):
        """Индексация новой статьи"""
        text_for_encoding = f"{title}. {text[:500]}"
        embedding = self.encoder.encode(text_for_encoding, normalize_embeddings=True)

        self.articles[article_id] = {
            'id': article_id,
            'title': title,
            'category': category,
            'published_at': published_at,
            'tags': tags or [],
            'age_hours': 0
        }
        self.article_embeddings[article_id] = embedding

    def compute_freshness_score(self, published_at: datetime,
                                 decay_rate: float = 0.15) -> float:
        """Экспоненциальное затухание по времени"""
        age_hours = (datetime.now() - published_at).total_seconds() / 3600
        # Полужизнь: ln(2)/decay_rate ≈ 4.6 часов при decay=0.15
        freshness = np.exp(-decay_rate * age_hours)
        return float(freshness)

    def recommend(self, user_profile: np.ndarray,
                   read_article_ids: list,
                   n: int = 10,
                   diversity_weight: float = 0.25,
                   freshness_weight: float = 0.3) -> list[dict]:
        """Персонализированные свежие рекомендации"""
        if user_profile is None:
            return self._trending_articles(n)

        scored = []
        category_count = {}

        for article_id, embedding in self.article_embeddings.items():
            if article_id in read_article_ids:
                continue

            article = self.articles[article_id]

            # Релевантность
            relevance = float(cosine_similarity(
                user_profile.reshape(1, -1), embedding.reshape(1, -1)
            )[0][0])

            # Свежесть
            freshness = self.compute_freshness_score(article['published_at'])

            # Штраф за перегрузку категории
            cat = article['category']
            category_count[cat] = category_count.get(cat, 0) + 1
            category_penalty = 1 / category_count[cat] if diversity_weight > 0 else 1

            # Финальный скор
            score = (
                (1 - freshness_weight - diversity_weight) * relevance +
                freshness_weight * freshness +
                diversity_weight * category_penalty
            )

            scored.append({
                'article_id': article_id,
                'title': article['title'],
                'score': score,
                'relevance': relevance,
                'freshness': freshness,
                'category': article['category']
            })

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

    def build_user_profile(self, reading_history: list[dict]) -> np.ndarray:
        """Профиль пользователя из истории чтения"""
        recent_articles = sorted(
            reading_history, key=lambda x: x['timestamp'], reverse=True
        )[:20]

        if not recent_articles:
            return None

        weights = np.exp(-0.1 * np.arange(len(recent_articles)))
        vectors = []
        valid_weights = []

        for article_hist, w in zip(recent_articles, weights):
            article_id = article_hist['article_id']
            if article_id in self.article_embeddings:
                # Умножаем на время чтения (engagement)
                read_ratio = article_hist.get('read_ratio', 1.0)
                vectors.append(self.article_embeddings[article_id])
                valid_weights.append(w * read_ratio)

        if not vectors:
            return None

        profile = np.average(np.vstack(vectors), axis=0,
                             weights=np.array(valid_weights))
        return profile / (np.linalg.norm(profile) + 1e-10)

    def _trending_articles(self, n: int) -> list[dict]:
        """Тренды для новых пользователей"""
        now = datetime.now()
        recent = [
            (aid, a) for aid, a in self.articles.items()
            if (now - a['published_at']).total_seconds() < 86400  # Последние 24 часа
        ]
        # Сортировка по свежести (placeholder: в реальности по просмотрам)
        recent.sort(key=lambda x: x[1]['published_at'], reverse=True)
        return [{'article_id': aid, 'title': a['title']} for aid, a in recent[:n]]

Борьба с эхо-камерой

    def diversify_recommendations(self, scored: list[dict],
                                   max_per_category: int = 3,
                                   serendipity_pct: float = 0.2) -> list[dict]:
        """Диверсификация + случайные открытия"""
        # Лимит по категориям
        cat_count = {}
        filtered = []
        for item in scored:
            cat = item['category']
            if cat_count.get(cat, 0) < max_per_category:
                cat_count[cat] = cat_count.get(cat, 0) + 1
                filtered.append(item)

        # Serendipity: добавляем случайные статьи вне профиля
        n_serendipity = int(len(filtered) * serendipity_pct)
        if n_serendipity > 0:
            all_unread = [
                {'article_id': aid, **a, 'score': 0.3}
                for aid, a in self.articles.items()
                if aid not in {f['article_id'] for f in filtered}
                and self.compute_freshness_score(a['published_at']) > 0.3
            ]
            import random
            serendipity = random.sample(all_unread, min(n_serendipity, len(all_unread)))
            filtered[-n_serendipity:] = serendipity

        return filtered

Freshness decay rate: для breaking news — aggressive (0.3+), для аналитики — gentle (0.05-0.1). Оптимальный serendipity: 15-25% контента вне привычных интересов. Метрики: CTR (2-5% хорошо для новостей), session depth (3+ статьи), return rate (daily active users %).