evaluation.py

from pathlib import Path
from typing import List, Tuple, Set, Dict, Iterable
import math
import time
import numpy as np
import pandas as pd
from sklearn.metrics import ndcg_score

from data_loader import load_kuairec_dataset
from content_based_filtering.feature_weighting import feature_specification_list
from content_based_filtering.content_based_filtering import (
    build_item_feature_matrix,
    compute_cosine_similarity,
)
from collaborative_filtering.svdpp import load_svdpp_artifacts, cf_get_items_score
from sasrec_recbole.sasrec_recbole import load_sasrec_artifacts, sasrec_score
from deepfm_recbole.deepfm_recbole import load_deepfm_artifacts, deepfm_score

if not hasattr(np, "long"):
    np.long = np.int64

ScoredItems = List[Tuple[int, float]]

K = 200
N_NEXT = 10
MAX_USERS = 100
LOG_EVERY_USERS = 10

NOVELTY_K = 200
NOVELTY_EPS = 1e-12

def log(msg: str) -> None:
    print(msg, flush=True)

def trainset_item_id_set(trainset) -> Set[int]:
    out: Set[int] = set()
    for inner_iid in trainset.all_items():
        raw = trainset.to_raw_iid(inner_iid)
        try:
            out.add(int(raw))
        except Exception:
            pass
    return out

def recbole_item_id_set(dataset) -> Set[int]:
    out: Set[int] = set()
    tokens = dataset.field2id_token[dataset.iid_field]
    for tok in tokens:
        try:
            out.add(int(tok))
        except Exception:
            pass
    return out

def sanitize_scores(items: ScoredItems) -> ScoredItems:
    out: ScoredItems = []
    for vid, sc in items:
        scf = float(sc)
        if not math.isfinite(scf):
            scf = -1e9
        out.append((int(vid), scf))
    return out

def minmax_norm(scores: Dict[int, float]) -> Dict[int, float]:
    if not scores:
        return {}
    vals = list(scores.values())
    mn, mx = min(vals), max(vals)
    if abs(mx - mn) < 1e-12:
        return {k: 1.0 for k in scores}
    return {k: (v - mn) / (mx - mn) for k, v in scores.items()}

def hybrid_blend_union(r1: ScoredItems, r2: ScoredItems, alpha: float = 0.5) -> ScoredItems:
    r1 = sanitize_scores(r1)
    r2 = sanitize_scores(r2)

    r1_map = {vid: s for vid, s in r1}
    r2_map = {vid: s for vid, s in r2}

    r1n = minmax_norm(r1_map)
    r2n = minmax_norm(r2_map)

    blended: ScoredItems = []
    for vid in (set(r1n) | set(r2n)):
        blended.append((vid, alpha * r1n.get(vid, 0.0) + (1.0 - alpha) * r2n.get(vid, 0.0)))

    blended.sort(key=lambda x: x[1], reverse=True)
    return blended

def sklearn_ndcg_at_k(scored_items: ScoredItems, true_item: int, k: int) -> float:
    if not scored_items:
        return 0.0

    vids: List[int] = []
    scrs: List[float] = []
    for vid, sc in scored_items:
        scf = float(sc)
        if math.isfinite(scf):
            vids.append(int(vid))
            scrs.append(scf)

    try:
        pos = vids.index(int(true_item))
    except ValueError:
        return 0.0

    rel = np.zeros((1, len(vids)), dtype=float)
    rel[0, pos] = 1.0
    scores = np.array([scrs], dtype=float)
    return float(ndcg_score(rel, scores, k=k))

def filter_seen_from_ranked(ranked: ScoredItems, seen_before: Set[int], keep_item: int) -> ScoredItems:
    keep_item = int(keep_item)
    out: ScoredItems = []
    for vid, sc in ranked:
        vid = int(vid)
        if vid == keep_item or vid not in seen_before:
            out.append((vid, float(sc)))
    return out

def iter_last_n_next_points(
    df: pd.DataFrame,
    user_id: int,
    valid_items: Set[int],
    n_next: int,
) -> Iterable[Tuple[int, int, Set[int]]]:
    rows = df[df["user_id"] == user_id]
    if rows.empty:
        return

    if "datetime" in rows.columns:
        rows = rows.sort_values("datetime")
    else:
        rows = rows.sort_values("timestamp")

    vids = rows["video_id"].astype(int).tolist()
    if len(vids) < 2:
        return

    start_i = max(1, len(vids) - n_next)
    for i in range(start_i, len(vids)):
        prev_item = vids[i - 1]
        true_next = vids[i]
        if prev_item == true_next:
            continue
        if prev_item not in valid_items or true_next not in valid_items:
            continue
        seen_before = set(vids[:i])
        yield prev_item, true_next, seen_before

def build_item_user_popularity(df: pd.DataFrame) -> Tuple[Dict[int, float], int]:
    users = df["user_id"].astype(int).unique()
    n_users = int(len(users))
    counts = df.groupby("video_id")["user_id"].nunique()
    pop = (counts / max(n_users, 1)).to_dict()
    pop = {int(k): float(v) for k, v in pop.items()}
    return pop, n_users

def novelty_of_ranked(ranked: ScoredItems, pop: Dict[int, float], topk: int) -> float:
    if not ranked:
        return 0.0
    top = ranked[: max(0, int(topk))]
    if not top:
        return 0.0
    vals: List[float] = []
    for vid, _ in top:
        p = float(pop.get(int(vid), 0.0))
        p = min(max(p, NOVELTY_EPS), 1.0)
        vals.append(-math.log(p, 2))
    return float(np.mean(vals)) if vals else 0.0

def rs1_ranked_items_union(
    user_id: int,
    prev_item: int,
    svdpp_model,
    trainset,
    item_matrix,
    vid_to_idx,
    idx_to_vid,
) -> ScoredItems:
    cb_scores = compute_cosine_similarity(
        target_video_id=int(prev_item),
        item_feature_matrix=item_matrix,
        video_id_to_matrix_index=vid_to_idx,
        matrix_index_to_video_id=idx_to_vid,
    )
    cf_scores = cf_get_items_score(svdpp_model, trainset, int(user_id))
    return hybrid_blend_union(cf_scores, cb_scores, alpha=0.5)

def evaluate_rs1(df: pd.DataFrame, video_features: pd.DataFrame, item_pop: Dict[int, float]) -> Tuple[float, float, int]:
    log("[RS1] Building item feature matrix...")
    t0 = time.time()
    item_matrix, vid_to_idx, idx_to_vid = build_item_feature_matrix(
        video_features,
        feature_specification_list,
        scale_numeric_features=True,
    )
    log(f"[RS1] Item matrix done in {time.time()-t0:.2f}s")

    log("[RS1] Loading SVD++ artifacts...")
    t0 = time.time()
    svdpp_model, trainset = load_svdpp_artifacts(Path("models/svdpp"))
    log(f"[RS1] SVD++ loaded in {time.time()-t0:.2f}s")

    valid_items = set(vid_to_idx) & trainset_item_id_set(trainset)
    users = sorted(df["user_id"].astype(int).unique().tolist())[:MAX_USERS]

    user_means_ndcg: List[float] = []
    user_means_novelty: List[float] = []
    start_all = time.time()

    for idx, user_id in enumerate(users, 1):
        per_user_ndcg: List[float] = []
        per_user_nov: List[float] = []

        for prev_item, true_next, seen_before in iter_last_n_next_points(df, user_id, valid_items, N_NEXT):
            ranked = rs1_ranked_items_union(
                user_id=user_id,
                prev_item=prev_item,
                svdpp_model=svdpp_model,
                trainset=trainset,
                item_matrix=item_matrix,
                vid_to_idx=vid_to_idx,
                idx_to_vid=idx_to_vid,
            )
            ranked = filter_seen_from_ranked(ranked, seen_before=seen_before, keep_item=true_next)

            per_user_ndcg.append(sklearn_ndcg_at_k(ranked, true_next, K))
            per_user_nov.append(novelty_of_ranked(ranked, item_pop, NOVELTY_K))

        if per_user_ndcg:
            user_means_ndcg.append(float(np.mean(per_user_ndcg)))
            user_means_novelty.append(float(np.mean(per_user_nov)))

        if idx % LOG_EVERY_USERS == 0:
            elapsed = time.time() - start_all
            log(f"[RS1] progress {idx}/{len(users)} users | users_with_points={len(user_means_ndcg)} | elapsed={elapsed:.1f}s")

    mean_ndcg_users = float(np.mean(user_means_ndcg)) if user_means_ndcg else float("nan")
    mean_novelty_users = float(np.mean(user_means_novelty)) if user_means_novelty else float("nan")
    return mean_ndcg_users, mean_novelty_users, len(user_means_ndcg)

def rs2_ranked_items_union(user_id: int, deepfm_bundle, sasrec_bundle) -> ScoredItems:
    deepfm_config, deepfm_model, deepfm_dataset, _train, _valid, deepfm_test = deepfm_bundle
    sasrec_config, sasrec_model, sasrec_dataset, _train2, _valid2, sasrec_test = sasrec_bundle

    deepfm_scores = deepfm_score(
        raw_user_id=int(user_id),
        recbole_config=deepfm_config,
        recbole_model=deepfm_model,
        recbole_dataset=deepfm_dataset,
        recbole_test_data=deepfm_test,
    )

    sasrec_scores = sasrec_score(
        raw_user_id=int(user_id),
        recbole_config=sasrec_config,
        recbole_model=sasrec_model,
        recbole_dataset=sasrec_dataset,
        recbole_test_data=sasrec_test,
    )

    return hybrid_blend_union(deepfm_scores, sasrec_scores, alpha=0.5)

def evaluate_rs2(df: pd.DataFrame, item_pop: Dict[int, float]) -> Tuple[float, float, int]:
    log("[RS2] Loading SASRec artifacts...")
    t0 = time.time()
    sasrec_bundle = load_sasrec_artifacts(Path("models/sasrec"))
    log(f"[RS2] SASRec loaded in {time.time()-t0:.2f}s")

    log("[RS2] Loading DeepFM artifacts...")
    t0 = time.time()
    deepfm_bundle = load_deepfm_artifacts(Path("models/deepfm"))
    log(f"[RS2] DeepFM loaded in {time.time()-t0:.2f}s")

    sasrec_dataset = sasrec_bundle[2]
    deepfm_dataset = deepfm_bundle[2]
    valid_items = recbole_item_id_set(sasrec_dataset) & recbole_item_id_set(deepfm_dataset)

    users = sorted(df["user_id"].astype(int).unique().tolist())[:MAX_USERS]

    user_means_ndcg: List[float] = []
    user_means_novelty: List[float] = []
    start_all = time.time()

    for idx, user_id in enumerate(users, 1):
        ranked_base = rs2_ranked_items_union(
            user_id=user_id,
            deepfm_bundle=deepfm_bundle,
            sasrec_bundle=sasrec_bundle,
        )

        per_user_ndcg: List[float] = []
        per_user_nov: List[float] = []

        for _prev_item, true_next, seen_before in iter_last_n_next_points(df, user_id, valid_items, N_NEXT):
            ranked = filter_seen_from_ranked(ranked_base, seen_before=seen_before, keep_item=true_next)

            per_user_ndcg.append(sklearn_ndcg_at_k(ranked, true_next, K))
            per_user_nov.append(novelty_of_ranked(ranked, item_pop, NOVELTY_K))

        if per_user_ndcg:
            user_means_ndcg.append(float(np.mean(per_user_ndcg)))
            user_means_novelty.append(float(np.mean(per_user_nov)))

        if idx % LOG_EVERY_USERS == 0:
            elapsed = time.time() - start_all
            log(f"[RS2] progress {idx}/{len(users)} users | users_with_points={len(user_means_ndcg)} | elapsed={elapsed:.1f}s")

    mean_ndcg_users = float(np.mean(user_means_ndcg)) if user_means_ndcg else float("nan")
    mean_novelty_users = float(np.mean(user_means_novelty)) if user_means_novelty else float("nan")
    return mean_ndcg_users, mean_novelty_users, len(user_means_ndcg)


def main():
    log("[EVAL] Loading dataset...")
    t0 = time.time()
    df, _user_features, video_features = load_kuairec_dataset()
    log(f"[EVAL] Dataset loaded in {time.time()-t0:.2f}s | rows={len(df)}")
    log(f"[EVAL] Settings: MAX_USERS={MAX_USERS}, K={K}, N_NEXT={N_NEXT}, NOVELTY_K={NOVELTY_K}")

    log("[EVAL] Precomputing item popularity for novelty...")
    item_pop, n_users = build_item_user_popularity(df)
    log(f"[EVAL] Popularity computed | users={n_users} | items_with_pop={len(item_pop)}")

    log("[EVAL] RS1 start")
    rs1_ndcg_user, rs1_novelty_user, rs1_users_eval = evaluate_rs1(df, video_features, item_pop)

    log("[EVAL] RS2 start")
    rs2_ndcg_user, rs2_novelty_user, rs2_users_eval = evaluate_rs2(df, item_pop)

    print("\n================ EVALUATION RESULTS ================")
    print(f"RS1 (Conventional Hybrid)  NDCG@{K}: {rs1_ndcg_user:.4f}  (users={rs1_users_eval})")
    print(f"RS1 (Conventional Hybrid)  Novelty@{NOVELTY_K}: {rs1_novelty_user:.4f}  (users={rs1_users_eval})")
    print(f"RS2 (Advanced Hybrid)      NDCG@{K}: {rs2_ndcg_user:.4f}  (users={rs2_users_eval})")
    print(f"RS2 (Advanced Hybrid)      Novelty@{NOVELTY_K}: {rs2_novelty_user:.4f}  (users={rs2_users_eval})")

    print("====================================================")

if __name__ == "__main__":
    main()