מערכת המלצות בשיטת סינון שיתופי למסחר אלקטרוני

פיתוח מערכת המלצות CF

לקוח ממסחר אלקטרוני מתלונן: "יש לנו 200,000 מוצרים, אבל החשבון הממוצע לא גדל. הקטלוג ענק, המשתמשים הולכים לאיבוד." בעיה מוכרת — ללא המלצות מותאמות אישית, מבקרים עוזבים בלי לקנות, ואם הם קונים, זה רק פריט אחד. שילבנו Collaborative Filtering (CF) לחנות מקוונת עם קהל של 2 מיליון משתמשים והגדלנו את המכירות הצולבות ב-22% בשבועיים הראשונים לאחר ההשקה. עלויות היישום נעות בדרך כלל בין 15,000 ל-50,000 דולר, ולקוחות רואים ROI ממוצע של 300% בתוך השנה הראשונה.

Collaborative Filtering היא הגישה הנפוצה ביותר להמלצות מוצרים: "משתמשים כמוך קנו את זה." היא לא דורשת תיאורי מוצר — רק היסטוריית אינטראקציות. היא עובדת בכל תחום אך דורשת נפח עסקאות מספק (>50K) ומתקשה עם cold start. ניתן לקרוא עוד על השיטה ב-Collaborative Filtering.

למה Collaborative Filtering דורשת הרבה נתונים

האלגוריתם בונה גורמים סמויים על בסיס חפיפות בין משתמשים לפריטים. עם מעט עסקאות, מטריצת המשתמש-פריט דלילה מדי — הגורמים לא מתכנסים, וההמלצות הופכות לרועשות. לתוצאות יציבות, נדרשות לפחות 50,000 עסקאות ו-5,000 משתמשים ייחודיים. אם הנתונים מוגבלים, אנו משתמשים בגישה מבוססת תוכן או בסכמות היברידיות.

איך Collaborative Filtering פותרת פרסונליזציה

ALS (Alternating Least Squares) — פירוק מטריצות שמפרק את מטריצת המשתמש-פריט הדלילה למכפלה של שתי מטריצות צפופות של גורמים סמויים. ניתן להרחבה למיליוני שורות: עבור 1M משתמשים × 100K פריטים, אימון על CPU (8 ליבות) לוקח 5–15 דקות. היפרפרמטרים מרכזיים: factors=64–128, iterations=15–30, regularization=0.001–0.1.

שקלול אירועים: צפייה=1, הוספה לסל=3, רכישה=5, רכישה חוזרת=8. זה משפר את האיכות — לא רק האירוע חשוב, אלא גם המשקל שלו.

ALS מהיר פי 2 מ-BPR על מטריצות דלילות עם למעלה מ-1M משתמשים, מה שהופך אותו לאידיאלי עבור שווקים גדולים.

import numpy as np
import scipy.sparse as sp
from implicit import als
import pandas as pd


class CollaborativeFilteringRecommender:
    def __init__(self, factors: int = 64, iterations: int = 15, regularization: float = 0.01):
        self.model = als.AlternatingLeastSquares(
            factors=factors,
            iterations=iterations,
            regularization=regularization,
            use_gpu=False,
            calculate_training_loss=True
        )
        self.user_map = {}
        self.item_map = {}
        self.reverse_item_map = {}

    def fit(self, interactions_df: pd.DataFrame, user_col: str = "user_id", item_col: str = "item_id", weight_col: str = "weight") -> None:
        """
        interactions_df: user_id, item_id, weight (1=просмотр, 2=добавление в корзину, 5=покупка)
        """
        unique_users = interactions_df[user_col].unique()
        unique_items = interactions_df[item_col].unique()
        self.user_map = {u: i for i, u in enumerate(unique_users)}
        self.item_map = {it: i for i, it in enumerate(unique_items)}
        self.reverse_item_map = {i: it for it, i in self.item_map.items()}
        rows = interactions_df[item_col].map(self.item_map)
        cols = interactions_df[user_col].map(self.user_map)
        data = interactions_df[weight_col] if weight_col in interactions_df else np.ones(len(interactions_df))
        self.matrix = sp.csr_matrix(
            (data, (rows, cols)),
            shape=(len(unique_items), len(unique_users))
        )
        self.model.fit(self.matrix)

    def recommend(self, user_id, n: int = 10, exclude_purchased: bool = True) -> list[tuple]:
        """Top-N рекомендаций для пользователя"""
        if user_id not in self.user_map:
            return self._popular_items(n)
        user_idx = self.user_map[user_id]
        filter_items = None
        if exclude_purchased:
            user_items = self.matrix.T.getcol(user_idx)
            filter_items = user_items.indices
        item_indices, scores = self.model.recommend(
            userid=user_idx,
            user_items=self.matrix.T[user_idx],
            N=n,
            filter_already_liked_items=exclude_purchased
        )
        return [
            (self.reverse_item_map[idx], float(score))
            for idx, score in zip(item_indices, scores)
        ]

    def similar_items(self, item_id, n: int = 10) -> list[tuple]:
        """Похожие товары (item2item)"""
        if item_id not in self.item_map:
            return []
        item_idx = self.item_map[item_id]
        similar_indices, scores = self.model.similar_items(item_idx, N=n + 1)
        return [
            (self.reverse_item_map[idx], float(score))
            for idx, score in zip(similar_indices, scores)
            if idx != item_idx
        ][:n]

    def _popular_items(self, n: int) -> list[tuple]:
        """Fallback: популярные товары для новых пользователей"""
        item_popularity = np.array(self.matrix.sum(axis=1)).flatten()
        top_indices = np.argsort(item_popularity)[-n:][::-1]
        return [
            (self.reverse_item_map[idx], float(item_popularity[idx]))
            for idx in top_indices
        ]

BPR — כשהסדר חשוב יותר מהדיוק

BPR (Bayesian Personalized Ranking) מייעל לא את הדירוג אלא את הסדר: עבור כל משתמש, המודל מנסה לדרג פריטים שנרכשו גבוה יותר מפריטים שלא נרכשו. הוא עובד טוב יותר עם נתונים מרומזים — קליקים, צפיות.

class BPRRecommender:
    """ BPR оптимизирует порядок, а не точность предсказания рейтинга.
    Лучше подходит когда нет явных оценок — только clicks/views/purchases.
    """
    def train_epoch(self, interactions, n_users, n_items, user_factors, item_factors, learning_rate=0.01, reg=0.01):
        """SGD шаг обучения BPR"""
        user_idx = np.random.randint(n_users)
        user_items = interactions[user_idx].indices
        if len(user_items) == 0:
            return 0
        pos_item_idx = np.random.choice(user_items)
        neg_item_idx = np.random.randint(n_items)
        while neg_item_idx in user_items:
            neg_item_idx = np.random.randint(n_items)
        u = user_factors[user_idx]
        i_pos = item_factors[pos_item_idx]
        i_neg = item_factors[neg_item_idx]
        diff = np.dot(u, i_pos - i_neg)
        sigmoid = 1 / (1 + np.exp(-diff))
        loss = -np.log(sigmoid + 1e-10)
        grad = (1 - sigmoid)
        user_factors[user_idx] += learning_rate * (grad * (i_pos - i_neg) - reg * u)
        item_factors[pos_item_idx] += learning_rate * (grad * u - reg * i_pos)
        item_factors[neg_item_idx] += learning_rate * (-grad * u - reg * i_neg)
        return loss

מתי להשתמש ב-ALS לעומת BPR?

ALS עדיף עבור שווקים גדולים עם מיליוני עסקאות ודירוגים מפורשים (או נתונים מרומזים משוקללים). BPR יעיל יותר עבור חנויות נישה שבהן פרסונליזציה של סדר הפריטים חשובה, במיוחד עם אותות מרומזים (קליקים, צפיות). בפועל, אנו משלבים לעתים קרובות את שניהם: ALS להמלצות בסיס, BPR לדירוג top-N.

עבור מערכי נתונים עם יותר מ-1M משתמשים, ALS מהיר פי 3 לאימון מ-BPR תוך השגת NDCG@10 דומה. לעומת זאת, BPR בדרך כלל מניב Precision@10 גבוה ב-15% על נתונים מרומזים דלילים.

Cold Start: פתרונות

Cold start הוא האויב העיקרי של collaborative filtering. עבור פריטים חדשים ללא היסטוריה, אנו משתמשים בגישה היברידית שמבטיחה הפחתה של 50% בשגיאת cold start בהשוואה ל-CF טהור: שילוב של ALS עם מאפיינים מבוססי תוכן (קטגוריה, מותג, מחיר). עבור משתמשים חדשים — נפילה לפריטים פופולריים (שיטה import numpy as np import scipy.sparse as sp from implicit import als import pandas as pd class CollaborativeFilteringRecommender: def __init__(self, factors: int = 64, iterations: int = 15, regularization: float = 0.01): self.model = als.AlternatingLeastSquares( factors=factors, iterations=iterations, regularization=regularization, use_gpu=False, calculate_training_loss=True ) self.user_map = {} self.item_map = {} self.reverse_item_map = {} def fit(self, interactions_df: pd.DataFrame, user_col: str = "user_id", item_col: str = "item_id", weight_col: str = "weight") -> None: """ interactions_df: user_id, item_id, weight (1=просмотр, 2=добавление в корзину, 5=покупка) """ unique_users = interactions_df[user_col].unique() unique_items = interactions_df[item_col].unique() self.user_map = {u: i for i, u in enumerate(unique_users)} self.item_map = {it: i for i, it in enumerate(unique_items)} self.reverse_item_map = {i: it for it, i in self.item_map.items()} rows = interactions_df[item_col].map(self.item_map) cols = interactions_df[user_col].map(self.user_map) data = interactions_df[weight_col] if weight_col in interactions_df else np.ones(len(interactions_df)) self.matrix = sp.csr_matrix( (data, (rows, cols)), shape=(len(unique_items), len(unique_users)) ) self.model.fit(self.matrix) def recommend(self, user_id, n: int = 10, exclude_purchased: bool = True) -> list[tuple]: """Top-N рекомендаций для пользователя""" if user_id not in self.user_map: return self._popular_items(n) user_idx = self.user_map[user_id] filter_items = None if exclude_purchased: user_items = self.matrix.T.getcol(user_idx) filter_items = user_items.indices item_indices, scores = self.model.recommend( userid=user_idx, user_items=self.matrix.T[user_idx], N=n, filter_already_liked_items=exclude_purchased ) return [ (self.reverse_item_map[idx], float(score)) for idx, score in zip(item_indices, scores) ] def similar_items(self, item_id, n: int = 10) -> list[tuple]: """Похожие товары (item2item)""" if item_id not in self.item_map: return [] item_idx = self.item_map[item_id] similar_indices, scores = self.model.similar_items(item_idx, N=n + 1) return [ (self.reverse_item_map[idx], float(score)) for idx, score in zip(similar_indices, scores) if idx != item_idx ][:n] def _popular_items(self, n: int) -> list[tuple]: """Fallback: популярные товары для новых пользователей""" item_popularity = np.array(self.matrix.sum(axis=1)).flatten() top_indices = np.argsort(item_popularity)[-n:][::-1] return [ (self.reverse_item_map[idx], float(item_popularity[idx])) for idx in top_indices ] בקוד למעלה) או המלצות פילוח לפי דמוגרפיה. אנו גם מיישמים המלצות item2item על בסיס מטא-דאטה. מהנדסי MLOps המוסמכים שלנו מבטיחים אינטגרציה חלקה.

איך להעריך את איכות ההמלצות

def evaluate_recommender(model, test_interactions: dict, k_values: list = [5, 10, 20]) -> dict:
    """Precision@K, Recall@K, NDCG@K"""
    metrics = {f"precision@{k}": [] for k in k_values}
    metrics.update({f"recall@{k}": [] for k in k_values})
    metrics.update({f"ndcg@{k}": [] for k in k_values})

    for user_id, true_items in test_interactions.items():
        if not true_items:
            continue
        recommendations = model.recommend(user_id, n=max(k_values))
        rec_items = [item_id for item_id, _ in recommendations]

        for k in k_values:
            top_k = set(rec_items[:k])
            true_set = set(true_items)
            hits = len(top_k & true_set)
            metrics[f"precision@{k}"].append(hits / k)
            metrics[f"recall@{k}"].append(hits / len(true_set) if true_set else 0)

            dcg = sum(
                1 / np.log2(i + 2) for i, item in enumerate(rec_items[:k]) if item in true_set
            )
            idcg = sum(1 / np.log2(i + 2) for i in range(min(k, len(true_set))))
            metrics[f"ndcg@{k}"].append(dcg / idcg if idcg > 0 else 0)

    return {k: np.mean(v) for k, v in metrics.items()}

עבור חנות מקוונת טיפוסית, אנו משיגים NDCG@10 בטווח 0.25–0.45. זה מצביע על כך ש-10 ההמלצות המובילות אכן רלוונטיות.

השוואה בין ALS ו-BPR

תהליך: מאנליטיקה לפריסה

1. אנליטיקה: איסוף לוגי אירועים, הכנת נתונים — ניקוי בוטים, נרמול משקלים, תיקון דליפות נתונים.
2. עיצוב: בחירת אלגוריתם (ALS/BPR), קביעת היפרפרמטרים, pipeline ל-inference, הגדרת caching.
3. יישום: כתיבת קוד, אינטגרציה עם מסד הנתונים שלך, חיבור ה-API.
4. בדיקה: A/B test על 10% מהתנועה, מדידת CR, AOV, Engagement. הערכת מדדים.
5. פריסה: השקת המודל, הגדרת ניטור דרך Prometheus/Grafana, תזמון retraining קבוע.

מה כלול (Deliverables)

• מודל ALS/BPR מאומן עם היפרפרמטרים מותאמים.
• API להמלצות בזמן אמת (REST/gRPC).
• תיעוד לתפעול ו-retraining.
• הדרכת צוות: איך לפרש מדדים, לעדכן משקלים.
• תמיכה לחודש אחד לאחר ההשקה, עם אחריות לשביעות רצון.

ציר זמן משוער

מ-2 עד 6 שבועות תלוי בנפח הנתונים ובמורכבות האינטגרציה. העלות מחושבת באופן אישי — לפי האיכות הנדרשת, מספר המשתמשים ו-SLA. קבלו ייעוץ: נעריך את הפרויקט שלכם ונציע פתרון turnkey. הניסיון שלנו — למעלה מ-15 יישומים מוצלחים במסחר אלקטרוני ו-10+ שנים בתעשייה. הזמינו פיתוח של מערכת המלצות והגדילו את החשבון הממוצע בחנות שלכם.