delayed_feedback.py

1"""2延迟反馈多塔粗排模型（第11章 §11.2.3 延迟反馈多塔模型）34论文出处：Chapelle, "Modeling Delayed Feedback in Display Advertising", KDD 20145（DFM）；Ktena et al., "Addressing Delayed Feedback for Continuous Training with6Neural Networks in CTR Prediction", RecSys 2019（FNW / Fake Negative Weighting）。78核心思想：9点击等即时反馈在曝光后数秒内产生，而购买、支付等延迟反馈可能数天后才到达。10模型共享一个用户塔，物品侧分别建「即时反馈物品塔」与「延迟反馈物品塔」：11    s_click = <e_user, e_item^click>,  s_delay = <e_user, e_item^delay>12    L = L_click + λ·L_delay13核心问题是假负样本（Fake Negative）：曝光时延迟标签初始化为 0，但其中一部分14样本未来仍会转化，直接当负样本训练会系统性低估转化概率。15本文件实现两条数据流下的建模方式：16(a) 实时数据流（在线增量训练）：延迟时间分布建模（d ~ Exp(rate)，仍会转化的17    概率 P(d > t) = exp(-rate·t)）+ 假负样本修正的重要性加权 w = p/q，其中18    w_neg = 1 / (1 + p_pos·exp(-rate·t))，曝光时间越短权重越低（削弱错误监督），19    t 增大时 w → 1。20(b) 离线数据流（周期训练）：等待窗口 W 成熟后回填（Backfill）标签21    y_W = 1 if (t1 - t0) <= W else 0，标签完整性高但数据新鲜度下降。2223本实现：纯 Python 标准库，反向传播采用近似梯度（只更新最后一层权重 + Embedding）。24"""2526import math27import random282930def _dot(a: list[float], b: list[float]) -> float:31    return sum(x * y for x, y in zip(a, b))3233def _relu(x: float) -> float:34    return max(0.0, x)3536def _sigmoid(x: float) -> float:37    return 1.0 / (1.0 + math.exp(-max(-30.0, min(30.0, x))))383940class MLP:41    """简单 MLP：前向传播 + 最后一层近似梯度更新。"""4243    def __init__(self, dims: list[int], lr: float, reg: float):44        self.lr, self.reg = lr, reg45        self.weights: list[list[list[float]]] = []46        self.biases: list[list[float]] = []47        for in_d, out_d in zip(dims[:-1], dims[1:]):48            s = math.sqrt(2.0 / in_d)49            self.weights.append([[random.gauss(0, s) for _ in range(out_d)] for _ in range(in_d)])50            self.biases.append([0.0] * out_d)51        self.input_dim, self.output_dim = dims[0], dims[-1]5253    def forward(self, x: list[float]) -> tuple[list[float], list[list[float]]]:54        assert len(x) == self.input_dim, f"输入维度 {len(x)} != 期望 {self.input_dim}"55        acts, h = [x], x56        for idx, (W, b) in enumerate(zip(self.weights, self.biases)):57            h = [sum(h[j] * W[j][k] for j in range(len(h))) + b[k] for k in range(len(b))]58            if idx < len(self.weights) - 1:59                h = [_relu(v) for v in h]60            acts.append(h)61        return h, acts6263    def update_last_layer(self, grad_out: list[float], acts: list[list[float]]) -> None:64        h_prev, W, b = acts[-2], self.weights[-1], self.biases[-1]65        for j in range(len(h_prev)):66            for k in range(len(grad_out)):67                W[j][k] -= self.lr * (grad_out[k] * h_prev[j] + self.reg * W[j][k])68        for k in range(len(grad_out)):69            b[k] -= self.lr * grad_out[k]707172class DelayDistribution:73    """转化延迟分布建模（DFM 思想）：d ~ Exp(rate)，P(d > t) = exp(-rate·t) 为在途概率。"""7475    def __init__(self, rate: float = 0.5):76        self.rate = rate7778    def prob_still_pending(self, elapsed: float) -> float:79        """曝光后经过 elapsed 时间，转化「尚未到达但仍会发生」的概率。"""80        return math.exp(-self.rate * max(0.0, elapsed))8182    def expected_delay(self) -> float:83        """指数分布期望延迟 E[d] = 1 / rate。"""84        return 1.0 / self.rate8586    def fit(self, observed_delays: list[float]) -> None:87        """极大似然估计：rate_hat = n / Σ d_i。"""88        total = sum(observed_delays)89        if observed_delays and total > 1e-9:90            self.rate = len(observed_delays) / total919293class DelayedFeedbackTower:94    """95    延迟反馈多塔模型：共享用户塔 + 即时反馈物品塔 + 延迟反馈物品塔。96    s_click = <e_user, e_item^click>,  s_delay = <e_user, e_item^delay>97    L = L_click + λ·L_delay（延迟分支支持重要性加权纠偏）98    """99100    def __init__(self, n_users: int, n_items: int, emb_dim: int = 16,101                 hidden_dims: list[int] | None = None, output_dim: int = 16,102                 lr: float = 0.05, reg: float = 1e-4, lambda_delay: float = 1.0,103                 delay_rate: float = 0.5, prior_pos: float = 0.3):104        hidden_dims = hidden_dims or [32, 16]105        self.output_dim, self.lr = output_dim, lr106        self.lambda_delay = lambda_delay107        self.prior_pos = prior_pos            # 转化率先验 p_pos，用于重要性权重108        self.delay_dist = DelayDistribution(delay_rate)109110        scale = math.sqrt(1.0 / emb_dim)111        self.user_emb = [[random.uniform(-scale, scale) for _ in range(emb_dim)] for _ in range(n_users)]112        self.item_emb = [[random.uniform(-scale, scale) for _ in range(emb_dim)] for _ in range(n_items)]113        dims = [emb_dim] + hidden_dims + [output_dim]114        self.user_tower = MLP(dims, lr, reg)         # 所有样本都更新115        self.click_tower = MLP(dims, lr, reg)        # 仅即时反馈信号更新116        self.delay_tower = MLP(dims, lr, reg)        # 仅延迟反馈信号更新117118    def encode_user(self, uid: int) -> tuple[list[float], list[list[float]]]:119        return self.user_tower.forward(self.user_emb[uid])120121    def encode_item_click(self, iid: int) -> tuple[list[float], list[list[float]]]:122        return self.click_tower.forward(self.item_emb[iid])123124    def encode_item_delay(self, iid: int) -> tuple[list[float], list[list[float]]]:125        return self.delay_tower.forward(self.item_emb[iid])126127    def predict_delay_prob(self, uid: int, iid: int) -> float:128        """延迟反馈（转化）概率 σ(s_delay)。"""129        return _sigmoid(_dot(self.encode_user(uid)[0], self.encode_item_delay(iid)[0]))130131    # ── (a) 实时数据流：假负样本修正 + 重要性加权 ──────────────────────────132    def importance_weight(self, y_delay: float, elapsed: float) -> float:133        """134        重要性权重 w = p/q，纠正观测标签分布与真实分布的偏差。135        观测为负的样本中混入了比例 p_fake = p_pos·exp(-rate·t) 的假负样本，故136            w_neg = 1 / (1 + p_fake)137        t 很小（刚曝光）时 w_neg 最低，监督被削弱；t 增大后 w_neg → 1。138        观测为正的样本给 >1 的权重，补偿被延迟「吞掉」的那部分正样本。139        """140        if y_delay > 0.0:141            return 1.0 + self.delay_dist.prob_still_pending(elapsed)142        p_fake = self.prior_pos * self.delay_dist.prob_still_pending(elapsed)143        return 1.0 / (1.0 + p_fake)144145    def _step(self, uid: int, iid: int, y_click: float, y_delay: float,146              elapsed: float, debias: bool) -> tuple[float, float]:147        """单条样本的一次联合更新，返回 (L_click, L_delay)。"""148        e_u, u_acts = self.encode_user(uid)149        e_c, c_acts = self.encode_item_click(iid)150        e_d, d_acts = self.encode_item_delay(iid)151        # 即时反馈分支152        p_c = _sigmoid(_dot(e_u, e_c))153        l_c = -(y_click * math.log(p_c + 1e-9) + (1 - y_click) * math.log(1 - p_c + 1e-9))154        g_c = p_c - y_click155        self.click_tower.update_last_layer([g_c * e_u[k] for k in range(self.output_dim)], c_acts)156        # 延迟反馈分支（可选重要性加权纠偏）157        w = self.importance_weight(y_delay, elapsed) if debias else 1.0158        p_d = _sigmoid(_dot(e_u, e_d))159        l_d = -(y_delay * math.log(p_d + 1e-9) + (1 - y_delay) * math.log(1 - p_d + 1e-9))160        g_d = (p_d - y_delay) * w161        self.delay_tower.update_last_layer([g_d * e_u[k] for k in range(self.output_dim)], d_acts)162        # 用户塔：两个任务的梯度共同更新 → 同时学习短期与长期兴趣163        d_u = [g_c * e_c[k] + self.lambda_delay * g_d * e_d[k] for k in range(self.output_dim)]164        self.user_tower.update_last_layer(d_u, u_acts)165        ue, ie = self.user_emb[uid], self.item_emb[iid]166        g_i = g_c + self.lambda_delay * g_d167        for j in range(len(ue)):168            ue[j] -= self.lr * d_u[j % self.output_dim] * 0.1169            ie[j] -= self.lr * g_i * e_u[j % self.output_dim] * 0.1170        return l_c, l_d * w171172    def train_realtime(self, stream: list[tuple[int, int, float, float, float]],173                       epochs: int = 6, debias: bool = True, verbose: bool = True) -> None:174        """175        实时数据流训练（事件驱动）：stream 每条为176        (uid, iid, y_click, y_delay_observed, elapsed)。同一条曝光可能出现两次：177        第一次 y_delay=0（标签未成熟），延迟行为到达后再进入一次 y_delay=1，178        完成标签修正（Label Correction）。179        """180        tag = "纠偏" if debias else "不纠偏"181        for epoch in range(epochs):182            random.shuffle(stream)183            lc = ld = 0.0184            for uid, iid, y_c, y_d, el in stream:185                a, b = self._step(uid, iid, y_c, y_d, el, debias)186                lc, ld = lc + a, ld + b187            n = max(len(stream), 1)188            if verbose:189                print(f"  [{tag}] epoch {epoch:2d}  L_click={lc / n:.4f}  "190                      f"L_delay={ld / n:.4f}  L_total={(lc + self.lambda_delay * ld) / n:.4f}")191192    # ── (b) 离线数据流：等待窗口 + 回填标签 ────────────────────────────────193    @staticmethod194    def backfill_labels(raw_logs: list[tuple[int, int, float, float, float]],195                        window: float) -> list[tuple[int, int, float, float, float]]:196        """197        离线周期训练的样本组织：等待窗口 W 成熟后回填（Backfill）延迟标签198            y_W = 1 if (t1 - t0) <= W else 0199        raw_logs 每条为 (uid, iid, y_click, true_delay, conv_happen)。200        返回与实时流同构的样本（elapsed 置为 window，标签已成熟）。201        """202        return [(u, i, y_c, 1.0 if (cv > 0.0 and d <= window) else 0.0, window)203                for u, i, y_c, d, cv in raw_logs]204205    def train_offline(self, raw_logs: list[tuple[int, int, float, float, float]],206                      window: float = 3.0, epochs: int = 6) -> None:207        """离线周期训练：标签已回填成熟，无需重要性加权纠偏（debias=False）。"""208        data = self.backfill_labels(raw_logs, window)209        pos = sum(1 for d in data if d[3] > 0)210        print(f"  [离线 W={window}天] 回填后正样本 {pos}/{len(data)}（标签完整性高、新鲜度低）")211        self.train_realtime(data, epochs=epochs, debias=False)212213    def pre_rank(self, uid: int, candidates: list[int], top_k: int = 20,214                 alpha_click: float = 1.0, alpha_delay: float = 0.5) -> list[tuple[int, float]]:215        """粗排：多塔分数融合 s_final = Σ α_i·s_i（两个物品塔均可离线预计算）。"""216        e_u, _ = self.encode_user(uid)217        scores = [(iid, alpha_click * _dot(e_u, self.encode_item_click(iid)[0])218                   + alpha_delay * _dot(e_u, self.encode_item_delay(iid)[0]))219                  for iid in candidates]220        scores.sort(key=lambda x: x[1], reverse=True)221        return scores[:top_k]222223224def _make_logs(n_users: int, n_items: int, true_cvr: float, rate: float):225    """合成曝光日志：(uid, iid, y_click, true_delay, conv_happen)。"""226    logs = []227    for uid in range(n_users):228        for iid in range(n_items):229            if random.random() > 0.45:230                continue231            y_c = 1.0 if random.random() < 0.5 else 0.0232            conv = 1.0 if (y_c > 0 and random.random() < true_cvr) else 0.0233            logs.append((uid, iid, y_c,234                         random.expovariate(rate) if conv > 0 else 99.0, conv))235    return logs236237238# ── 快速验证 ──────────────────────────────────────────────────────────────────239if __name__ == "__main__":240    random.seed(42)241    n_users, n_items, TRUE_CVR, RATE, OBS_T = 8, 24, 0.40, 0.5, 0.6242    raw_logs = _make_logs(n_users, n_items, TRUE_CVR, RATE)243    # 实时流：曝光时刻只能观测到 delay <= OBS_T 的转化，其余都是假负样本244    stream = [(u, i, yc, 1.0 if (cv > 0 and d <= OBS_T) else 0.0, OBS_T)245              for u, i, yc, d, cv in raw_logs]246    obs_pos = sum(1 for s in stream if s[3] > 0)247    real_pos = sum(1 for l in raw_logs if l[4] > 0)248    truth = real_pos / len(raw_logs)249    print(f"合成数据：曝光 {len(raw_logs)} 条，真实转化 {real_pos} 条，实时流观测到 "250          f"{obs_pos} 条（假负样本 {real_pos - obs_pos} 条）\n真实转化率 = {truth:.4f}，"251          f"实时流朴素观测转化率 = {obs_pos / len(raw_logs):.4f}\n")252253    print("【实时数据流 A】不做纠偏（假负样本直接当负样本）：")254    m_naive = DelayedFeedbackTower(n_users, n_items, 16, lr=0.05,255                                   delay_rate=RATE, prior_pos=TRUE_CVR)256    m_naive.train_realtime(list(stream), epochs=5, debias=False)257258    print("\n【实时数据流 B】重要性加权纠偏（FNW + 指数延迟分布）：")259    m_fix = DelayedFeedbackTower(n_users, n_items, 16, lr=0.05,260                                 delay_rate=RATE, prior_pos=TRUE_CVR)261    m_fix.delay_dist.fit([l[3] for l in raw_logs if l[4] > 0])262    print(f"  延迟分布 MLE 估计 rate={m_fix.delay_dist.rate:.4f}，"263          f"期望延迟 {m_fix.delay_dist.expected_delay():.4f} 天")264    m_fix.train_realtime(list(stream), epochs=5, debias=True)265266    pairs = [(u, i) for u, i, *_ in raw_logs]267    avg_naive = sum(m_naive.predict_delay_prob(u, i) for u, i in pairs) / len(pairs)268    avg_fix = sum(m_fix.predict_delay_prob(u, i) for u, i in pairs) / len(pairs)269    print(f"\n正样本概率估计对比（真实转化率 {truth:.4f}）："270          f"\n  不做纠偏：平均预估 {avg_naive:.4f}，偏差 {avg_naive - truth:+.4f}"271          f"\n  做了纠偏：平均预估 {avg_fix:.4f}，偏差 {avg_fix - truth:+.4f}")272273    print("\n【离线数据流】等待窗口 + 回填标签：")274    m_off = DelayedFeedbackTower(n_users, n_items, 16, lr=0.05, delay_rate=RATE)275    m_off.train_offline(raw_logs, window=3.0, epochs=5)276277    uid = 0278    print(f"\n用户 {uid} 粗排 Top-5（s = 1.0·s_click + 0.5·s_delay）：")279    for iid, s in m_off.pre_rank(uid, list(range(n_items)), top_k=5):280        print(f"  item {iid:2d}: {s:.4f}")