Turning Implicit Feedback into Labels: Click, Dwell, and Multi-Signal Aggregation

# Hu-Koren-Volinsky 2008 implicit ALS — manuel implementasyon
import numpy as np
from scipy.sparse import csr_matrix
 
def implicit_als(
    R: csr_matrix,
    factors: int = 64,
    alpha: float = 40.0,
    reg: float = 0.1,
    iterations: int = 15,
    random_state: int = 42,
) -> tuple[np.ndarray, np.ndarray]:
    """
    Hu-Koren-Volinsky 2008 — Implicit ALS sıfırdan.
 
    R: sparse interaction matrix (entries = etkileşim sayısı)
 
    Returns:
        P: (n_users, factors) — user embedding
        Q: (n_items, factors) — item embedding
    """
    np.random.seed(random_state)
    n_users, n_items = R.shape
 
    # Confidence matrix: c_ui = 1 + alpha * r_ui
    # Preference: p_ui = 1 if r > 0 else 0
    C = R.copy() * alpha
    C.data += 1  # 1 + alpha * r
    # NOT: C sadece pozitif girişleri tutuyor. Negative pair'lar implicit olarak c=1.
 
    # Random init
    P = np.random.normal(0, 0.01, (n_users, factors)).astype(np.float32)
    Q = np.random.normal(0, 0.01, (n_items, factors)).astype(np.float32)
 
    I_k = reg * np.eye(factors, dtype=np.float32)
 
    for it in range(iterations):
        # P'yi güncelle (Q fixed)
        QtQ = Q.T @ Q  # (k, k) — precompute
        for u in range(n_users):
            # User u'nun positive items'ı
            start, end = R.indptr[u], R.indptr[u + 1]
            pos_items = R.indices[start:end]
            confidences = 1 + alpha * R.data[start:end]
 
            # (Cu - I) Q'Q + Q'Q + reg I = Q'Q + Q'(Cu - I)Q + reg I
            # Q'(Cu - I)Q = Σ (c_ui - 1) q_i q_i^T  (only over positive items)
            Cu_minus_I_QtQ = ((confidences - 1)[:, None] * Q[pos_items]).T @ Q[pos_items]
            A = QtQ + Cu_minus_I_QtQ + I_k
 
            # Q'Cu p_u = Σ c_ui · 1 · q_i  (only over positive items)
            b = (confidences[:, None] * Q[pos_items]).sum(axis=0)
 
            P[u] = np.linalg.solve(A, b)
 
        # Q'yu güncelle (P fixed) — simetrik
        PtP = P.T @ P
        Rt = R.T.tocsr()  # transpoze, item-major
        for i in range(n_items):
            start, end = Rt.indptr[i], Rt.indptr[i + 1]
            pos_users = Rt.indices[start:end]
            confidences = 1 + alpha * Rt.data[start:end]
 
            Cu_minus_I_PtP = ((confidences - 1)[:, None] * P[pos_users]).T @ P[pos_users]
            A = PtP + Cu_minus_I_PtP + I_k
            b = (confidences[:, None] * P[pos_users]).sum(axis=0)
 
            Q[i] = np.linalg.solve(A, b)
 
        if it % 3 == 0:
            print(f"Iter {it+1}/{iterations}")
 
    return P, Q
 
 
# Smoke test (50 user x 100 item dummy)
from scipy.sparse import random as sparse_random
R_test = sparse_random(50, 100, density=0.1, format="csr", dtype=np.float32) * 10
R_test.data = np.round(R_test.data).astype(np.float32)
P, Q = implicit_als(R_test, factors=16, iterations=5)
print(f"P shape: {P.shape}, Q shape: {Q.shape}")
print(f"User 0 top item: {(P[0] @ Q.T).argmax()}")
 

# data/multi_signal.py — Multi-signal aggregation
import polars as pl
 
# Funnel-aware ağırlıklar
SIGNAL_WEIGHTS = {
    "view": 1.0,
    "addtocart": 5.0,
    "transaction": 20.0,
}
STAGE_ORDER = ["view", "addtocart", "transaction"]
STAGE_RANK = {s: i for i, s in enumerate(STAGE_ORDER)}
 
 
def aggregate_funnel_aware(events: pl.DataFrame) -> pl.DataFrame:
    """
    Funnel-aware aggregation: aynı user-item için max funnel stage.
 
    Returns DataFrame with [user_id, item_id, signal_value]
    """
    # Her event'e numeric rank ekle
    events_ranked = events.with_columns(
        pl.col("event").replace(STAGE_RANK).cast(pl.Int8).alias("stage_rank")
    )
 
    # User-item bazında max rank al
    agg = (
        events_ranked.group_by(["visitorid", "itemid"])
        .agg([
            pl.col("stage_rank").max().alias("max_stage"),
            pl.col("timestamp").max().alias("last_timestamp"),
        ])
    )
 
    # Rank'i ağırlığa çevir
    weight_map = {STAGE_RANK[s]: w for s, w in SIGNAL_WEIGHTS.items()}
    agg = agg.with_columns(
        pl.col("max_stage").replace(weight_map).cast(pl.Float32).alias("signal_value")
    )
 
    return agg.select(["visitorid", "itemid", "signal_value", "last_timestamp"])
 
 
def aggregate_log_linear(events: pl.DataFrame) -> pl.DataFrame:
    """
    Log-linear aggregation: sum(weight * log1p(count)) per event type.
    """
    import numpy as np
    counts = (
        events.group_by(["visitorid", "itemid", "event"])
        .agg(pl.count().alias("n"))
    )
 
    counts = counts.with_columns(
        (pl.col("event").replace(SIGNAL_WEIGHTS).cast(pl.Float32)
         * (pl.col("n").cast(pl.Float32) + 1).log()).alias("contribution")
    )
 
    return (
        counts.group_by(["visitorid", "itemid"])
        .agg(pl.col("contribution").sum().alias("signal_value"))
    )
 
 
# Test
import polars as pl
events = pl.read_csv("data/retailrocket/events.csv")
print(f"Raw events: {events.height:,}")
 
agg = aggregate_funnel_aware(events)
print(f"Aggregated (user, item) pairs: {agg.height:,}")
print(agg.head(5))
print(f"Signal value distribution:")
print(agg["signal_value"].value_counts())
 

# data/sessions.py — Session reconstruction
import polars as pl
 
SESSION_GAP_MS = 30 * 60 * 1000  # 30 dakika
 
def reconstruct_sessions(events: pl.DataFrame) -> pl.DataFrame:
    """
    Flat log → session-tagged events.
 
    Yeni session başlangıcı: user'ın aynı user için ardışık iki event arasında
    30 dakikadan fazla gap varsa.
 
    Returns DataFrame with [user, item, timestamp, session_id, position]
    """
    return (
        events.sort(["visitorid", "timestamp"])
        .with_columns([
            # Aynı user içinde önceki timestamp'i al
            pl.col("timestamp").shift(1).over("visitorid").alias("prev_ts"),
        ])
        .with_columns([
            # Yeni session başlıyor mu?
            (
                (pl.col("prev_ts").is_null())
                | ((pl.col("timestamp") - pl.col("prev_ts")) > SESSION_GAP_MS)
            ).alias("session_start"),
        ])
        .with_columns([
            # Session ID = cumulative sum of session_start, per user
            pl.col("session_start").cast(pl.Int32).cum_sum().over("visitorid").alias("session_id"),
        ])
        .drop("prev_ts", "session_start")
        .with_columns([
            # Session içinde pozisyon
            pl.int_range(0, pl.count()).over(["visitorid", "session_id"]).alias("position"),
        ])
    )
 
 
# Test
import polars as pl
events = pl.read_csv("data/retailrocket/events.csv")
sessions = reconstruct_sessions(events)
print(sessions.head(10))
 
# Session istatistikleri
stats = (
    sessions.group_by(["visitorid", "session_id"])
    .agg(pl.count().alias("session_len"))
)
print(f"Total sessions: {stats.height:,}")
print(f"Avg session length: {stats['session_len'].mean():.1f}")
print(f"P50 / P90 / P99: "
      f"{int(stats['session_len'].quantile(0.5))} / "
      f"{int(stats['session_len'].quantile(0.9))} / "
      f"{int(stats['session_len'].quantile(0.99))}")