强化学习中的反事实评估：不部署策略如何评估效果

第一章：反事实评估的理论基石与业务痛点

1.1 传统评估方法的困境：为什么必须部署才能评估？

在推荐系统、广告出价、动态定价等场景中，策略迭代面临根本性矛盾：离线训练的策略必须上线才能知道真实效果，这导致：

痛点维度	具体表现	业务损失	发生频率
迭代成本高	每版策略需2周A/B测试，100+工程师日投入	延迟上线损失¥500万/次	每周1-2次
策略风险大	新策略可能崩溃、转化率暴跌	曾致GMV单日下跌12%	每季度0.3次严重事故
探索受限	只能测试少数"安全"策略，激进策略无法评估	错失潜在+30%收益机会	持续存在
数据效率低	99%用户流量用于 exploitation，仅1%用于 exploration	学习速度慢100倍	持续存在
冷启动问题	新用户/新品类无历史数据，策略完全随机	首周体验差，流失率+8%	每日新增

核心难题在于：离线训练的RL策略基于历史策略生成的数据，当新策略行为分布偏离历史策略时（covariate shift），离线指标无法反映在线性能，这就是**离线策略评估（Off-Policy Evaluation, OPE）**问题。

---

1.2 反事实评估的核心思想

反事实评估通过 重要性加权（Importance Weighting） 解决分布偏移问题。其数学本质是：

Ⅰ. 问题形式化

• 历史策略（行为策略）π_b 生成日志数据 D = {(s_t, a_t, r_t)}_t=1^T
• 目标策略 π_e 需要被评估（未部署）
• 核心挑战：P_π_e(s,a) ≠ P_π_b(s,a)

Ⅱ. 反事实推理
通过重要性比率 w(s,a) = π_e(a|s) / π_b(a|s)，对历史数据加权，使加权后的分布逼近目标策略分布：

E_π_e[r] ≈ 1/n Σ w(s_i,a_i) * r_i

Ⅲ. 双重稳健保证
结合直接方法（Direct Method）减少方差，只要模型或重要性权重任一正确，估计即无偏。

---

1.3 反事实评估的业务价值

价值维度	实现方式	量化收益	适用场景
零风险策略迭代	离线评估100+策略，仅前3名上线测试	策略迭代成本降低90%	激进策略探索
超参数高效调优	离线搜索10万组RL超参	调优效率提升1000倍	深度强化学习
冷启动加速	基于历史相似用户数据评估新策略	冷启动期缩短至1天	新用户/新场景
持续学习	每日评估并选择最佳策略	收益周周提升	非平稳环境
安全审计	上线前验证策略无负向效应	减少30%严重事故	金融监管/医疗

---

1.4 Mermaid：反事实评估理论框架

Parse error on line 19: ... K --> L[策略价值 V(π_e)] subg ----------------------^ Expecting 'SEMI', 'NEWLINE', 'SPACE', 'EOF', 'GRAPH', 'DIR', 'subgraph', 'SQS', 'SQE', 'end', 'AMP', 'PE', '-)', 'STADIUMEND', 'SUBROUTINEEND', 'ALPHA', 'COLON', 'PIPE', 'CYLINDEREND', 'DIAMOND_STOP', 'TAGEND', 'TRAPEND', 'INVTRAPEND', 'START_LINK', 'LINK', 'STYLE', 'LINKSTYLE', 'CLASSDEF', 'CLASS', 'CLICK', 'DOWN', 'UP', 'DEFAULT', 'NUM', 'COMMA', 'MINUS', 'BRKT', 'DOT', 'PCT', 'TAGSTART', 'PUNCTUATION', 'UNICODE_TEXT', 'PLUS', 'EQUALS', 'MULT', 'UNDERSCORE', got 'PS'

---

第二章：技术架构与数据流水线设计

2.1 反事实评估的数据收集要求

反事实评估对日志数据的完备性要求远超普通A/B测试。每条日志必须包含：

数据字段	数据类型	是否必需	说明
user_id	String	✅ 必需	用户唯一标识，用于状态追踪
context_s	JSON	✅ 必需	状态向量：用户历史、场景特征
action_a	String/Int	✅ 必需	实际采取的动作（商品ID、出价等）
reward_r	Float	✅ 必需	即时奖励（点击、GMV、时长）
propensity_b	Float	✅ 必需	行为策略概率 π_b(a
propensity_e	Float	✅ 必需	目标策略概率 π_e(a
session_id	String	⭕ 推荐	用于会话级状态建模
timestamp	Timestamp	✅ 必需	用于时序分析与非平稳检测

关键要求：必须记录行为策略对每个动作的完整概率分布，而非仅采样动作。这对推荐系统意味着：需要记录完整的排序分数或logits，而非仅Top-K结果。

---

2.2 数据架构与存储策略

# etl/log_collection_pipeline.py
from pyspark.sql import SparkSession
from pyspark.sql.functions import udf, col, struct, explode, posexplode
from pyspark.sql.types import ArrayType, FloatType, StringType, StructType, StructField
import json

class LogCollectionPipeline:
    """RL日志收集管道"""
    
    def __init__(self, spark: SparkSession):
        self.spark = spark
        
    def process_recommendation_logs(self, raw_logs_path: str, output_table: str):
        """Ⅰ. 处理推荐系统原始日志"""
        
        # 原始日志格式：每次请求返回100个候选商品及分数
        schema = StructType([
            StructField("request_id", StringType()),
            StructField("user_id", StringType()),
            StructField("timestamp", StringType()),
            StructField("context", StructType([  # 上下文状态
                StructField("user_history", ArrayType(StringType())),
                StructField("time_of_day", StringType()),
                StructField("device_type", StringType())
            ])),
            StructField("candidates", ArrayType(StructType([  # 完整候选集
                StructField("item_id", StringType()),
                StructField("score", FloatType()),  # 行为策略分数
                StructField("rank", IntegerType())
            ]))),
            StructField("shown_items", ArrayType(StringType())),  # 实际展示
            StructField("clicked_item", StringType()),  # 用户反馈
            StructField("gmv", FloatType())
        ])
        
        raw_df = self.spark.read.json(raw_logs_path, schema=schema)
        
        # Ⅱ. 计算每个展示动作的 propensity_b
        # 假设使用softmax选择：P(a) = exp(score) / Σ exp(scores)
        
        @udf(returnType=ArrayType(FloatType()))
        def calculate_softmax_probabilities(scores):
            if not scores:
                return []
            scores_np = np.array([s['score'] for s in scores])
            exp_scores = np.exp(scores_np - np.max(scores_np))  # 数值稳定
            probs = exp_scores / np.sum(exp_scores)
            return probs.tolist()
        
        # 展开候选商品，计算概率
        candidates_df = raw_df.select(
            "request_id",
            "user_id",
            "timestamp",
            "context",
            posexplode("candidates").alias("rank", "candidate")
        ).withColumn("item_id", col("candidate.item_id")) \
         .withColumn("score", col("candidate.score")) \
         .groupBy("request_id") \
         .agg(
             F.collect_list("candidate").alias("candidates"),
             F.collect_list("score").alias("scores")
         )
        
        # 计算softmax概率
        candidates_df = candidates_df.withColumn(
            "propensity_distribution",
            calculate_softmax_probabilities(col("candidates"))
        )
        
        # Ⅲ. 与展示日志join，获取每个动作的propensity_b
        shown_df = raw_df.select(
            "request_id",
            explode("shown_items").alias("item_id"),
            "clicked_item",
            "gmv"
        )
        
        # 将概率分布与展示商品关联
        rl_logs = shown_df.join(candidates_df, on="request_id") \
            .select(
                "user_id",
                "item_id",
                "gmv",
                "context",
                "propensity_distribution",
                "clicked_item",
                "timestamp"
            )
        
        # Ⅳ. 写入Delta Lake（支持ACID和时间旅行）
        rl_logs.write.format("delta") \
            .mode("append") \
            .partitionBy("date") \
            .option("mergeSchema", "true") \
            .save("s3://data-lake/rl-logs/")
        
        # Ⅴ. 创建行为策略表（每日快照）
        self._create_policy_snapshot(raw_df)
        
        return rl_logs.count()
    
    def _create_policy_snapshot(self, raw_df):
        """创建行为策略每日快照（用于离线策略评估）"""
        
        policy_df = raw_df.select(
            F.date_trunc("day", "timestamp").alias("date"),
            F.explode("candidates").alias("candidate")
        ).select(
            "date",
            F.avg("candidate.score").alias("avg_score"),
            F.stddev("candidate.score").alias("std_score")
        ).groupBy("date") \
         .agg(
             F.map_from_entries(
                 F.collect_list(
                     F.struct("candidate.item_id", "candidate.score")
                 )
             ).alias("policy_params")
         )
        
        policy_df.write.format("delta") \
            .mode("overwrite") \
            .save("s3://data-lake/rl-policies/behavior_policy/")
        
        print(f"策略快照已创建，日期: {policy_df.select('date').distinct().count()}")

# 使用示例
if __name__ == "__main__":
    spark = SparkSession.builder \
        .appName("RLLogCollection") \
        .config("spark.jars.packages", "io.delta:delta-core_2.12:2.4.0") \
        .getOrCreate()
    
    pipeline = LogCollectionPipeline(spark)
    log_count = pipeline.process_recommendation_logs(
        raw_logs_path="s3://raw-logs/recommendation/2024-01-*/",
        output_table="rl_logs"
    )
    
    print(f"处理了{log_count}条RL日志")

代码深度解析：

1. 完整候选集记录：必须保存所有候选商品的排序分数，而非仅Top-K。这是因为重要性加权需要计算未展示动作的propensity_b（通常是长尾商品）。

2. Softmax概率计算：exp(scores - max)避免数值溢出，这是稳定计算重要性比率的关键。

3. Delta Lake时间旅行：保留所有历史策略版本，支持回溯评估任意日期的策略。

4. 分区策略：按日期分区使评估特定时间段的策略时，仅需扫描对应分区，成本降低95%。

---

2.3 Mermaid：数据收集架构

---

第三章：反事实评估方法体系与代码实现

3.1 重要性采样（Importance Sampling）

重要性采样（IPS）是OPE的基石，通过加权历史数据模拟新策略表现。

# evaluation/importance_sampling.py
import numpy as np
from typing import List, Tuple, Dict
import pandas as pd

class ImportanceSamplingEvaluator:
    """重要性采样评估器"""
    
    def __init__(self, epsilon: float = 1e-8):
        """
        Args:
            epsilon: 截断概率防止无穷权重，通常1e-6到1e-8
        """
        self.epsilon = epsilon
        
    def calculate_propensity_ratios(self, action_log_probs: np.ndarray, 
                                    target_action_probs: np.ndarray) -> np.ndarray:
        """
        Ⅰ. 计算重要性比率 w = π_e(a|s) / π_b(a|s)
        
        Args:
            action_log_probs: 形状 [n_samples,]，log(π_b(a|s))
            target_action_probs: 形状 [n_samples,]，π_e(a|s)
            
        Returns:
            ratios: 形状 [n_samples,]，重要性比率
        """
        
        # 转换log概率到概率
        action_probs = np.exp(action_log_probs)
        
        # 截断避免除零和爆炸权重
        action_probs = np.clip(action_probs, self.epsilon, 1.0)
        
        # 计算比率
        ratios = target_action_probs / action_probs
        
        # 诊断：权重分布统计
        diagnostics = {
            "mean_weight": np.mean(ratios),
            "max_weight": np.max(ratios),
            "min_weight": np.min(ratios),
            "std_weight": np.std(ratios),
            "weight_cv": np.std(ratios) / np.mean(ratios),  # 变异系数
            "prop_weights > 10": np.mean(ratios > 10)
        }
        
        print(f"重要性权重诊断: {diagnostics}")
        
        # 警告：如果变异系数>2，说明方差过高，估计不可靠
        if diagnostics['weight_cv'] > 2.0:
            print("⚠️ 警告：重要性权重变异系数>2，建议使用SNIPS或DR")
        
        return ratios
    
    def estimate_value_ips(self, rewards: np.ndarray, 
                          action_log_probs: np.ndarray, 
                          target_action_probs: np.ndarray) -> Tuple[float, float]:
        """
        Ⅱ. 基础IPS估计
        
        V̂(π_e) = 1/n Σ w_i * r_i
        
        Args:
            rewards: [n_samples,]，即时奖励
            action_log_probs: [n_samples,]，行为策略log概率
            target_action_probs: [n_samples,]，目标策略概率
            
        Returns:
            value_estimate: 目标策略价值估计
            stderr: 标准误（自助法）
        """
        
        # 计算重要性权重
        weights = self.calculate_propensity_ratios(
            action_log_probs, target_action_probs
        )
        
        # IPS估计
        weighted_rewards = weights * rewards
        value_estimate = np.mean(weighted_rewards)
        
        # 使用自助法计算标准误
        n_bootstrap = 1000
        bootstrap_estimates = []
        
        for _ in range(n_bootstrap):
            # 有放回采样
            sample_idx = np.random.choice(len(rewards), size=len(rewards), replace=True)
            bootstrap_val = np.mean(weighted_rewards[sample_idx])
            bootstrap_estimates.append(bootstrap_val)
        
        stderr = np.std(bootstrap_estimates)
        
        # 95%置信区间
        ci_lower = np.percentile(bootstrap_estimates, 2.5)
        ci_upper = np.percentile(bootstrap_estimates, 97.5)
        
        return {
            "value_estimate": value_estimate,
            "stderr": stderr,
            "ci_95": [ci_lower, ci_upper],
            "n_samples": len(rewards),
            "method": "IPS"
        }
    
    def estimate_value_snips(self, rewards: np.ndarray,
                           action_log_probs: np.ndarray,
                           target_action_probs: np.ndarray) -> Dict:
        """
        Ⅲ. 自归一化重要性采样（SNIPS）
        
        V̂_SNIPS = Σ w_i * r_i / Σ w_i
        
        通过归一化减少方差，是IPS的标准改进
        """
        
        weights = self.calculate_propensity_ratios(
            action_log_probs, target_action_probs
        )
        
        # SNIPS归一化
        weighted_rewards = weights * rewards
        snips_value = np.sum(weighted_rewards) / np.sum(weights)
        
        # 自助法标准误（需要调整以考虑归一化）
        n_bootstrap = 1000
        snips_estimates = []
        
        for _ in range(n_bootstrap):
            sample_idx = np.random.choice(len(rewards), size=len(rewards), replace=True)
            bootstrap_weights = weights[sample_idx]
            bootstrap_rewards = rewards[sample_idx]
            snips_estimates.append(
                np.sum(bootstrap_weights * bootstrap_rewards) / 
                np.sum(bootstrap_weights)
            )
        
        return {
            "value_estimate": snips_value,
            "stderr": np.std(snips_estimates),
            "ci_95": [np.percentile(snips_estimates, 2.5), 
                     np.percentile(snips_estimates, 97.5)],
            "method": "SNIPS",
            "normalization_factor": np.mean(weights)
        }

# 实战调用示例
if __name__ == "__main__":
    # 模拟推荐系统数据
    np.random.seed(42)
    n_samples = 100000
    
    # 行为策略：ε-贪婪，ε=0.1
    # 对50个商品采用softmax，探索10%
    epsilon = 0.1
    n_actions = 50
    
    # 生成用户状态（简化）
    user_interests = np.random.dirichlet([1.0]*n_actions, size=n_samples)
    
    # 行为策略概率（添加噪声模拟真实策略）
    action_log_probs = np.log(
        0.9 * user_interests[np.arange(n_samples), np.random.randint(0, n_actions, n_samples)] + 
        0.1 / n_actions
    )
    
    # 目标策略：更贪婪，ε=0.05
    target_probs = 0.95 * user_interests[np.arange(n_samples), np.random.randint(0, n_actions, n_samples)] + \
                   0.05 / n_actions
    
    # 奖励：点击或购买
    rewards = np.random.binomial(1, p=0.05, size=n_samples) * np.random.exponential(50, size=n_samples)
    
    evaluator = ImportanceSamplingEvaluator(epsilon=1e-8)
    
    # IPS估计
    ips_result = evaluator.estimate_value_ips(
        rewards, action_log_probs, target_probs
    )
    print(f"IPS估计: V̂ = {ips_result['value_estimate']:.4f} ± {ips_result['stderr']:.4f}")
    
    # SNIPS估计
    snips_result = evaluator.estimate_value_snips(
        rewards, action_log_probs, target_probs
    )
    print(f"SNIPS估计: V̂ = {snips_result['value_estimate']:.4f} ± {snips_result['stderr']:.4f}")
    
    # 结果比较：SNIPS方差通常降低30-50%
    variance_reduction = 1 - (snips_result['stderr'] / ips_result['stderr'])**2
    print(f"SNIPS方差缩减: {variance_reduction:.2%}")

---

3.2 双稳健估计（Doubly Robust）

DR估计结合IPS和直接方法，只要任一组件正确即保证无偏。

# evaluation/doubly_robust.py
from sklearn.ensemble import RandomForestRegressor, GradientBoostingRegressor
from sklearn.model_selection import train_test_split
from typing import Dict, Optional

class DoublyRobustEvaluator:
    """双稳健评估器"""
    
    def __init__(self, model_type: str = "rf", n_estimators: int = 100):
        """
        Args:
            model_type: "rf" (随机森林) 或 "gbdt" (梯度提升)
            n_estimators: 基学习器数量
        """
        self.model_type = model_type
        self.n_estimators = n_estimators
        self.q_model = None  # Q(s,a)函数模型
        
    def fit_q_function(self, contexts: np.ndarray, actions: np.ndarray, 
                      rewards: np.ndarray, action_log_probs: np.ndarray) -> float:
        """
        Ⅰ. 训练Q函数模型（直接方法组件）
        
        Args:
            contexts: [n_samples, n_features]，状态特征
            actions: [n_samples,]，采取的动作
            rewards: [n_samples,]，即时奖励
            action_log_probs: [n_samples,]，行为策略log概率（用于样本权重）
            
        Returns:
            model_rmse: 模型预测误差
        """
        
        # 构建训练数据：特征 = context + action
        action_onehot = np.eye(actions.max() + 1)[actions]
        X = np.hstack([contexts, action_onehot])
        
        # 样本权重 = 1 / π_b(a|s)（逆概率加权减少偏差）
        sample_weights = 1.0 / np.exp(action_log_probs)
        
        # 训练模型
        if self.model_type == "rf":
            model = RandomForestRegressor(
                n_estimators=self.n_estimators,
                max_depth=10,
                min_samples_leaf=100,
                random_state=42,
                n_jobs=-1
            )
        else:
            model = GradientBoostingRegressor(
                n_estimators=self.n_estimators,
                max_depth=5,
                learning_rate=0.1,
                random_state=42
            )
        
        # 交叉验证训练
        X_train, X_val, y_train, y_val, w_train, w_val = train_test_split(
            X, rewards, sample_weights, test_size=0.2, random_state=42
        )
        
        model.fit(X_train, y_train, sample_weight=w_train)
        
        # 验证集评估
        val_pred = model.predict(X_val)
        rmse = np.sqrt(np.mean((val_pred - y_val)**2))
        
        self.q_model = model
        
        print(f"Q函数模型验证RMSE: {rmse:.4f}")
        return rmse
    
    def estimate_value_dr(self, contexts: np.ndarray, actions: np.ndarray,
                         rewards: np.ndarray, action_log_probs: np.ndarray,
                         target_action_probs: np.ndarray) -> Dict:
        """
        Ⅱ. 双稳健估计
        
        V̂_DR = 1/n Σ [ Q(s_i,π_e) + w_i * (r_i - Q(s_i,a_i)) ]
        
        第一项直接使用Q函数预测，第二项用IPS校正残差
        """
        
        if self.q_model is None:
            raise ValueError("必须先调用fit_q_function训练模型")
        
        # 计算重要性权重（与IPS相同）
        ips_evaluator = ImportanceSamplingEvaluator()
        weights = ips_evaluator.calculate_propensity_ratios(
            action_log_probs, target_action_probs
        )
        
        # 预测Q(s_i, a_i) - 直接方法部分
        action_onehot = np.eye(actions.max() + 1)[actions]
        X = np.hstack([contexts, action_onehot])
        q_pred = self.q_model.predict(X)
        
        # 预测Q(s_i, π_e) - 目标策略的期望奖励
        # 方法一：对目标策略的所有动作加权平均
        # q_pred_pi_e = Σ_a π_e(a|s_i) * Q(s_i,a)
        
        q_pred_pi_e = np.zeros_like(rewards)
        
        for i in range(len(contexts)):
            # 为当前状态构建所有动作的特征
            context_repeat = np.tile(contexts[i], (50, 1))  # 假设50个动作
            actions_all = np.eye(50)
            X_all_actions = np.hstack([context_repeat, actions_all])
            
            # 预测所有动作的Q值
            q_all = self.q_model.predict(X_all_actions)
            
            # 加权平均（目标策略概率）
            q_pred_pi_e[i] = np.sum(target_action_probs[i] * q_all)
        
        # 双重稳健公式
        dr_term = q_pred_pi_e + weights * (rewards - q_pred)
        value_estimate = np.mean(dr_term)
        
        # 标准误（需考虑模型方差）
        stderr = np.std(dr_term) / np.sqrt(len(dr_term))
        
        return {
            "value_estimate": value_estimate,
            "stderr": stderr,
            "ci_95": [value_estimate - 1.96*stderr, value_estimate + 1.96*stderr],
            "model_rmse": np.sqrt(np.mean((q_pred - rewards)**2)),
            "method": "Doubly Robust"
        }
    
    def estimate_value_wdr(self, contexts: np.ndarray, actions: np.ndarray,
                          rewards: np.ndarray, action_log_probs: np.ndarray,
                          target_action_probs: np.ndarray) -> Dict:
        """
        Ⅲ. 加权双稳健估计（WDR）
        
        在DR基础上对权重归一化，进一步降低方差
        """
        
        weights = ImportanceSamplingEvaluator().calculate_propensity_ratios(
            action_log_probs, target_action_probs
        )
        
        # 归一化权重
        weights_normalized = weights / np.sum(weights) * len(weights)
        
        # 模型预测（与DR相同）
        action_onehot = np.eye(actions.max() + 1)[actions]
        X = np.hstack([contexts, action_onehot])
        q_pred = self.q_model.predict(X)
        
        # 预测Q(s_i, π_e)
        q_pred_pi_e = np.zeros_like(rewards)
        for i in range(len(contexts)):
            context_repeat = np.tile(contexts[i], (50, 1))
            actions_all = np.eye(50)
            X_all = np.hstack([context_repeat, actions_all])
            q_all = self.q_model.predict(X_all)
            q_pred_pi_e[i] = np.sum(target_action_probs[i] * q_all)
        
        # 加权的双重稳健
        wdr_term = q_pred_pi_e + weights_normalized * (rewards - q_pred)
        value_estimate = np.mean(wdr_term)
        
        return {
            "value_estimate": value_estimate,
            "stderr": np.std(wdr_term) / np.sqrt(len(wdr_term)),
            "method": "Weighted DR",
            "weight_normalization_factor": np.mean(weights)
        }

# 实战调用
if __name__ == "__main__":
    # 生成带状态特征的数据
    np.random.seed(42)
    n_samples = 50000
    
    # 上下文特征（10维用户状态）
    n_features = 10
    contexts = np.random.randn(n_samples, n_features)
    
    # 行为策略：ε-贪婪
    n_actions = 20
    optimal_action = np.argmax(contexts @ np.random.randn(n_features, n_actions), axis=1)
    explore = np.random.binomial(1, 0.1, n_samples)
    actions = np.where(explore, np.random.randint(0, n_actions, n_samples), optimal_action)
    
    # 真实奖励函数（带噪声）
    true_q = contexts @ np.random.randn(n_features) + 0.5 * actions + np.random.randn(n_samples) * 0.1
    rewards = np.maximum(true_q, 0)
    
    # 策略概率
    action_log_probs = np.log(0.9 * (actions == optimal_action) + 0.1 / n_actions)
    target_probs = np.random.rand(n_samples)  # 目标策略概率
    
    # DR评估
    dr_evaluator = DoublyRobustEvaluator(model_type="rf", n_estimators=200)
    rmse = dr_evaluator.fit_q_function(contexts, actions, rewards, action_log_probs)
    
    dr_result = dr_evaluator.estimate_value_dr(
        contexts, actions, rewards, action_log_probs, target_probs
    )
    print(f"DR估计: V̂ = {dr_result['value_estimate']:.4f} ± {dr_result['stderr']:.4f}")
    
    # WDR估计
    wdr_result = dr_evaluator.estimate_value_wdr(
        contexts, actions, rewards, action_log_probs, target_probs
    )
    print(f"WDR估计: V̂ = {wdr_result['value_estimate']:.4f} ± {wdr_result['stderr']:.4f}")

---

3.3 方法对比与选择指南

评估方法	偏差（Bias）	方差（Variance）	模型依赖	计算成本	适用场景
IPS	低（若π_b已知）	极高	无	低	探索策略与行为策略相似时
SNIPS	轻微（归一化）	高	无	低	大多数OPE场景的标准基线
Direct Method	高（模型错误）	极低	强	中	需要准确Q(s,a)模型
Doubly Robust	极低	中	中等	中	推荐作为主力方法
WDR	极低	低	中等	中	大规模数据首选
MAGIC	自适应	自适应	可调	高	专家级调参场景

选择决策树：

Ⅰ. 数据规模 < 10万样本

• 使用SNIPS，避免模型过拟合风险

Ⅱ. 数据规模 > 10万且有丰富特征

• 使用WDR，平衡偏差与方差
• 训练Q函数模型（轻量级GBDT）

Ⅲ. 行为策略概率未知（黑盒日志）

• 只能使用Direct Method
• 需要强模型保证

Ⅳ. 策略间差异度 > 50%

• 优先使用DR/WDR防止IPS爆炸
• 对权重做截断（truncate at 95分位）

---

3.4 Mermaid：OPE方法选择流程

Parse error on line 9: ...度?} F -->|大(>50%)| G[DR/WDR + 权 ---------------------^ Expecting 'SEMI', 'NEWLINE', 'SPACE', 'EOF', 'GRAPH', 'DIR', 'subgraph', 'SQS', 'SQE', 'end', 'AMP', 'PE', '-)', 'STADIUMEND', 'SUBROUTINEEND', 'ALPHA', 'COLON', 'PIPE', 'CYLINDEREND', 'DIAMOND_STOP', 'TAGEND', 'TRAPEND', 'INVTRAPEND', 'START_LINK', 'LINK', 'STYLE', 'LINKSTYLE', 'CLASSDEF', 'CLASS', 'CLICK', 'DOWN', 'UP', 'DEFAULT', 'NUM', 'COMMA', 'MINUS', 'BRKT', 'DOT', 'PCT', 'TAGSTART', 'PUNCTUATION', 'UNICODE_TEXT', 'PLUS', 'EQUALS', 'MULT', 'UNDERSCORE', got 'PS'

---

第四章：实战案例 - 电商推荐策略离线迭代评估（2000+字）

4.1 业务背景与策略迭代挑战

某头部电商平台（月活2.3亿）在2024年Q2面临推荐策略迭代困境：

策略迭代现状诊断：

迭代阶段	耗时	参与人数	成本	成功率	主要瓶颈
离线训练	3天	2算法工程师	¥5万	95%	数据准备耗时
在线测试	14天	8人（算法+工程+数据科学家）	¥80万	60%	工程联调、数据对齐、业务方审批
效果分析	3天	2数据科学家	¥10万	90%	统计功效不足需延长测试
回滚决策	1天	全员	¥30万	40%策略需回滚	离线指标与在线不一致

核心矛盾：每周只能测试2-3个策略，而算法团队每天产生10-20个新策略想法，95%的创新被测试能力瓶颈扼杀。

业务目标：

• 实现离线评估100+策略/周，仅Top-5策略上线测试
• 将迭代周期从14天缩短至2天
• 将策略回滚率从40%降至5%

---

4.2 数据准备与策略日志

行为策略π_b：生产环境运行的DNN排序模型（2024-01-15至2024-02-15，30天日志）

目标策略集π_e：15个候选策略，包括：

• 策略A：GBDT排序（轻量级对比）
• 策略B：DeepFM（特征交叉增强）
• 策略C：DIN+（注意力机制优化）
• 策略D-G：不同超参的DIN变体
• 策略H-O：融合多目标（CTR+CVR+GMV）

关键数据规模：

数据维度	数值	说明
日志天数	30天	覆盖春节假期，包含非平稳性
日活用户	4200万	真实生产流量
每日请求	8.7亿次	PV量级
每请求候选	200商品	完整候选集（非截断）
日志总量	260TB	原始JSON数据
处理后RL日志	18TB	包含propensity_b的parquet格式

---

数据ETL核心代码：

# case_study/prepare_rl_logs.py
def prepare_rl_logs_for_ope(spark, input_path, output_table, date_range):
    """
    为OPE准备RL日志（30天数据）
    关键：确保每个请求的所有候选商品都有排序分数
    """
    
    # Ⅰ. 读取原始日志（包含完整候选集）
    raw_df = spark.read.parquet(f"{input_path}/date={date_range}")
    
    # Ⅱ. 过滤无效请求
    filtered_df = raw_df.filter(
        (F.size("candidates") == 200) &  # 确保候选集完整
        (F.col("user_id").isNotNull()) &
        (F.col("timestamp").isNotNull())
    )
    
    # Ⅲ. 计算softmax概率分布
    # 注意：生产系统使用自定义温度参数τ=0.5
    temperature = 0.5
    
    @udf(returnType=ArrayType(FloatType()))
    def compute_softmax_with_temperature(scores):
        scores_np = np.array(scores, dtype=np.float64)
        exp_scores = np.exp(scores_np / temperature - np.max(scores_np / temperature))
        probs = exp_scores / np.sum(exp_scores)
        return probs.tolist()
    
    processed_df = filtered_df.withColumn(
        "propensity_distribution",
        compute_softmax_with_temperature(F.col("scores"))
    )
    
    # Ⅳ. 与展示日志join（仅保留top-10展示商品）
    # 这是核心：即使只展示10个，也需要200个商品的完整概率分布
    shown_df = processed_df.select(
        "request_id",
        F.explode(F.arrays_zip(
            F.col("shown_items"),
            F.range(0, 10)  # 位置0-9
        )).alias("shown")
    ).select(
        "request_id",
        F.col("shown.shown_items").alias("item_id"),
        F.col("shown.0").alias("position")
    )
    
    # 关联得到每条展示记录的propensity_b
    rl_logs = shown_df.join(
        processed_df.select("request_id", "propensity_distribution"),
        on="request_id"
    ).select(
        F.col("request_id").alias("sample_id"),
        "user_id",
        "item_id",
        "position",
        "propensity_distribution",
        F.col("clicked").cast("float").alias("reward"),
        "timestamp"
    )
    
    # Ⅴ. 写入Delta表（分区加速查询）
    rl_logs.write.format("delta") \
        .mode("overwrite") \
        .partitionBy("date", "hour") \
        .option("mergeSchema", "true") \
        .save(f"s3://ope-data/{output_table}")
    
    print(f"RL日志处理完成：{rl_logs.count()}条样本")

# 执行数据准备
spark = SparkSession.builder.appName("OPEDataPrep").getOrCreate()
prepare_rl_logs_for_ope(
    spark, 
    input_path="s3://prod-logs/recommendation/",
    output_table="rl_logs_v1",
    date_range="2024-01-15 to 2024-02-15"
)

---

4.3 策略评估流水线实现

并行评估15个策略的Airflow DAG：

# case_study/ope_evaluation_dag.py
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta
import json

default_args = {
    'owner': 'rl_team',
    'depends_on_past': False,
    'email_on_failure': True,
    'retries': 2,
    'retry_delay': timedelta(minutes=10),
    'execution_timeout': timedelta(hours=4)
}

with DAG(
    'ope_strategy_evaluation',
    default_args=default_args,
    schedule_interval=None,  # 手动触发
    start_date=datetime(2024, 2, 1),
    catchup=False,
    tags=['rl', 'ope', 'production']
) as dag:
    
    # Ⅰ. 加载RL日志
    load_logs = PythonOperator(
        task_id='load_rl_logs',
        python_callable=load_delta_table_to_pandas,
        op_kwargs={
            'table_path': 's3://ope-data/rl_logs_v1',
            'date_range': '2024-01-15 to 2024-02-15',
            'sample_rate': 0.1  # 1000万样本抽样到100万
        }
    )
    
    # Ⅱ. 定义15个待评估策略
    strategies = [
        {"name": "gbdt_baseline", "type": "gbdt", "params": {}},
        {"name": "deepfm_v1", "type": "deepfm", "params": {"embedding_dim": 64}},
        {"name": "din_attention", "type": "din", "params": {"attention_units": 128}},
        {"name": "din_wide", "type": "din", "params": {"wide_features": True}},
        # ... 省略11个策略
    ]
    
    # Ⅲ. 并行评估每个策略
    evaluation_tasks = []
    
    for strategy in strategies:
        task_id = f"evaluate_{strategy['name']}"
        
        evaluate_task = PythonOperator(
            task_id=task_id,
            python_callable=run_ope_evaluation,
            op_kwargs={
                'rl_logs': "{{ task_instance.xcom_pull(task_ids='load_rl_logs') }}",
                'strategy_config': strategy,
                'methods': ['snips', 'wdr'],  # 使用SNIPS+WDR双保险
                'output_path': f"s3://ope-results/{strategy['name']}.json"
            }
        )
        
        load_logs >> evaluate_task
        evaluation_tasks.append(evaluate_task)
    
    # Ⅳ. 结果聚合与排名
    aggregate_results = PythonOperator(
        task_id='aggregate_strategy_ranking',
        python_callable=rank_strategies_by_confidence,
        op_kwargs={
            'strategy_results': [
                f"s3://ope-results/{s['name']}.json" for s in strategies
            ],
            'confidence_threshold': 0.05,  # 剔除置信区间过宽的策略
            'select_top_k': 5  # 选Top-5上线验证
        }
    )
    
    # Ⅴ. 生成对比报告
    generate_report = PythonOperator(
        task_id='generate_ope_report',
        python_callable=create_html_comparison_report,
        op_kwargs={
            'ranking_result': "{{ task_instance.xcom_pull(task_ids='aggregate_strategy_ranking') }}",
            'report_template': 'ope_strategy_comparison.html',
            'output_path': 's3://ope-reports/daily_report.html'
        }
    )
    
    # Ⅵ. 元数据记录
    log_metadata = PythonOperator(
        task_id='log_evaluation_metadata',
        python_callable=record_to_mlflow,
        op_kwargs={
            'experiment_name': 'recommendation_ope',
            'run_params': {
                'n_strategies': len(strategies),
                'evaluation_period': '30days',
                'method': 'SNIPS+WDR'
            }
        }
    )
    
    # 依赖关系
    evaluation_tasks >> aggregate_results >> generate_report >> log_metadata

---

核心评估函数实现：

# evaluation/ope_runner.py
def run_ope_evaluation(rl_logs: pd.DataFrame, strategy_config: dict, 
                       methods: list, output_path: str) -> dict:
    """
    对单个策略执行OPE评估
    rl_logs必须包含: user_id, item_id, reward, propensity_distribution
    """
    
    # Ⅰ. 构建目标策略π_e的概率分布
    if strategy_config['type'] == 'din':
        # 加载DIN模型，为每个样本计算目标概率
        from models.din_model import DINModel
        
        din_model = DINModel.load(f"models/{strategy_config['name']}")
        
        # 为每个(s,a)对计算π_e(a|s)
        # 注意：需要与rl_logs中的item_id对齐
        def compute_target_probs(row):
            context = json.loads(row['context'])
            item_id = row['item_id']
            
            # DIN模型前向传播
            scores = din_model.predict(context, [item_id] + get_top99_candidates(context))
            
            # 同样使用temperature=0.5的softmax
            probs = softmax_with_temperature(scores, temp=0.5)
            
            # 找到item_id对应的概率
            target_prob = probs[0]  # item_id在首位
            
            return target_prob
        
        # 向量化计算（实际使用Spark或批量预测）
        target_probs = rl_logs.apply(compute_target_probs, axis=1)
        
    else:
        # GBDT等模型采用类似方式
        from models.gbdt_model import GBDTModel
        
        gbdt_model = GBDTModel.load(f"models/{strategy_config['name']}")
        target_probs = gbdt_model.batch_predict(rl_logs[['context', 'item_id']])
    
    # Ⅱ. 提取行为策略概率
    # RL日志中存储的是完整分布，需要根据item_id索引
    rl_logs['propensity_b'] = rl_logs.apply(
        lambda row: row['propensity_distribution'][get_item_index(row['item_id'])],
        axis=1
    )
    
    # Ⅲ. 多方法评估
    results = {
        'strategy_name': strategy_config['name'],
        'sample_size': len(rl_logs)
    }
    
    evaluator = ImportanceSamplingEvaluator()
    
    for method in methods:
        if method == 'snips':
            est = evaluator.estimate_value_snips(
                rewards=rl_logs['reward'].values,
                action_log_probs=np.log(rl_logs['propensity_b'].values),
                target_action_probs=target_probs.values
            )
        elif method == 'wdr':
            dr_evaluator = DoublyRobustEvaluator()
            
            # 训练Q函数模型（使用behavior policy数据）
            contexts = np.array([json.loads(c)['features'] for c in rl_logs['context']])
            actions = rl_logs['item_id'].astype('category').cat.codes.values
            rewards = rl_logs['reward'].values
            
            dr_evaluator.fit_q_function(
                contexts, actions, rewards, np.log(rl_logs['propensity_b'].values)
            )
            
            est = dr_evaluator.estimate_value_wdr(
                contexts, actions, rewards,
                np.log(rl_logs['propensity_b'].values),
                target_probs.values
            )
        
        results[method] = est
    
    # Ⅳ. 结果保存
    with open(output_path, 'w') as f:
        json.dump(results, f, indent=2)
    
    return results

# 执行单个策略评估（示例）
rl_logs_sample = pd.read_parquet("s3://ope-data/rl_logs_v1/date=2024-01-15")
strategy = {"name": "din_attention", "type": "din"}

result = run_ope_evaluation(
    rl_logs=rl_logs_sample,
    strategy_config=strategy,
    methods=['snips', 'wdr'],
    output_path="s3://ope-results/din_attention.json"
)
print(f"DIN策略评估完成: {result}")

---

4.4 评估结果深度分析

15个策略的OPE评估结果：

策略名称	SNIPS估计	WDR估计	95% CI	效率提升	排名	是否上线
din_attention	+7.23%	+7.31%	[6.12%, 8.50%]	+35%	1	✅
deepfm_v1	+6.45%	+6.52%	[5.41%, 7.63%]	+28%	2	✅
din_wide	+5.98%	+6.01%	[4.91%, 7.11%]	+31%	3	✅
gbdt_baseline	+0.89%	+0.92%	[-0.12%, 1.96%]	+15%	8	❌
…	…	…	…	…	…	…

核心洞察：

1. OPE成功识别增量：Top-3策略的置信区间完全分离，说明差异统计显著，无需上线即可判断优劣。

2. 与离线AUC对比：传统离线AUC显示deepfm_v1（AUC=0.724）> din_attention（AUC=0.718），但OPE结果相反。原因在于DIN的注意力机制对长尾用户效果更好，而AUC过度拟合头部热门商品。

3. 方差缩减效果：WDR相比SNIPS平均降低35%方差，标准误从±1.2%降至±0.78%，这意味着评估同等精度所需样本量减少60%。

4. 冷启动策略评估：策略N专为新用户设计，OPE显示其**在新用户上+12.3%**提升，但在全量用户仅+2.1%。这精准识别了其适用场景。

---

4.5 线上验证与效果对比

上线验证流程：

仅将Top-3策略部署到5%流量进行7天A/B验证（而非传统14天50%流量）。

OPE预测 vs 在线真实效果：

策略	OPE预测	在线真实	绝对误差	相对误差	结论
din_attention	+7.31%	+7.52%	0.21pp	2.87%	✅ 高精度
deepfm_v1	+6.52%	+6.38%	0.14pp	2.15%	✅ 高精度
din_wide	+6.01%	+5.89%	0.12pp	2.00%	✅ 高精度

关键验证指标：

• OPE预测误差中位数：2.3%（业界通常为5-15%）
• 上线策略成功率：100%（3/3），传统流程为60%
• 迭代周期：从14天缩短至7天（OPE筛选+快速验证）

---

4.6 业务价值量化

成本节约：

成本项	传统方案	OPE方案	节约	年度影响
A/B测试流量成本	50%流量 × 14天	5%流量 × 7天	93%	节约¥420万/年
人力成本	8人 × 14天	3人 × 7天	85%	节约¥1,200万/年
机会成本	每周2个策略	每周20个策略	10倍探索	额外收益¥8,500万/年
回滚损失	40%策略回滚	5%策略回滚	88%	避免损失¥1,600万/年
总计	-	-	综合ROI 23.8倍	净收益¥1.07亿/年

---

战略价值：

• 创新能力解放：算法团队从"害怕测试"转为"大胆探索"，实验性策略提交增长300%。
• 敏捷响应：热点事件（如明星直播）可在2小时内完成策略定制+离线评估+上线，传统流程需2周。
• 人才留存：算法工程师因"想法能快速验证"，离职率从15%降至8%。

---