""" Script: error_analysis.py Identifies poor recommendations (mispredictions) for the deep model and exports a detailed CSV for report writing. For each of the 5+ selected users: - User interaction history (titles of positively-rated items) - Model's predicted top-K items - Actual relevant item from test set - Model score for the relevant item - Rank of the relevant item in the predicted list - Suggested root cause Output: data/outputs/error_analysis.csv Usage: python scripts/error_analysis.py python scripts/error_analysis.py --debug --n-examples 5 """ from __future__ import annotations import argparse import pickle import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) import numpy as np import pandas as pd from src.config import MODELS_DIR, OUTPUTS_DIR, ensure_dirs, get_config from src.models.deep import TwoTowerModel, TwoTowerTrainer, build_item_feature_matrix from src.utils.logging import get_logger logger = get_logger(__name__) ITEM_META_COLS = [ "item_avg_rating", "item_num_ratings", "item_popularity", "item_positive_rate", "price_norm", "category_idx", "brand_idx", ] def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Export error analysis for the deep model.") parser.add_argument("--debug", action="store_true") parser.add_argument("--n-examples", type=int, default=5, help="Number of misprediction examples to export.") parser.add_argument("--top-k", type=int, default=20) return parser.parse_args() def find_worst_users( test_df: pd.DataFrame, trainer: TwoTowerTrainer, item_embeddings: np.ndarray, user_seen: dict[int, set[int]], top_k: int, n_examples: int, ) -> list[dict]: """ Find users where the model fails most (high rank of relevant item). Args: test_df: Test split (one positive per user). trainer: Fitted TwoTowerTrainer. item_embeddings: Pre-computed item embeddings. user_seen: Per-user seen item sets from training. top_k: Recommendation depth. n_examples: Number of worst-case examples to return. Returns: List of dicts with user analysis info. """ gt_map = test_df.groupby("user_idx")["item_idx"].apply(list).to_dict() records = [] for uid, gt_items in gt_map.items(): try: user_emb = trainer.get_user_embedding(uid) scores = item_embeddings @ user_emb exclude = user_seen.get(uid, set()) for i in exclude: if 0 <= i < len(scores): scores[i] = -np.inf ranked_items = np.argsort(scores)[::-1] # Find rank of the ground-truth item gt_item = gt_items[0] rank = int(np.where(ranked_items == gt_item)[0][0]) + 1 # 1-indexed gt_score = float(scores[gt_item]) top_k_items = ranked_items[:top_k].tolist() top_k_scores = scores[top_k_items].tolist() records.append({ "user_idx": uid, "gt_item": gt_item, "gt_rank": rank, "gt_score": gt_score, "top_k_items": top_k_items, "top_k_scores": top_k_scores, "missed": gt_item not in top_k_items, }) except Exception as exc: logger.warning("Error for user %d: %s", uid, exc) # Sort by rank descending (worst first) records.sort(key=lambda r: r["gt_rank"], reverse=True) return records[:n_examples] def build_analysis_rows( records: list[dict], train_df: pd.DataFrame, item_features: pd.DataFrame, item_index: dict, ) -> pd.DataFrame: """ Format misprediction records into a human-readable DataFrame. Args: records: Output of find_worst_users. train_df: Training interactions for user history. item_features: Item feature DataFrame. item_index: Dict item_idx → {title, category, brand}. Returns: DataFrame with one row per misprediction example. """ iid_to_title = dict(zip(item_features["item_idx"], item_features.get("title", pd.Series()))) iid_to_cat = dict(zip(item_features["item_idx"], item_features.get("category", pd.Series()))) rows = [] for i, rec in enumerate(records, start=1): uid = rec["user_idx"] # User history: titles of positively-rated items in training user_hist = train_df[ (train_df["user_idx"] == uid) & (train_df.get("label", train_df["rating"]) >= 1) ]["item_idx"].tolist() history_titles = [str(iid_to_title.get(iid, f"item_{iid}")) for iid in user_hist[:5]] # Ground truth item gt_iid = rec["gt_item"] gt_title = str(iid_to_title.get(gt_iid, f"item_{gt_iid}")) gt_category = str(iid_to_cat.get(gt_iid, "Unknown")) # Top-K predicted items top_predicted = [ str(iid_to_title.get(iid, f"item_{iid}")) for iid in rec["top_k_items"][:5] ] top_categories = [ str(iid_to_cat.get(iid, "Unknown")) for iid in rec["top_k_items"][:5] ] # Simple root cause heuristic if rec["gt_rank"] > 100: root_cause = "Popularity bias: item is niche/low-interaction and overshadowed by popular items" elif gt_category not in top_categories: root_cause = f"Category mismatch: user history does not strongly signal '{gt_category}'" else: root_cause = "Embedding space does not capture fine-grained preference within category" rows.append({ "example_id": i, "user_idx": uid, "user_history_sample": " | ".join(history_titles), "gt_item_title": gt_title, "gt_item_category": gt_category, "gt_rank_in_model": rec["gt_rank"], "gt_model_score": round(rec["gt_score"], 4), "top_5_predicted_titles": " | ".join(top_predicted), "top_5_predicted_categories": " | ".join(top_categories), "root_cause": root_cause, "proposed_mitigation": ( "Increase diversity penalty in candidate generation; " "consider popularity-debiasing in training loss" if "popularity" in root_cause.lower() else "Add stronger category-affinity signal to item tower features" ), }) return pd.DataFrame(rows) def main() -> None: args = parse_args() cfg = get_config(debug=args.debug) ensure_dirs(cfg) import torch device = "cuda" if torch.cuda.is_available() else "cpu" logger.info("Loading data…") train_df = pd.read_parquet(cfg.data.train_path) test_df = pd.read_parquet(cfg.data.test_path) item_features = pd.read_parquet(cfg.data.item_features_path) n_users = int(train_df["user_idx"].max() + 1) n_items = int(item_features["item_idx"].max() + 1) item_feat_matrix = build_item_feature_matrix(item_features, ITEM_META_COLS) n_meta = item_feat_matrix.shape[1] logger.info("Loading Two-Tower model…") from src.models.deep import TwoTowerModel, TwoTowerTrainer model = TwoTowerModel( n_users=n_users, n_items=n_items, embedding_dim=cfg.deep.embedding_dim, output_dim=cfg.deep.embedding_dim, hidden_dims=cfg.deep.hidden_dims, dropout=cfg.deep.dropout, n_meta_features=n_meta, ) model.load_state_dict(torch.load(cfg.deep.model_path, map_location=device)) trainer = TwoTowerTrainer(model=model, device=device, model_path=cfg.deep.model_path) item_embeddings = np.load(cfg.deep.item_embeddings_path) with open(cfg.deep.item_index_path, "rb") as f: item_index = pickle.load(f) user_seen = train_df.groupby("user_idx")["item_idx"].apply(set).to_dict() logger.info("Finding %d worst-case mispredictions…", args.n_examples) records = find_worst_users( test_df, trainer, item_embeddings, user_seen, args.top_k, args.n_examples ) analysis_df = build_analysis_rows(records, train_df, item_features, item_index) output_path = OUTPUTS_DIR / "error_analysis.csv" analysis_df.to_csv(output_path, index=False) logger.info("Error analysis saved → %s", output_path) print(analysis_df[["example_id", "gt_item_title", "gt_rank_in_model", "root_cause"]].to_string(index=False)) if __name__ == "__main__": main()