Content-Based Filtering

"""Content-based filtering - the item vectors, the profile, and the score, transcribed.

There is no fit step in the neural sense. Item vectors are TF-IDF weights, the
user profile is the weighted mean of the liked items (the centroid), and the
score is the cosine between the profile and every unseen item.
"""

import math


def tokenize(text):
    return text.lower().split()


def tfidf_vectors(documents):
    """One dict per document mapping term -> (1 + log tf) * idf."""
    n_docs = len(documents)
    token_lists = [tokenize(document) for document in documents]

    document_frequency = {}
    for tokens in token_lists:
        for term in set(tokens):
            document_frequency[term] = document_frequency.get(term, 0) + 1

    vectors = []
    for tokens in token_lists:
        counts = {}
        for term in tokens:
            counts[term] = counts.get(term, 0) + 1

        vector = {}
        for term, count in counts.items():
            idf = math.log((1 + n_docs) / (1 + document_frequency[term])) + 1.0
            vector[term] = (1.0 + math.log(count)) * idf
        vectors.append(vector)
    return vectors


def build_profile(vectors, liked_items, weights):
    """Weighted mean of the liked item vectors - the closed-form minimizer."""
    profile = {}
    total_weight = 0.0
    for position in range(len(liked_items)):
        weight = weights[position]
        total_weight += weight
        for term, value in vectors[liked_items[position]].items():
            profile[term] = profile.get(term, 0.0) + weight * value

    if total_weight == 0.0:
        raise ValueError("liked items carry no weight")
    for term in profile:
        profile[term] = profile[term] / total_weight
    return profile


def norm(vector):
    total = 0.0
    for value in vector.values():
        total += value * value
    return math.sqrt(total)


def cosine(profile, item):
    dot = 0.0
    for term, value in item.items():
        if term in profile:
            dot += profile[term] * value

    denominator = norm(profile) * norm(item)
    if denominator == 0.0:
        return 0.0
    return dot / denominator


def recommend(vectors, liked_items, weights, k=5):
    profile = build_profile(vectors, liked_items, weights)

    scored = []
    for item in range(len(vectors)):
        if item in liked_items:
            continue
        scored.append((cosine(profile, vectors[item]), item))

    # Full sort of every candidate to keep k of them - correct, and wasteful.
    scored.sort(reverse=True)
    return scored[:k]

"""Content-based filtering - typed, validated, sparse, rows normalised once."""

from collections.abc import Sequence
from dataclasses import dataclass

import numpy as np
from numpy.typing import NDArray
from scipy import sparse


class ColdStartError(ValueError):
    """Raised when a user has no usable feedback, so no profile exists."""


@dataclass(frozen=True)
class ItemIndex:
    """The catalogue as a sparse matrix whose rows are unit vectors.

    Rows are L2-normalised once, here, because cosine similarity is then a
    plain dot product - length drops out of the ranking and the per-item
    denominator never has to be recomputed at query time.
    """

    matrix: sparse.csr_matrix      # (n_items, n_terms), rows have norm 1
    item_ids: tuple[str, ...]


def build_index(counts: sparse.csr_matrix, item_ids: Sequence[str]) -> ItemIndex:
    """TF-IDF with sublinear tf and smoothed idf, then L2 row normalisation."""
    n_items, n_terms = counts.shape
    if n_items != len(item_ids):
        raise ValueError(f"{n_items} rows but {len(item_ids)} item ids")
    if n_items == 0:
        raise ValueError("the catalogue is empty")

    counts = counts.tocsr().astype(np.float64)
    counts.sum_duplicates()

    document_frequency = np.bincount(counts.indices, minlength=n_terms)
    idf = np.log((1.0 + n_items) / (1.0 + document_frequency)) + 1.0

    counts.data = 1.0 + np.log(counts.data)               # sublinear tf
    weighted = sparse.csr_matrix(counts @ sparse.diags(idf))

    norms = np.sqrt(np.asarray(weighted.multiply(weighted).sum(axis=1)).ravel())
    norms[norms == 0.0] = 1.0                              # empty documents stay zero
    normalised = sparse.csr_matrix(sparse.diags(1.0 / norms) @ weighted)
    return ItemIndex(matrix=normalised, item_ids=tuple(item_ids))


def build_profile(
    index: ItemIndex,
    liked_rows: NDArray[np.int64],
    ratings: NDArray[np.float64],
    neutral_rating: float,
) -> NDArray[np.float64]:
    """Mean-centred weighted centroid, returned as a unit vector.

    A rating below neutral_rating gets a negative weight, so a dislike pushes
    the profile away from an item instead of being ignored (Rocchio's gamma).
    """
    if liked_rows.size == 0:
        raise ColdStartError("no interactions to build a profile from")
    if liked_rows.shape != ratings.shape:
        raise ValueError("liked_rows and ratings must align")

    weights = ratings - neutral_rating
    profile = np.asarray(index.matrix[liked_rows].T @ weights).ravel()

    length = np.linalg.norm(profile)
    if length == 0.0:
        raise ColdStartError("feedback weights cancel out; profile has no direction")
    return profile / length


def recommend(
    index: ItemIndex,
    profile: NDArray[np.float64],
    seen_rows: NDArray[np.int64],
    k: int,
) -> list[tuple[str, float]]:
    """Top-k unseen items by cosine, which is a dot product on unit rows."""
    n_items = index.matrix.shape[0]
    if not 1 <= k <= n_items:
        raise ValueError(f"k must lie in [1, {n_items}], got {k}")

    scores = index.matrix @ profile                       # one sparse mat-vec
    scores[seen_rows] = -np.inf                           # never re-recommend

    # argpartition is O(n) selection; only the k survivors are then ordered.
    top = np.argpartition(scores, n_items - k)[n_items - k:]
    top = top[np.argsort(scores[top])[::-1]]
    return [(index.item_ids[row], float(scores[row])) for row in top]

"""Content-based filtering - batched users, one sparse product, float32."""

import numpy as np
from numpy.typing import NDArray
from scipy import sparse


class BatchRecommender:
    """Scores a block of users against the catalogue with sparse products.

    Items are unit rows (cosine == dot product). User profiles do not need
    normalising either: dividing a user's scores by a constant leaves their
    ranking untouched, so that norm is never computed on the hot path.
    """

    def __init__(self, unit_items: sparse.csr_matrix, k: int = 10) -> None:
        if not 1 <= k <= unit_items.shape[0]:
            raise ValueError(f"k must lie in [1, {unit_items.shape[0]}], got {k}")
        # One dtype, contiguous CSR: scipy hands the buffers to BLAS-style kernels
        # without an internal conversion, and float32 halves memory traffic.
        self._items = unit_items.astype(np.float32).tocsr()
        self._items_t = self._items.T.tocsr()          # (n_terms, n_items), built once
        self._k = k

    def recommend(
        self,
        feedback: sparse.csr_matrix,     # (n_users, n_items) mean-centred weights
        chunk: int = 2048,
    ) -> NDArray[np.int64]:
        """Top-k item rows for every user, best first."""
        n_users = feedback.shape[0]
        n_items = self._items.shape[0]
        k = self._k

        out = np.empty((n_users, k), dtype=np.int64)          # allocated once

        # Chunking bounds the dense score block at (chunk, n_items) rather than
        # (n_users, n_items), which is what makes a million-user batch possible.
        for start in range(0, n_users, chunk):
            stop = min(start + chunk, n_users)
            weights = feedback[start:stop]

            # Profiles for the whole block in one sparse-sparse product, then
            # scores in a second: (chunk, items) @ (items, terms) @ (terms, items).
            profiles = weights @ self._items
            scores = (profiles @ self._items_t).toarray()

            # Mask seen items in place using the feedback sparsity pattern.
            rows, columns = weights.nonzero()
            scores[rows, columns] = -np.inf

            partitioned = np.argpartition(scores, n_items - k, axis=1)[:, n_items - k:]
            chosen = np.take_along_axis(scores, partitioned, axis=1)
            order = np.argsort(-chosen, axis=1)               # sorts k, not n
            out[start:stop] = np.take_along_axis(partitioned, order, axis=1)

        return out