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

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

Netflix-style рекомендации: основная задача — удержание пользователя на платформе. Метрика успеха — не CTR, а watch time и session continuation rate. Специфика: контент дорогой и ограниченный, важен не только вкус но и настроение, пользователи делятся профилями семьёй.

Мультиконтекстная модель

import numpy as np
import pandas as pd
import torch
import torch.nn as nn

class VideoStreamingRecommender(nn.Module):
    """Учитывает контекст просмотра: время, устройство, соратники"""

    def __init__(self, n_users, n_items, n_genres, embed_dim=128):
        super().__init__()
        # User tower
        self.user_emb = nn.Embedding(n_users + 1, embed_dim)
        self.genre_emb = nn.Embedding(n_genres + 1, 32)

        # Context features
        self.context_mlp = nn.Sequential(
            nn.Linear(10, 32),  # hour, day, device_type, etc.
            nn.ReLU()
        )

        # Item tower
        self.item_emb = nn.Embedding(n_items + 1, embed_dim)
        self.genre_item_emb = nn.Embedding(n_genres + 1, 32)

        # Scoring head
        self.scoring = nn.Sequential(
            nn.Linear(embed_dim + 32 + 32 + embed_dim + 32, 128),
            nn.ReLU(),
            nn.Dropout(0.3),
            nn.Linear(128, 1),
            nn.Sigmoid()
        )

    def forward(self, user_id, context_features, item_id,
                 item_genre, user_top_genres):
        u = self.user_emb(user_id)
        g = self.genre_emb(user_top_genres).mean(dim=1)
        c = self.context_mlp(context_features)

        i = self.item_emb(item_id)
        ig = self.genre_item_emb(item_genre)

        combined = torch.cat([u, g, i, ig], dim=1)
        return self.scoring(combined).squeeze(1)


class WatchHistoryFeatureExtractor:
    """Признаки из истории просмотров"""

    def extract_user_features(self, watch_history: pd.DataFrame) -> dict:
        """
        watch_history: user_id, item_id, watched_seconds, total_seconds,
                       genre, timestamp, device
        """
        completion_rates = watch_history['watched_seconds'] / watch_history['total_seconds'].clip(1)

        features = {
            'completion_rate_avg': completion_rates.mean(),
            'completion_rate_std': completion_rates.std(),
            'binge_sessions': self._count_binge_sessions(watch_history),
            'preferred_genres': watch_history.groupby('genre')['watched_seconds'].sum().nlargest(3).index.tolist(),
            'preferred_device': watch_history['device'].value_counts().index[0],
            'avg_session_items': self._avg_items_per_session(watch_history),
            'evening_watcher': self._is_evening_watcher(watch_history),
            'weekend_preference': self._weekend_ratio(watch_history),
        }
        return features

    def _count_binge_sessions(self, history: pd.DataFrame) -> int:
        """Сессии с 3+ эпизодами подряд"""
        history = history.sort_values('timestamp')
        history['session_gap'] = history['timestamp'].diff().dt.total_seconds() > 1800
        history['session_id'] = history['session_gap'].cumsum()
        session_counts = history.groupby('session_id').size()
        return int((session_counts >= 3).sum())

    def _is_evening_watcher(self, history: pd.DataFrame) -> bool:
        evening_views = history[
            pd.to_datetime(history['timestamp']).dt.hour.between(18, 23)
        ]
        return len(evening_views) / max(len(history), 1) > 0.5

    def _weekend_ratio(self, history: pd.DataFrame) -> float:
        weekend = pd.to_datetime(history['timestamp']).dt.dayofweek >= 5
        return weekend.mean()

    def _avg_items_per_session(self, history: pd.DataFrame) -> float:
        if 'session_id' not in history.columns:
            return 1.5
        return history.groupby('session_id').size().mean()

Ранжирование с учётом продолжаемости серий

class SeriesContinuationBooster:
    """Буст для следующих эпизодов серий, которые пользователь смотрит"""

    def boost_continuation(self, candidates: list[tuple],
                            user_watch_history: pd.DataFrame,
                            content_metadata: dict) -> list[tuple]:
        """Повышение приоритета продолжений"""
        # Серии в прогрессе
        series_progress = (
            user_watch_history
            .groupby('series_id')['episode_number']
            .max()
            .to_dict()
        )

        boosted = []
        for item_id, score in candidates:
            meta = content_metadata.get(item_id, {})
            series_id = meta.get('series_id')
            episode = meta.get('episode_number', 1)

            boost = 1.0
            if series_id and series_id in series_progress:
                watched_episode = series_progress[series_id]
                if episode == watched_episode + 1:
                    boost = 2.5  # Следующий эпизод
                elif episode <= watched_episode:
                    boost = 0.1  # Уже смотрел

            boosted.append((item_id, score * boost))

        return sorted(boosted, key=lambda x: x[1], reverse=True)

Ключевые метрики стриминга: watch time per session (главная), continuation rate (% пользователей, посмотревших следующий эпизод), diversity of content consumed, content-driven subscriber retention. Netflix сообщает, что рекомендательная система генерирует ~80% потребляемого контента. Правильная реализация серийного буста увеличивает session depth на 30-40%.