Spaces:
Running on Zero
Running on Zero
| """ | |
| Scalable recommender core for BookRec on the FULL Goodbooks-10k dataset. | |
| We never build a dense item-item similarity matrix. Instead we use the | |
| standard production pattern: | |
| * Collaborative filtering : truncated SVD (scipy.sparse.linalg.svds) of the | |
| sparse user x item rating matrix -> low-dimensional item factors. | |
| "Books similar to X" = X's factor vector's nearest neighbours (ANN via | |
| sklearn NearestNeighbors / brute-force dot product). All CPU-feasible. | |
| * Popularity : rank the full catalogue by explicit-rating count | |
| and average rating (a strong baseline). | |
| * Content / search : the resilient encoders (TF-IDF default, or | |
| sentence-transformer) over book text (title + authors + genres) -> top-K | |
| by similarity. | |
| * Hybrid : combine the CF factor similarity and content | |
| similarity for a tuned 'more like this' list. | |
| All components degrade gracefully (empty catalogue, no factors, offline) so the | |
| app always renders something. | |
| """ | |
| from __future__ import annotations | |
| from typing import List, Optional, Sequence, Tuple | |
| import numpy as np | |
| import pandas as pd | |
| import scipy.sparse as sp | |
| from scipy.sparse.linalg import svds | |
| try: | |
| from . import loader | |
| from .encoders import build_content_encoder, build_item_matrix, cosine_similarity_vector | |
| except ImportError: # allow running as a script with src on sys.path | |
| import loader | |
| from encoders import build_content_encoder, build_item_matrix, cosine_similarity_vector | |
| # --------------------------------------------------------------------------- | |
| # Ratings matrix + factorization | |
| # --------------------------------------------------------------------------- | |
| def build_user_item_matrix(ratings: pd.DataFrame) -> Tuple[sp.csr_matrix, np.ndarray, np.ndarray]: | |
| """Return (user_item csr, user_ids, item_ids) for a ratings frame. | |
| Assumes the frame already holds the ratings we want to factor (e.g. explicit). | |
| Matrix is users x items, values = Book-Rating. | |
| """ | |
| ratings = ratings.copy() | |
| ratings["User-ID"] = ratings["User-ID"].astype(str) | |
| ratings["ISBN"] = ratings["ISBN"].astype(str) | |
| users, ui = np.unique(ratings["User-ID"], return_inverse=True) | |
| items, ii = np.unique(ratings["ISBN"], return_inverse=True) | |
| data = ratings["Book-Rating"].astype(np.float32).values | |
| mat = sp.csr_matrix((data, (ui, ii)), shape=(len(users), len(items))) | |
| return mat, users, items | |
| def factorize(mat: sp.csr_matrix, rank: int = 50, | |
| seed: int = 0) -> Tuple[np.ndarray, np.ndarray]: | |
| """Truncated SVD of a sparse user-item matrix. | |
| Returns (user_factors, item_factors), both (n, rank), where | |
| item_factors[i] is the latent vector of item i and can be used directly | |
| for nearest-neighbour "similar items". | |
| """ | |
| # Centre by subtracting the global mean of observed entries (helps SVD). | |
| nnz = mat.nnz | |
| if nnz == 0: | |
| return np.zeros((mat.shape[0], rank)), np.zeros((mat.shape[1], rank)) | |
| mu = mat.data.mean() | |
| m_c = mat.copy() | |
| m_c.data = m_c.data - mu | |
| rank = min(rank, min(mat.shape) - 1) | |
| if rank < 1: | |
| return np.zeros((mat.shape[0], 1)), np.zeros((mat.shape[1], 1)) | |
| try: | |
| u, s, vt = svds(m_c, k=rank, which="LM", random_state=seed) | |
| except Exception: # noqa: BLE001 -- fall back to uncentred | |
| u, s, vt = svds(mat, k=rank, which="LM", random_state=seed) | |
| # Reorder descending (svds returns ascending) | |
| order = np.argsort(s)[::-1] | |
| s, u, vt = s[order], u[:, order], vt[order, :] | |
| user_factors = u * s[np.newaxis, :] # (n_users, rank) | |
| item_factors = vt.T # (n_items, rank) | |
| return user_factors, item_factors | |
| # --------------------------------------------------------------------------- | |
| # Popularity | |
| # --------------------------------------------------------------------------- | |
| def popular_books(ratings: pd.DataFrame, books: pd.DataFrame, top_n: int = 50, | |
| min_ratings: int = 5) -> pd.DataFrame: | |
| """Rank the most-popular books from explicit ratings (count & avg).""" | |
| df = ratings[ratings["Book-Rating"] > 0].copy() | |
| df["ISBN"] = df["ISBN"].astype(str) | |
| stats = df.groupby("ISBN")["Book-Rating"].agg(["count", "mean"]).reset_index() | |
| stats.columns = ["ISBN", "num_ratings", "avg_rating"] | |
| stats = stats[stats["num_ratings"] >= min_ratings] | |
| stats = stats.sort_values(["avg_rating", "num_ratings"], | |
| ascending=False).head(top_n) | |
| m = books.copy() | |
| m["ISBN"] = m["ISBN"].astype(str) | |
| out = stats.merge(m.drop_duplicates("ISBN"), on="ISBN", how="left") | |
| out["cover"] = out.apply(lambda r: loader.pick_cover_url(r), axis=1) | |
| return out.reset_index(drop=True) | |
| # --------------------------------------------------------------------------- | |
| # ANN neighbours (sklearn, no extra deps) | |
| # --------------------------------------------------------------------------- | |
| class ApproxNN: | |
| """Tiny wrapper around scikit-learn NearestNeighbors. | |
| Keeps the factor/search index in memory and returns top-K indices. | |
| """ | |
| def __init__(self, X: np.ndarray, n_neighbors: int = 50, metric: str = "cosine"): | |
| from sklearn.neighbors import NearestNeighbors | |
| self.X = X | |
| self.metric = metric | |
| self.knn = NearestNeighbors(n_neighbors=min(n_neighbors, X.shape[0]), | |
| metric=metric, algorithm="auto", n_jobs=-1) | |
| self.knn.fit(X) | |
| def query(self, vec: np.ndarray, k: int = 10) -> Tuple[np.ndarray, np.ndarray]: | |
| """Return (distances, indices) of the k closest rows to `vec` (excl. self).""" | |
| k = min(k + 1, self.X.shape[0]) | |
| dists, idx = self.knn.kneighbors(np.asarray(vec).reshape(1, -1) | |
| if vec.ndim == 1 else vec, n_neighbors=k) | |
| return dists, idx | |
| # --------------------------------------------------------------------------- | |
| # Recommender | |
| # --------------------------------------------------------------------------- | |
| def fused_ranking(content_ranked, cf_ranked, alpha: float = 0.5, | |
| n: int = 10) -> list: | |
| """Blend two ordered lists by reciprocal rank, matching the reference | |
| notebook's hybrid fusion.""" | |
| from collections import defaultdict | |
| score: dict = defaultdict(float) | |
| for rank, it in enumerate(content_ranked): | |
| score[it] += alpha / (rank + 1.0) | |
| for rank, it in enumerate(cf_ranked): | |
| score[it] += (1.0 - alpha) / (rank + 1.0) | |
| ranked = sorted(score.items(), key=lambda kv: -kv[1]) | |
| return [it for it, _ in ranked[:n]] | |
| class BookRecommender: | |
| """Serves popularity, search, content-similarity and CF-similarity. | |
| The content encoder is resilient: defaults to TF-IDF, optionally uses a | |
| sentence-transformer, and never crashes if the model is unavailable. | |
| """ | |
| def __init__(self, books: pd.DataFrame, ratings: pd.DataFrame, | |
| users: pd.DataFrame | None = None, | |
| encoder_kind: Optional[str] = None, | |
| model_name: str = "sentence-transformers/all-MiniLM-L6-v2", | |
| svd_rank: int = 50, | |
| text_col: str = "Book-Title"): | |
| self.books = books.reset_index(drop=True) | |
| self.books["ISBN"] = self.books["ISBN"].astype(str) | |
| self.ratings = ratings | |
| self.users = users | |
| self.encoder_kind = encoder_kind or "tfidf" | |
| self.svd_rank = svd_rank | |
| # full-catalogue text: title + author (+ genres if present) | |
| parts = [self.books.get(col, pd.Series(index=self.books.index, dtype=str)) | |
| for col in ("Book-Title", "Book-Author", "Book-Genres")] | |
| self.texts = pd.concat(parts, axis=1).fillna("").agg(" ".join, axis=1) | |
| # --- content index (search + similar-by-text) ----------------------- | |
| print(f"[recommender] building content index ({self.encoder_kind}) ...") | |
| self.content_enc = build_content_encoder(encoder_kind, verbose=True) | |
| self.content_mat = self.content_enc.build_matrix(self.texts.tolist()) | |
| self.content_enc.matrix_ = self.content_mat # ensure cached | |
| # --- CF factorization ------------------------------------------------ | |
| print("[recommender] building user-item matrix ...") | |
| self.mat, self.user_ids, self.item_ids = build_user_item_matrix(ratings) | |
| self.item_pos = {isbn: i for i, isbn in enumerate(self.item_ids)} | |
| self.item_isbn_at = self.item_ids.copy() | |
| self.user_pos = {u: i for i, u in enumerate(self.user_ids)} | |
| print(f"[recommender] factorizing (rank={svd_rank}) ...") | |
| self.user_factors, self.item_factors = factorize(self.mat, rank=svd_rank) | |
| self.cf_nn = ApproxNN(self.item_factors) if self.item_factors.shape[0] else None | |
| # item -> row index in self.books | |
| self.isbn_to_row = {row.ISBN: i for i, row in self.books.iterrows()} | |
| # ---- helpers ----------------------------------------------------------- | |
| def row_for_title(self, title: str) -> Optional[int]: | |
| d = self.books[self.books["Book-Title"].str.strip().str.lower() | |
| == str(title).strip().lower()] | |
| return int(d.index[0]) if len(d) else None | |
| # ---- search ------------------------------------------------------------- | |
| def search(self, query: str, top_k: int = 10) -> pd.DataFrame: | |
| res = self.content_enc.search(query, top_k=top_k) | |
| rows = [] | |
| for idx, score in res: | |
| r = self.books.iloc[idx] | |
| rows.append({ | |
| "rank": len(rows) + 1, | |
| "title": r["Book-Title"], | |
| "author": r.get("Book-Author", ""), | |
| "isbn": r["ISBN"], | |
| "cover": loader.pick_cover_url(r), | |
| "score": float(score), | |
| }) | |
| return pd.DataFrame(rows) | |
| # ---- similar items (hybrid of content + CF) ------------------------------ | |
| def similar(self, title: str, top_k: int = 0, hybrid_alpha: float = 0.5, | |
| limit: int = 10, fusion: str = "score") -> pd.DataFrame: | |
| """Return similar books to a title, blending content & CF similarity. | |
| hybrid_alpha in [0,1]: 1 = pure content, 0 = pure CF. fusion is | |
| 'score' (weighted score sum) or 'reciprocal_rank' (reciprocal-rank | |
| fusion). If a factor path is unavailable, falls back to content. | |
| """ | |
| row = self.row_for_title(title) | |
| if row is None: | |
| return pd.DataFrame() | |
| isbn = self.books.iloc[row]["ISBN"] | |
| # content neighbours (text) - cosine similarity, higher is better | |
| q = self.content_enc.encode([self.texts[row]])[0] | |
| content_ids = self._nearest_content(q, limit * 5) | |
| content_scores = {int(i): float(s) for i, s in content_ids.items()} | |
| content_scores.pop(row, None) # exclude the seed book itself | |
| # CF neighbours (latent factor) -- only if the ISBN is in the rating set | |
| # score = 1/(1 + cosine distance), so higher is better in (0, 1] | |
| cf_scores: dict = {} | |
| if self.cf_nn is not None and isbn in self.item_pos: | |
| f = self.item_factors[self.item_pos[isbn]] | |
| dists, idx = self.cf_nn.query(f, k=limit * 5 + 1) | |
| for d, it in zip(dists[0], idx[0]): | |
| if it == self.item_pos[isbn]: | |
| continue | |
| iid = self.item_ids[it] | |
| if iid in self.isbn_to_row: | |
| cf_scores[self.isbn_to_row[iid]] = 1.0 / (1.0 + d) | |
| # Min-max normalise each source to [0, 1] so the alpha blend compares | |
| # content and CF on a fair scale. | |
| def _norm(src: dict) -> dict: | |
| if not src: | |
| return {} | |
| lo, hi = min(src.values()), max(src.values()) | |
| span = hi - lo | |
| if span <= 1e-9: | |
| return {k: 1.0 for k in src} | |
| return {k: (v - lo) / span for k, v in src.items()} | |
| content_scores = _norm(content_scores) | |
| cf_scores = _norm(cf_scores) | |
| # fuse | |
| if fusion == "reciprocal_rank": | |
| content_ranked = sorted(content_scores, key=content_scores.get, | |
| reverse=True) | |
| cf_ranked = sorted(cf_scores, key=cf_scores.get, reverse=True) | |
| picked = fused_ranking(content_ranked, cf_ranked, | |
| alpha=hybrid_alpha, n=limit) | |
| fused = {i: float(len(picked) - j) for j, i in enumerate(picked)} | |
| fused = sorted(fused.items(), key=lambda kv: -kv[1]) | |
| else: | |
| all_idx = set(content_scores) | set(cf_scores) | |
| fused = {} | |
| for i in all_idx: | |
| c = content_scores.get(i, 0.0) | |
| cf = cf_scores.get(i, 0.0) | |
| fused[i] = hybrid_alpha * c + (1 - hybrid_alpha) * cf | |
| fused = sorted(fused.items(), key=lambda kv: -kv[1])[:limit] | |
| out = [] | |
| for idx, score in fused: | |
| r = self.books.iloc[idx] | |
| out.append({"rank": len(out) + 1, "title": r["Book-Title"], | |
| "author": r.get("Book-Author", ""), "isbn": r["ISBN"], | |
| "cover": loader.pick_cover_url(r), "score": float(score)}) | |
| return pd.DataFrame(out) | |
| def _nearest_content(self, q: np.ndarray, k: int) -> dict: | |
| sims = cosine_similarity_vector(q, self.content_mat) | |
| order = np.argsort(-sims)[:k] | |
| return {int(i): float(sims[i]) for i in order} | |
| # ---- content-based user profile ---------------------------------------- | |
| def recommend_content_for_user(self, liked_isbns, k: int = 10, | |
| exclude_known: bool = True) -> pd.DataFrame: | |
| """Content-based recommendation for a user from their liked books. | |
| Averages the content vectors of the books a user liked into a user | |
| profile, then ranks the catalogue by text proximity to that profile. | |
| This is the content-based 'recommend for user' path (no other users | |
| needed), the natural first responder for a reader with no neighbours. | |
| """ | |
| liked = [str(x) for x in liked_isbns if str(x) in self.isbn_to_row] | |
| if not liked: | |
| return pd.DataFrame() | |
| rows = [self.isbn_to_row[x] for x in liked] | |
| profile = np.asarray(self.content_mat[rows].mean(axis=0)).ravel() | |
| sims = cosine_similarity_vector(profile, self.content_mat) | |
| know = set(rows) | |
| out = [] | |
| for i in np.argsort(-sims): | |
| if exclude_known and int(i) in know: | |
| continue | |
| if sims[i] <= 0: | |
| break | |
| r = self.books.iloc[i] | |
| out.append({"rank": len(out) + 1, "title": r["Book-Title"], | |
| "author": r.get("Book-Author", ""), | |
| "isbn": r["ISBN"], | |
| "cover": loader.pick_cover_url(r), | |
| "score": float(sims[i])}) | |
| if len(out) >= k: | |
| break | |
| return pd.DataFrame(out) | |
| # ---- cold start --------------------------------------------------------- | |
| def cold_start_for_new_user(self, k: int = 10, | |
| min_ratings: int = 20) -> pd.DataFrame: | |
| """A brand-new reader has no history. Start them with popular, well | |
| rated books (the bestseller shelf).""" | |
| return popular_books(self.ratings, self.books, top_n=k, | |
| min_ratings=min_ratings) | |
| def cold_start_for_new_item(self, title_text: str, k: int = 10) -> pd.DataFrame: | |
| """A brand-new book has no ratings. Surface it next to the closest | |
| known books by text, boosting the popular ones among them, so it does | |
| not vanish into the long tail.""" | |
| q = self.content_enc.encode([str(title_text)])[0] | |
| sims = cosine_similarity_vector(q, self.content_mat) | |
| counts = self.ratings["ISBN"].value_counts() | |
| order = np.argsort(-sims) | |
| scored = [] | |
| for i in order: | |
| if sims[i] <= 0: | |
| break | |
| isbn = str(self.books.iloc[i]["ISBN"]) | |
| scored.append((int(i), float(sims[i]), int(counts.get(isbn, 0)))) | |
| scored.sort(key=lambda t: (t[2] > 0, t[1]), reverse=True) | |
| out = [] | |
| for i, score, pop in scored[:k]: | |
| r = self.books.iloc[i] | |
| out.append({"rank": len(out) + 1, "title": r["Book-Title"], | |
| "author": r.get("Book-Author", ""), | |
| "isbn": r["ISBN"], | |
| "cover": loader.pick_cover_url(r), | |
| "score": float(score), "n_ratings": pop}) | |
| return pd.DataFrame(out) | |