biplobgon/product-recommendation-system

1"""2models/session_based.py3------------------------4Session-Based Recommender using Item-KNN with session co-occurrence.5 6Pure numpy/scipy implementation — no PyTorch required.7 8Algorithm9---------10For a current session S = [i1, i2, ..., in]:11  1. Look up all items that co-occurred with any item in S across training sessions.12  2. Score each candidate by sum of co-occurrence counts weighted by recency13     (more recent items in the session get higher weight: 1, 2, …, n).14  3. Exclude items already seen in the current session.15  4. Return top-k by score.16 17EDA rationale18-------------19- >70% of visitors have only 1–3 events — sessions are the most reliable signal.20- Average session length ~2-4 items → co-occurrence is well-defined.21- Item-KNN is competitive with GRU4Rec on short sessions (see Ludewig & Jannach 2018).22"""23from __future__ import annotations24 25import pickle26from collections import defaultdict27from pathlib import Path28 29import numpy as np30import pandas as pd31 32from utils.logger import get_logger33 34logger = get_logger(__name__)35 36 37class SessionBasedRecommender:38    """Item-KNN session-based recommender using co-occurrence counts.39 40    Parameters41    ----------42    max_session_length:43        Maximum number of recent items in a session to consider.44    top_k_similar:45        Number of co-occurrence neighbours to store per item.46    """47 48    def __init__(49        self,50        emb_dim: int = 64,           # kept for API compatibility (unused)51        hidden_size: int = 128,      # kept for API compatibility (unused)52        num_layers: int = 1,         # kept for API compatibility (unused)53        dropout: float = 0.2,        # kept for API compatibility (unused)54        max_session_length: int = 20,55    ) -> None:56        self.max_session_length = max_session_length57        # co-occurrence index: item_id → {neighbour_id: count}58        self._cooc: dict[int, dict[int, float]] = {}59        # global item popularity scores (fallback)60        self._popularity: dict[int, float] = {}61 62    # ------------------------------------------------------------------63    # Training64    # ------------------------------------------------------------------65 66    def fit(67        self,68        sequences: pd.DataFrame,69        epochs: int = 1,        # unused, kept for API compatibility70        batch_size: int = 256,  # unused, kept for API compatibility71        lr: float = 0.001,      # unused, kept for API compatibility72    ) -> "SessionBasedRecommender":73        """Build co-occurrence index from session sequences.74 75        Parameters76        ----------77        sequences:78            DataFrame with columns [session_id, item_sequence, target_item].79        """80        logger.info("Building session co-occurrence index from %d sequences …", len(sequences))81 82        cooc: dict[int, dict[int, float]] = defaultdict(lambda: defaultdict(float))83        popularity: dict[int, float] = defaultdict(float)84 85        for row in sequences.itertuples(index=False):86            seq: list[int] = list(row.item_sequence)87            target: int = int(row.target_item)88            n = len(seq)89            # Weight items by position (last item = highest weight)90            for pos, item in enumerate(seq):91                weight = float(pos + 1) / n92                popularity[item] += weight93                # co-occurrence: every item in seq co-occurs with target94                cooc[item][target] += weight95                cooc[target][item] += weight96 97        self._cooc = {k: dict(v) for k, v in cooc.items()}98        self._popularity = dict(popularity)99        logger.info(100            "Co-occurrence index built: %d items, %d total entries.",101            len(self._cooc),102            sum(len(v) for v in self._cooc.values()),103        )104        return self105 106    # ------------------------------------------------------------------107    # Inference108    # ------------------------------------------------------------------109 110    def recommend(111        self, session_items: list[int], top_k: int = 10112    ) -> list[tuple[int, float]]:113        """Score candidates given the current session.114 115        Parameters116        ----------117        session_items:118            Ordered list of item IDs viewed in the current session.119        top_k:120            Number of top items to return.121        """122        session_items = session_items[-self.max_session_length:]123        seen = set(session_items)124        n = len(session_items)125 126        scores: dict[int, float] = defaultdict(float)127        for pos, item in enumerate(session_items):128            weight = float(pos + 1) / n   # recency weighting129            for neighbour, co_score in self._cooc.get(item, {}).items():130                if neighbour not in seen:131                    scores[neighbour] += weight * co_score132 133        if not scores:134            # Fallback: global popularity135            candidates = [136                (iid, sc) for iid, sc in self._popularity.items() if iid not in seen137            ]138        else:139            candidates = list(scores.items())140 141        candidates.sort(key=lambda x: x[1], reverse=True)142        return candidates[:top_k]143 144    # ------------------------------------------------------------------145    # Persistence146    # ------------------------------------------------------------------147 148    def save(self, path: str | Path) -> None:149        path = Path(path)150        path.parent.mkdir(parents=True, exist_ok=True)151        with open(path, "wb") as fh:152            pickle.dump(self, fh)153        logger.info("SessionBasedRecommender saved to %s", path)154 155    @classmethod156    def load(cls, path: str | Path) -> "SessionBasedRecommender":157        with open(path, "rb") as fh:158            obj = pickle.load(fh)159        logger.info("SessionBasedRecommender loaded from %s", path)160        return obj161