01 · 数据、特征与 ItemCF

from pathlib import Path
import sys
import inspect
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from IPython.display import display, Code


PROJECT = next(p for p in [Path.cwd(), *Path.cwd().parents]
               if (p / "reclab/models.py").exists())
if str(PROJECT) not in sys.path:
    sys.path.insert(0, str(PROJECT))
from reclab.data import Dataset
from reclab.paths import ARTIFACTS, PROCESSED, read_json


plt.rcParams.update({"figure.figsize": (9, 3.8), "font.size": 11,
    "axes.spines.top": False, "axes.spines.right": False,
    "axes.labelcolor": "#182b2b", "text.color": "#182b2b",
    "axes.facecolor": "white", "figure.facecolor": "white",
    "grid.color": "#d4ddda", "axes.axisbelow": True})
dataset = Dataset()
print("Project:", PROJECT)
print("Dataset version:", dataset.manifest["dataset_version"])
print("Python:", sys.executable)
assert Path(sys.executable).parent == PROJECT / ".venv/bin", "请选择 Recommendation Lab 内核"

Project: /Users/lizuolin/local/recommendation-lab
Dataset version: 2772cab0aca8cb81
Python: /Users/lizuolin/local/recommendation-lab/.venv/bin/python

m = dataset.manifest
display(pd.Series({k: m[k] for k in ["raw_events", "cohort_users", "catalog_items", "cohort_events", "feature_version"]}, name="value").to_frame())
boundaries = pd.Series({"start": m["start_ms"], "train_end": m["train_end_ms"], "valid_end": m["valid_end_ms"], "end": m["end_ms"]})
display(pd.to_datetime(boundaries, unit="ms", utc=True).to_frame("UTC boundary"))
display(dataset.events.head(5))

import duckdb
with duckdb.connect(":memory:") as con:
    split_summary = con.execute("""
        SELECT split, count(*) AS events,
               count(DISTINCT visitorid) AS users,
               count(*) FILTER (WHERE item_idx = 0) AS oov_events
        FROM read_parquet(?) GROUP BY split ORDER BY split
    """, [str(PROCESSED / "events.parquet")]).df()
display(split_summary)

USER_POSITION = 0
HISTORY_WINDOW = 20
TOP_K = 10
eligible, audit = dataset.evaluation("valid", 1000)
user_idx = list(eligible)[USER_POSITION]
history = dataset.histories[user_idx]
user_events = dataset.train[dataset.train.user_idx == user_idx].sort_values("timestamp", kind="stable")
print("User ID:", dataset.users[user_idx], "User index:", user_idx)
print("Training-only history:", history[-HISTORY_WINDOW:])
preview = user_events.tail(HISTORY_WINDOW)[["timestamp", "event", "itemid", "item_idx"]].copy()
preview["timestamp"] = pd.to_datetime(preview.timestamp, unit="ms", utc=True)
display(preview)
assert user_events.timestamp.max() < m["train_end_ms"]

User ID: 1055524 User index: 5071
Training-only history: [4008, 2088, 2633, 3949, 211, 211, 2913]

display(Code(inspect.getsource(Dataset.examples), language="python"))
sample_users, sample_history, sample_targets, sample_times = dataset.examples(HISTORY_WINDOW)
selected = np.flatnonzero(sample_users == user_idx)
assert len(selected), "当前用户没有可用训练样本，请换 USER_POSITION"
j = int(selected[0])
context = sample_history[j][sample_history[j] != 0].tolist()
target = int(sample_targets[j])
prior = user_events[user_events.timestamp < sample_times[j]]
display(pd.DataFrame([{"user_idx": user_idx, "history": context,
    "target_item_idx": target, "target_time_utc": pd.to_datetime(sample_times[j], unit="ms", utc=True)}]))
assert context == prior.item_idx.astype(int).tail(HISTORY_WINDOW).tolist()
assert target not in set(prior.item_idx)

    def examples(self, history_len=20):
        users, histories, targets, times = [], [], [], []
        for user, group in self.train.groupby("user_idx", sort=True):
            past, seen, last_time = [], set(), -1
            for row in group.sort_values("timestamp", kind="stable").itertuples():
                item = int(row.item_idx)
                if (
                    len(seen) >= 2
                    and item not in seen
                    and last_time < int(row.timestamp)
                ):
                    h = past[-history_len:]
                    users.append(int(user))
                    histories.append([0] * (history_len - len(h)) + h)
                    targets.append(item)
                    times.append(int(row.timestamp))
                past.append(item)
                seen.add(item)
                last_time = int(row.timestamp)
        return (
            np.array(users, dtype=np.int64),
            np.array(histories, dtype=np.int64),
            np.array(targets, dtype=np.int64),
            np.array(times, dtype=np.int64),
        )

from scipy import sparse
from reclab.retrieval import itemcf, cf_scores, topk
display(Code(inspect.getsource(itemcf), language="python"))
similarity = sparse.load_npz(ARTIFACTS / "baselines/itemcf.npz")
scores = cf_scores(similarity, history, dataset.n_items)
recommendations = topk(scores, TOP_K, dataset.seen[user_idx])
display(pd.DataFrame([{**dataset.items[i-1], "itemcf_score": float(scores[i])}
                      for i in recommendations]))
assert not set(recommendations) & dataset.seen[user_idx]

def itemcf(dataset):
    pairs = dataset.train[["user_idx", "item_idx"]].drop_duplicates()
    matrix = sparse.csr_matrix(
        (np.ones(len(pairs), dtype=np.float32), (pairs.user_idx, pairs.item_idx)),
        shape=(dataset.n_users, dataset.n_items),
    )
    # Inverse user activity reduces extremely active users' co-occurrence contribution.
    activity = np.asarray(matrix.sum(1)).ravel()
    weighted = sparse.diags(1 / np.log1p(np.maximum(activity, 1))) @ matrix
    co = (matrix.T @ weighted).tocsr()
    frequency = np.asarray(matrix.sum(0)).ravel()
    norm = sparse.diags(1 / np.sqrt(np.maximum(frequency, 1)))
    sim = (norm @ co @ norm).tocsr()
    sim.setdiag(0)
    sim.eliminate_zeros()
    # Keep 100 neighbors per item, bounded memory and easy inspection.
    for i in range(sim.shape[0]):
        a, b = sim.indptr[i : i + 2]
        if b - a > 100:
            loc = np.argpartition(sim.data[a:b], -100)[:-100]
            sim.data[a + loc] = 0
    sim.eliminate_zeros()
    return sim

chosen_item = recommendations[0]
recent = history[-50:]
recency_weights = np.geomspace(0.2, 1.0, len(recent))
contributions = pd.DataFrame([{"history_item_idx": i,
    "history_item_id": dataset.items[i-1]["item_id"],
    "similarity": float(similarity[i, chosen_item]), "weight": float(w),
    "contribution": float(similarity[i, chosen_item]) * w}
    for i, w in zip(recent, recency_weights)])
display(contributions.sort_values("contribution", ascending=False).head(10))
print("Sum of contributions:", contributions.contribution.sum())
print("Saved scorer output:", float(scores[chosen_item]))
assert np.isclose(contributions.contribution.sum(), scores[chosen_item], atol=1e-6)

Sum of contributions: 0.14533540791386457
Saved scorer output: 0.14533540606498718

baseline = read_json(ARTIFACTS / "baselines/metrics.json")
result = pd.DataFrame([{"model": name, "Recall@20": row["metrics"]["recall@20"],
    "NDCG@20": row["metrics"]["ndcg@20"], "users": row["audit"]["evaluated_users"]}
    for name, row in baseline["models"].items()]).set_index("model")
display(result)
ax = (result["Recall@20"] * 100).plot.barh(color="#006e60", figsize=(8, 3))
ax.set(xlabel="Recall@20 (%)", ylabel="Model",
       title=f"Frozen validation / {int(result.users.iloc[0])} eligible users")
ax.set_xlim(left=0)
for y, value in enumerate(result["Recall@20"] * 100):
    ax.text(value + 0.2, y, f"{value:.2f}%", va="center")
ax.set_xlim(0, max(result["Recall@20"] * 100) * 1.2)
plt.tight_layout(); plt.show()

Notebook 保存的输出 · 本页未执行

	Recall@20	NDCG@20	users
model
popular	0.020754	0.017910	339
itemcf	0.145167	0.079874	339

<Figure size 800x300 with 1 Axes>