Engines

CoreRec provides multiple recommendation engine categories.

Deep Learning Models

Top-level models available via from corerec.engines import <Model>.

DCN (Deep & Cross Network)

class corerec.engines.dcn.DCN(name: str = 'DCN', embedding_dim: int = 16, num_cross_layers: int = 3, deep_layers: List[int] = None, dropout: float = 0.2, learning_rate: float = 0.001, batch_size: int = 256, epochs: int = 20, early_stopping_patience: int | None = 5, checkpoint_dir: str | None = None, trainable: bool = True, verbose: bool = False, device: str = 'auto', task: str = 'auto', num_negatives: int = 4)

Bases: BaseRecommender

Deep & Cross Network for Feature-rich Recommendation

Combines a cross network for explicit feature interactions with a deep neural network for implicit feature interactions. Especially effective for feature-rich recommendation scenarios.

Reference: Wang et al. “Deep & Cross Network for Ad Click Predictions” (2017)

Parameters:

• name – Model name for identification

• embedding_dim – Dimension of feature embeddings (default: 16)

• num_cross_layers – Number of cross network layers (default: 3)

• deep_layers – List of hidden layer sizes for deep network (default: [128, 64])

• dropout – Dropout rate (default: 0.2)

• learning_rate – Learning rate for optimizer (default: 0.001)

• batch_size – Training batch size (default: 256)

• epochs – Number of training epochs (default: 20)

• early_stopping_patience – Patience for early stopping, None to disable (default: 5)

• checkpoint_dir – Directory to save checkpoints, None to disable (default: None)

• trainable – Whether model is trainable (default: True)

• verbose – Whether to print training progress (default: False)

• device – Device for computation (‘cuda’ or ‘cpu’, default: auto-detect)

Example

>>> model = DCN(embedding_dim=64, epochs=20, verbose=True)
>>> model.fit(user_ids=[1,2,3], item_ids=[10,20,30], ratings=[5.0,4.0,3.0])
>>> recommendations = model.recommend(user_id=1, top_k=10)

fit(user_ids: List[int], item_ids: List[int], ratings: List[float], user_features: Dict[int, Dict[str, Any]] | None = None, item_features: Dict[int, Dict[str, Any]] | None = None) → DCN

Train the DCN model.

predict(user_id: int, item_id: int, **kwargs) → float

Predict score for a user-item pair.

recommend(user_id: int, top_k: int = 10, exclude_items: List[int] | None = None, **kwargs) → List[int]

Generate top-K recommendations for a user.

save(path: str | Path, safe: bool = True, **kwargs) → None

Save model to disk (safe bundle by default; legacy full checkpoint if safe=False).

classmethod load(path: str | Path) → DCN

Load model from disk (safe bundle or legacy checkpoint).

DeepFM

class corerec.engines.deepfm.DeepFM(name: str = 'DeepFM', embedding_dim: int = 16, hidden_layers: List[int] = [400, 400, 400], dropout: float = 0.3, learning_rate: float = 0.001, batch_size: int = 256, epochs: int = 20, trainable: bool = True, verbose: bool = False, device: str = 'auto', task: str = 'auto', num_negatives: int = 4)

Bases: BaseRecommender

Deep Factorization Machine (DeepFM)

Combines factorization machines for recommendation with deep neural networks. It jointly learns a factorization machine for recommendation and deep representations of features through a neural network.

Reference: Guo et al. “DeepFM: A Factorization-Machine based Neural Network for CTR Prediction” (IJCAI 2017)

fit(user_ids: List[int], item_ids: List[int], ratings: List[float], user_features: Dict[int, Dict[str, Any]] | None = None, item_features: Dict[int, Dict[str, Any]] | None = None, batch_size: int | None = None, **kwargs) → DeepFM

Train the DeepFM model.

Parameters:

• user_ids – List of user IDs

• item_ids – List of item IDs

• ratings – List of ratings

• user_features – Dictionary of user features

• item_features – Dictionary of item features

• batch_size – Batch size for training (overrides init param if provided)

predict(user_id: Any, item_id: Any, **kwargs) → float

Predict the probability of interaction between user and item.

Parameters:

• user_id – User ID

• item_id – Item ID

Returns:

Predicted probability of interaction

recommend(user_id: Any, top_k: int = 10, exclude_items: List[Any] | None = None, *, top_n: int | None = None, exclude_seen: bool = True, **kwargs) → List[Any]

Recommend top-K items for a user.

save(path: str | Path, safe: bool = True, **kwargs) → None

Save the model to disk.

Parameters:

• path – Path to save the model

• safe – Use corerec_safe_v1 bundle (default). Set False for legacy torch checkpoint.

classmethod load(path: str | Path, **kwargs) → DeepFM

Load the model from disk.

Parameters:

path – Path to load the model from

Returns:

Loaded DeepFM instance

TwoTower

Two-Tower architecture for efficient retrieval.

This is the industry standard for large-scale recsys. Used by YouTube, Netflix, Uber, etc.

Key idea: separate towers encode users and items independently, then match via dot product in embedding space.

class corerec.engines.two_tower.TwoTowerModel(*args: Any, **kwargs: Any)

Bases: Module

Dual encoder architecture.

User tower and item tower project inputs into shared embedding space. Similarity = dot product of embeddings.

encode_user(user_features: torch.Tensor) → torch.Tensor

Project user into embedding space.

encode_item(item_features: torch.Tensor) → torch.Tensor

Project item into embedding space.

forward(user_features: torch.Tensor, item_features: torch.Tensor) → torch.Tensor

Compute similarity between users and items.

user_features: [batch_size, user_dim] item_features: [batch_size, item_dim] or [num_items, item_dim]

Returns: similarity scores [batch_size] or [batch_size, num_items]

batch_score(user_emb: torch.Tensor, item_embs: torch.Tensor) → torch.Tensor

Score a user against multiple items efficiently.

class corerec.engines.two_tower.TwoTower(name: str = 'TwoTower', user_input_dim: int = 64, item_input_dim: int = 64, embedding_dim: int = 128, hidden_dims: List[int] = [256, 128], dropout: float = 0.1, loss_type: str = 'bce', learning_rate: float = 0.001, batch_size: int = 256, epochs: int = 10, device: torch.device | None = None, negative_samples: int = 4, temperature: float = 0.07, verbose: bool = True, num_epochs: int | None = None)

Bases: BaseRecommender

Two-Tower recommender with training logic.

Supports various loss functions: - pointwise: BCE on (user, item) pairs - pairwise: BPR-style ranking loss - contrastive: InfoNCE (good for in-batch negatives)

fit(user_ids: List, item_ids: List, interactions: ndarray | None = None, user_features: ndarray | None = None, item_features: ndarray | None = None, validation_split: float = 0.1, ratings: ndarray | None = None)

Train the two-tower model.

Accepts either:

fit(user_ids, item_ids, interactions) # [n_users, n_items] matrix fit(user_ids, item_ids, ratings) # one entry per interaction

ratings= is an alias for the third arg — rest of the zoo uses that name and callers building kwargs were dying on TypeError before.

user_features: optional [n_users, user_dim] feature matrix item_features: optional [n_items, item_dim] feature matrix

recommend(user_id: Any, top_k: int = 10, exclude_items: List[Any] | None = None, exclude_seen: bool = True, **kwargs) → List[Any]

Generate recommendations for a user.

exclude_items: item IDs to keep out of the results exclude_seen: also drop items this user interacted with during fit

Returns list of item IDs ranked by score.

get_user_embedding(user_id: Any) → ndarray

Get embedding vector for a user.

predict(user_id: Any, item_id: Any, **kwargs) → float

Predict affinity score for a single user-item pair.

save(path: str | Path, safe: bool = True, **kwargs) → None

Save model to disk.

classmethod load(path: str | Path) → TwoTower

Load model from disk.

get_item_embeddings() → ndarray

Get all item embeddings (for building vector index).

SASRec

class corerec.engines.sasrec.PointWiseFeedForward(*args: Any, **kwargs: Any)

Bases: Module

Point-wise feed-forward network for SASRec.

forward(inputs)

class corerec.engines.sasrec.SASRecModel(*args: Any, **kwargs: Any)

Bases: Module

Self-Attentive Sequential Recommendation model (SASRec).

forward(input_seqs: torch.LongTensor, padding_mask: torch.BoolTensor | None = None, return_attention_weights: bool = False)

input_seqs: LongTensor of shape [batch_size, seq_len] padding_mask: BoolTensor of shape [batch_size, seq_len] where True indicates padding positions

class corerec.engines.sasrec.SASRec(name: str = 'SASRec', hidden_units: int = 64, num_blocks: int = 2, num_heads: int = 1, dropout_rate: float = 0.1, max_seq_length: int = 50, position_encoding: str = 'learned', attention_type: str = 'causal', activation: str = 'gelu', device: torch.device | None = None, learning_rate: float = 0.001, l2_reg: float = 1e-06, batch_size: int = 128, epochs: int = 10, neg_samples: int = 1, loss_type: str = 'bce', early_stopping_patience: int = 3, save_checkpoints: bool = False, checkpoint_dir: str = './checkpoints', export_embeddings: bool = False, item_popularity_bias: bool = False, user_cooling: bool = False, log_interval: int = 100, verbose: bool = True, num_epochs: int | None = None)

Bases: BaseRecommender

SASRec wrapper that ties the SASRecModel to training, evaluation and I/O helpers.

classmethod load(path: str | Path, device: torch.device | None = None)

Load model from disk.

This method must be implemented by all subclasses.

Parameters:

path – File path to load model from

Returns:

Loaded model instance

Example

model = NCF.load(‘models/ncf_model.pkl’)

Author: Vishesh Yadav (mail: sciencely98@gmail.com)

save(path: str | Path, safe: bool = True)

Save model to disk.

This method must be implemented by all subclasses.

Parameters:

• path – File path to save model

• **kwargs – Additional save parameters

Example

model.save(‘models/ncf_model.pkl’)

Author: Vishesh Yadav (mail: sciencely98@gmail.com)

get_item_embeddings() → ndarray

export_item_embeddings(filepath: str | None = None)

predict_scores(last_emb: torch.Tensor, item_indices: torch.LongTensor | None = None) → torch.Tensor

Compute scores for items given last_emb [batch_size, hidden]. If item_indices is None -> return scores over all items [batch_size, n_items+1] Otherwise return scores for those indices [batch_size, len(item_indices)]

predict(user_id: Any, item_id: Any, **kwargs) → float

Predict score for a single user-item pair (BaseRecommender interface).

fit(arg1: List[Any] | ndarray | Any = None, arg2: List[Any] | ndarray | None = None, arg3: List[Any] | ndarray | None = None, validation_data: Dict[Any, Tuple[List[int], List[int]]] | None = None, item_embedding_init: ndarray | None = None, user_item_timestamps: Dict[Any, List[Tuple[Any, Any]]] | None = None, user_ids: List[Any] | None = None, item_ids: List[Any] | None = None, ratings: List[Any] | ndarray | None = None, interactions: List[Any] | ndarray | None = None, interaction_matrix: List[Any] | ndarray | None = None, **kwargs)

Fit SASRec model.

Calling conventions (same as the rest of the zoo where possible): 1. fit(interaction_matrix, user_ids, item_ids) # legacy matrix-first 2. fit(user_ids, item_ids, interaction_matrix) # dense [n_users, n_items] 3. fit(user_ids, item_ids, ratings) # one row per event 4. fit(user_ids=…, item_ids=…, ratings=…) # keyword form

interactions= / interaction_matrix= also accepted as aliases for the third slot / matrix form.

recommend(user_id: Any, top_k: int = 10, exclude_items: List[Any] | None = None, *, top_n: int | None = None, exclude_seen: bool = True, **kwargs) → List[Any]

Recommend top-K items for a user.

evaluate(eval_data: Dict[Any, Tuple[List[int], List[int]]], metrics: List[str] | None = None, cutoffs: List[int] | None = None) → Dict[str, float]

Evaluate model on eval_data: dict {user_id: (input_seq, ground_truth_list)} metrics: list e.g. [“hit”, “ndcg”, “precision”, “recall”] cutoffs: e.g. [5, 10]

Collaborative Filtering

Available via from corerec.engines.collaborative import <Model> or from corerec import collaborative.

SAR (Simple Algorithm for Recommendation)

Simple Algorithm for Recommendation (SAR)

A fast neighborhood-based collaborative filter using item-item similarity. Works well on sparse implicit feedback data without needing gradient descent.

The basic idea: 1. Build a user-item affinity matrix (possibly time-weighted) 2. Compute item-item similarity from co-occurrence patterns 3. Score items for a user by combining their history with similarity

This impl supports multiple similarity metrics, time decay, thresholding, and score normalization. It can handle DataFrame or list-based input.

Reference: Microsoft Recommenders SAR implementation

class corerec.engines.collaborative.sar.SAR(col_user: str = 'userID', col_item: str = 'itemID', col_rating: str = 'rating', col_timestamp: str = 'timestamp', col_prediction: str = 'prediction', similarity_type: str = 'jaccard', time_decay_coefficient: float = 30.0, time_now: int | None = None, timedecay_formula: bool = False, threshold: int = 1, normalize: bool = False)

Bases: BaseRecommender

Simple Algorithm for Recommendation.

A fast item-based collaborative filter that: - Uses co-occurrence to estimate item similarity - Applies time decay to weight recent interactions higher - Can normalize predictions to original rating scale - Supports multiple similarity metrics

Good for: - Quick baselines without deep learning - Implicit feedback (views, clicks, purchases) - Situations where interpretability matters

Not ideal for: - Very sparse data with few interactions per item - Cold start (needs some user history) - When item/user features are important

Example

>>> model = SAR(similarity_type='jaccard')
>>> model.fit(train_df)
>>> recs = model.recommend_k_items(test_df, top_k=10)

__init__(col_user: str = 'userID', col_item: str = 'itemID', col_rating: str = 'rating', col_timestamp: str = 'timestamp', col_prediction: str = 'prediction', similarity_type: str = 'jaccard', time_decay_coefficient: float = 30.0, time_now: int | None = None, timedecay_formula: bool = False, threshold: int = 1, normalize: bool = False)

Initialize SAR model.

Parameters:

• col_user – name of user column in input data

• col_item – name of item column

• col_rating – name of rating/weight column

• col_timestamp – name of timestamp column (needed if timedecay_formula=True)

• col_prediction – name for prediction column in output

• similarity_type – one of ‘cooccurrence’, ‘cosine’, ‘jaccard’, ‘lift’, ‘inclusion_index’, ‘mutual_information’, ‘lexicographers_mi’. ‘lift’, ‘inclusion_index’ and ‘mutual_information’ need a meaningfully raised threshold to be usable – at the default threshold=1 they are dominated by coincidental co-occurrences between rare items and score close to random. ‘jaccard’ and ‘cosine’ are bounded and don’t have this failure mode; prefer them unless you have a specific reason to want lift’s or PMI’s semantics.

• time_decay_coefficient – half-life in days for time decay (default 30 days)

• time_now – reference time for decay calc (defaults to max timestamp in data)

• timedecay_formula – whether to apply time decay weighting

• threshold – minimum co-occurrence count to keep (filters noise)

• normalize – whether to rescale predictions to original rating range

fit(df: DataFrame) → SAR

Train the SAR model on interaction data.

Parameters:

df – DataFrame with user, item, rating columns. Must not have duplicate (user, item) pairs. If timedecay_formula=True, needs timestamp column too.

Returns:

self (for method chaining)

Raises:

InvalidDataError – if input has issues

score(test: DataFrame, remove_seen: bool = False) → ndarray

Score all items for users in the test set.

Parameters:

• test – DataFrame with user column (items are scored for all users in test)

• remove_seen – set scores of seen items to -inf

Returns:

2D array of shape (n_test_users, n_items) with scores

recommend_k_items(test: DataFrame, top_k: int = 10, sort_top_k: bool = True, remove_seen: bool = False) → DataFrame

Get top-k recommendations for users in test set.

Parameters:

• test – DataFrame with user column

• top_k – number of items to recommend per user

• sort_top_k – whether to sort results by score

• remove_seen – exclude items user interacted with in training

Returns:

DataFrame with columns [user, item, prediction]

recommend(user_id: Any, top_k: int = 10, exclude_items: List[Any] | None = None, **kwargs) → List[Any]

Get top-k recommendations for a single user.

This is the BaseRecommender interface method. For batch recommendations, use recommend_k_items() which is more efficient.

Parameters:

• user_id – user to recommend for

• top_k – number of items

• exclude_items – additional items to exclude (beyond seen items)

Returns:

list of recommended item IDs

predict(user_id: Any, item_id: Any, **kwargs) → float

Predict score for a single user-item pair.

Parameters:

• user_id – user identifier

• item_id – item identifier

Returns:

predicted affinity score

predict_batch(test: DataFrame) → DataFrame

Predict scores for specific user-item pairs in test.

Parameters:

test – DataFrame with user and item columns

Returns:

DataFrame with user, item, prediction columns

get_popularity_based_topk(top_k: int = 10, sort_top_k: bool = True, items: bool = True) → DataFrame

Get most popular items or users.

Useful as a fallback for cold users or as a simple baseline.

Parameters:

• top_k – number of items/users to return

• sort_top_k – whether to sort by popularity

• items – if True return items, if False return users

Returns:

DataFrame with item/user and prediction (popularity count)

get_item_based_topk(items: DataFrame, top_k: int = 10, sort_top_k: bool = True) → DataFrame

Get recommendations based on seed items.

Useful for cold users - provide some items they like and get similar item recommendations without needing training data.

Parameters:

• items – DataFrame with item column (and optionally user, rating)

• top_k – number of recommendations

• sort_top_k – whether to sort by score

Returns:

DataFrame with user (if provided), item, prediction

get_topk_most_similar_users(user: Any, top_k: int = 10, sort_top_k: bool = True) → DataFrame

Find users most similar to the given user.

Similarity is based on affinity vectors (what items they interacted with).

Parameters:

• user – user ID to find similar users for

• top_k – number of similar users

• sort_top_k – whether to sort by similarity

Returns:

DataFrame with user and prediction (similarity score)

save(path: str, safe: bool = True, **kwargs) → None

Save model to disk (safe npz bundle by default).

classmethod load(path: str) → SAR

Load model from disk (safe bundle or legacy pickle).

fit_from_lists(user_ids: List[Any], item_ids: List[Any], ratings: List[float], timestamps: List[int] | None = None) → SAR

Fit model from separate lists instead of DataFrame.

This is for compatibility with older code that passes lists. Internally converts to DataFrame and calls fit().

Parameters:

• user_ids – list of user IDs

• item_ids – list of item IDs

• ratings – list of ratings/weights

• timestamps – optional list of timestamps

Returns:

self

Content-Based Filtering

Available via from corerec.engines.content_based import <Model> or from corerec import content_based.

TFIDFRecommender

class corerec.engines.content_based.TFIDFRecommender(feature_matrix=None, name: str | None = None, verbose: bool = False)

Bases: BaseRecommender

TF-IDF based content recommender.

Uses TF-IDF vectorization to find similar items based on text content. Can recommend items by text query or by item similarity.

Supports two initialization modes: 1. Old API: TFIDFRecommender(feature_matrix) - for backward compatibility 2. New API: TFIDFRecommender() then fit(items, docs)

property similarity_matrix

Get or compute the similarity matrix.

fit(items: List[Any], docs: Dict[Any, str], **kwargs) → TFIDFRecommender

Fit the TF-IDF model on item documents.

Parameters:

• items – List of item IDs

• docs – Dictionary mapping item IDs to text documents

• **kwargs – Additional arguments (unused)

Returns:

Self for method chaining

predict(user_id: Any, item_id: Any, **kwargs) → float

Predict relevance score for a user-item pair.

For TF-IDF, this returns a default score since we don’t have user preferences. Consider using recommend_by_text or recommend instead.

Parameters:

• user_id – User identifier (unused for content-based)

• item_id – Item identifier

• **kwargs – Additional arguments (unused)

Returns:

Default score of 0.5 (neutral)

recommend(user_id_or_indices: Any = None, top_k: int = 10, top_n: int = None, exclude_items: List[Any] | None = None, **kwargs) → List[Any] | ndarray

Generate recommendations.

Supports two modes: 1. Old API: recommend(item_indices, top_n) - returns numpy array of indices 2. New API: recommend(user_id, top_k) - returns list of item IDs

Parameters:

• user_id_or_indices – User ID (new API) or list of item indices (old API)

• top_k – Number of recommendations (new API, alias for top_n)

• top_n – Number of recommendations (old API, takes precedence)

• exclude_items – Items to exclude (new API only)

• **kwargs – Additional arguments (unused)

Returns:

List of recommended item IDs (new API) or numpy array of indices (old API)

recommend_by_text(query_text: str, top_k: int = 10, top_n: int | None = None) → List[Any]

Recommend items based on text query using TF-IDF similarity.

Parameters:

• query_text – Text query to find similar items

• top_k – Number of recommendations to return

• top_n – old name for top_k, kept for existing callers

Returns:

List of recommended item IDs sorted by relevance

save(path: str | Path, safe: bool = True, **kwargs) → None

Save model to disk.

Parameters:

• path – File path to save model

• safe – Use corerec_safe_v1 bundle (default). Set False for legacy pickle.

• **kwargs – Additional arguments (unused)

classmethod load(path: str | Path, **kwargs) → TFIDFRecommender

Load model from disk.

Parameters:

• path – File path to load model from

• **kwargs – Additional arguments (unused)

Returns:

Loaded TFIDFRecommender instance