biplobgon/product-recommendation-system

1"""2models/content_based.py3------------------------4Content-Based Filtering using TF-IDF on item property values.5 6EDA rationale7-------------8- ~230k items have metadata but no behavioral events (pure cold-start).9- ~50k items have events but no metadata — no CB signal for those.10- TF-IDF on concatenated property values captures item similarity without11  requiring any user interaction history.12- Cosine similarity on the TF-IDF matrix is the core similarity function.13"""14from __future__ import annotations15 16import pickle17from pathlib import Path18 19import numpy as np20import pandas as pd21import scipy.sparse as sp22from sklearn.metrics.pairwise import cosine_similarity23 24from utils.logger import get_logger25 26logger = get_logger(__name__)27 28 29class ContentBasedRecommender:30    """Item-to-item content similarity recommender.31 32    Parameters33    ----------34    top_k_similar:35        Number of most-similar items to pre-compute per item.36    """37 38    def __init__(self, top_k_similar: int = 50) -> None:39        self.top_k_similar = top_k_similar40        self._similarity_index: dict[int, list[tuple[int, float]]] = {}41        self._item_ids: list[int] = []42        self._vectorizer = None43        self._tfidf_matrix = None44 45    # ------------------------------------------------------------------46    # Training47    # ------------------------------------------------------------------48 49    def fit(50        self,51        item_ids: pd.Series,52        tfidf_matrix: sp.spmatrix,53        vectorizer=None,54    ) -> "ContentBasedRecommender":55        """Build the item similarity index.56 57        Parameters58        ----------59        item_ids:60            Series of item IDs aligned with rows of tfidf_matrix.61        tfidf_matrix:62            Sparse TF-IDF matrix (n_items × n_features).63        vectorizer:64            Fitted TfidfVectorizer (stored for later inference on new items).65 66        Returns67        -------68        self69        """70        logger.info(71            "Building content similarity index for %d items …", len(item_ids)72        )73        self._item_ids = item_ids.tolist()74        self._tfidf_matrix = tfidf_matrix75        self._vectorizer = vectorizer76 77        id_to_idx = {iid: i for i, iid in enumerate(self._item_ids)}78        batch_size = 100079        n = len(self._item_ids)80 81        for start in range(0, n, batch_size):82            end = min(start + batch_size, n)83            batch = tfidf_matrix[start:end]84            sims = cosine_similarity(batch, tfidf_matrix)   # (batch, n)85 86            for local_i, global_i in enumerate(range(start, end)):87                row = sims[local_i]88                row[global_i] = -1.0          # exclude self89                top_indices = np.argpartition(row, -self.top_k_similar)[-self.top_k_similar:]90                top_indices = top_indices[np.argsort(row[top_indices])[::-1]]91                self._similarity_index[self._item_ids[global_i]] = [92                    (self._item_ids[j], float(row[j])) for j in top_indices93                ]94 95            if start % 10000 == 0:96                logger.info("  Similarity index: %d / %d items processed.", end, n)97 98        logger.info("Content similarity index built.")99        return self100 101    # ------------------------------------------------------------------102    # Inference103    # ------------------------------------------------------------------104 105    def recommend_similar(106        self, item_id: int, top_k: int = 10107    ) -> list[tuple[int, float]]:108        """Return items most similar to a given item.109 110        Parameters111        ----------112        item_id:113            Seed item.114        top_k:115            Number of similar items to return.116 117        Returns118        -------119        List of (itemid, similarity_score) tuples.120        """121        if item_id not in self._similarity_index:122            logger.warning("Item %s not in similarity index.", item_id)123            return []124        return self._similarity_index[item_id][:top_k]125 126    def recommend_for_session(127        self, session_items: list[int], top_k: int = 10128    ) -> list[tuple[int, float]]:129        """Recommend items based on the items viewed in the current session.130 131        Aggregates similarity scores across all session items and returns132        the top-k candidates not already in the session.133 134        Parameters135        ----------136        session_items:137            Ordered list of item IDs in the current session.138        top_k:139            Number of recommendations.140 141        Returns142        -------143        List of (itemid, aggregated_score) tuples.144        """145        scores: dict[int, float] = {}146        seen = set(session_items)147 148        for seed_item in session_items:149            for candidate_id, sim in self.recommend_similar(seed_item, top_k=50):150                if candidate_id not in seen:151                    scores[candidate_id] = scores.get(candidate_id, 0.0) + sim152 153        ranked = sorted(scores.items(), key=lambda x: x[1], reverse=True)154        return ranked[:top_k]155 156    # ------------------------------------------------------------------157    # Persistence158    # ------------------------------------------------------------------159 160    def save(self, path: str | Path) -> None:161        path = Path(path)162        path.parent.mkdir(parents=True, exist_ok=True)163        with open(path, "wb") as fh:164            pickle.dump(self, fh)165        logger.info("ContentBasedRecommender saved to %s", path)166 167    @classmethod168    def load(cls, path: str | Path) -> "ContentBasedRecommender":169        with open(path, "rb") as fh:170            obj = pickle.load(fh)171        logger.info("ContentBasedRecommender loaded from %s", path)172        return obj173