马尔可夫链与去除效应在广告归因模型中的应用

1. 广告效果归因模型：马尔可夫链与去除效应

在数字营销领域，准确衡量各渠道对最终转化的贡献一直是个难题。传统的第一点击、最后点击或平均分配模型都存在明显缺陷，无法真实反映用户决策路径中各个触点的实际价值。马尔可夫链归因模型通过模拟用户状态转移过程，为这个问题提供了更科学的解决方案。

1.1 马尔可夫链建模原理

马尔可夫链的核心假设是"无记忆性"——下一个状态只取决于当前状态。在广告归因场景中，我们将用户的每个触点（如看到展示广告、点击搜索广告、收到邮件等）以及最终状态（转化或未转化）定义为不同的状态。

数学上，我们定义：

• 状态空间 S =
• 转移概率矩阵 P，其中 P[i][j] 表示从状态i转移到状态j的概率

通过分析大量用户的转化路径，我们可以统计出状态间的转移频次，进而计算出转移概率矩阵。例如，如果有1000次从"搜索广告"到"产品页访问"的转移，而"搜索广告"总共出现了5000次，那么对应的转移概率就是0.2。

1.2 去除效应计算与价值分配

去除效应的核心思想是：一个渠道的重要性可以通过移除该渠道后转化率的变化程度来衡量。具体计算步骤：

1. 计算原始转化率：通过求解马尔可夫链的稳态分布，得到最终到达"转化"状态的概率
2. 对于每个渠道i，构建新的转移矩阵P'，移除所有包含该渠道的转移
3. 计算新的转化率
4. 去除效应 = 1 - (新转化率 / 原始转化率)

最终，各渠道的贡献权重按其去除效应占总去除效应的比例分配。

实际应用中，我们通常会对长度超过7的路径进行截断处理，因为过长的路径往往包含噪声且计算复杂度呈指数增长。

1.3 工程实现与优化

在大规模应用中，我们需要考虑以下工程优化：

1. 路径采样：对于海量用户数据，可以使用采样技术减少计算量，同时保持统计显著性
2. 稀疏矩阵存储：转移矩阵通常非常稀疏，使用压缩存储格式可大幅减少内存占用
3. 增量更新：新数据到来时，只需更新相关转移计数，不必重新计算整个矩阵
4. 平滑处理：对于低频路径，应用加一平滑或其他平滑技术避免零概率问题

python复制class MarkovAttribution:
    def __init__(self, conversion_paths, max_path_length=7):
        self.paths = [p[:max_path_length] for p in conversion_paths]
        self.states = self._extract_states(self.paths)
        self.transition_counts = None
        
    def _extract_states(self, paths):
        states = set()
        for path in paths:
            states.update(path)
        return sorted(states)
    
    def build_transition_matrix(self):
        n = len(self.states)
        P = np.zeros((n, n))
        # 统计转移频次
        for path in self.paths:
            for i in range(len(path)-1):
                from_state = self.states.index(path[i])
                to_state = self.states.index(path[i+1])
                P[from_state][to_state] += 1
        # 加一平滑
        P = P + 1
        # 归一化
        row_sums = P.sum(axis=1, keepdims=True)
        P = np.divide(P, row_sums, where=row_sums!=0)
        return P
    
    def _steady_state_conversion(self, P):
        # 计算稳态分布
        eigenvalues, eigenvectors = np.linalg.eig(P.T)
        steady_state = np.real(eigenvectors[:, np.isclose(eigenvalues, 1)])
        steady_state = steady_state / steady_state.sum()
        conv_idx = self.states.index('conversion')
        return steady_state[conv_idx][0]
    
    def removal_effect(self, channel):
        P_original = self.build_transition_matrix()
        original_conv = self._steady_state_conversion(P_original)
        
        # 构建移除该渠道后的转移矩阵
        P_removed = P_original.copy()
        channel_idx = self.states.index(channel)
        # 将该渠道的转入和转出概率置为0
        P_removed[channel_idx, :] = 0
        P_removed[:, channel_idx] = 0
        # 重新归一化
        row_sums = P_removed.sum(axis=1, keepdims=True)
        P_removed = np.divide(P_removed, row_sums, where=row_sums!=0)
        
        new_conv = self._steady_state_conversion(P_removed)
        return 1 - new_conv / original_conv if original_conv > 0 else 0

1.4 实际应用中的挑战与解决方案

1. 数据稀疏性问题：

   • 现象：某些路径组合出现频率极低
   • 解决：使用回溯窗口限制路径长度，或采用层次化建模将稀有路径归类

2. 跨设备归因：

   • 现象：用户可能在不同设备上与广告互动
   • 解决：结合用户识别技术，构建跨设备统一视图

3. 时间衰减效应：

   • 现象：较早的触点影响力可能随时间衰减
   • 解决：在转移概率中引入时间衰减因子

4. 渠道协同效应：

   • 现象：某些渠道组合效果优于单独效果之和
   • 解决：在模型中引入高阶转移（考虑前几个触点的组合）

2. 广告预算分配模型：线性规划与预测结合

广告主面临的核心挑战之一是如何在有限的预算下，将资金分配到各个渠道以获得最大回报。这本质上是一个带约束的优化问题，线性规划提供了数学上的完美解决方案。

2.1 问题建模与数学表达

预算分配问题可以形式化为：

最大化：∑(c_i * x_i)
约束条件：

1. ∑x_i ≤ B （总预算限制）
2. x_i ≥ 0 （非负分配）
3. 可选：x_i ≥ m_i （渠道最低预算要求）

其中：

• c_i 是渠道i的预测转化率（每元投入带来的转化量）
• x_i 是分配给渠道i的预算
• B 是总预算

2.2 转化率预测模型

准确的转化率预测是预算分配的基础。我们通常使用时序模型预测各渠道未来表现：

1. 数据准备：

   • 收集各渠道历史投放数据和对应转化
   • 计算历史转化率（转化量/投入）
   • 加入季节性、节假日等外部变量

2. 模型选择：

   • 简单场景：ARIMA、指数平滑
   • 复杂场景：Prophet、LSTM
   • 极端稀疏数据：贝叶斯结构时间序列

3. 评估指标：

   • MAE（平均绝对误差）
   • MAPE（平均绝对百分比误差）
   • 业务指标：预算分配后的实际ROI

python复制class ConversionRatePredictor:
    def __init__(self, model_type='prophet'):
        self.model_type = model_type
        self.models = {}
        
    def fit(self, channel_data):
        """训练各渠道的转化率预测模型"""
        for channel, data in channel_data.items():
            if self.model_type == 'prophet':
                model = Prophet()
                model.fit(data)
            elif self.model_type == 'arima':
                model = auto_arima(data)
            self.models[channel] = model
    
    def predict(self, channels, periods=7):
        """预测未来periods天的转化率"""
        predictions = {}
        for channel in channels:
            if self.model_type == 'prophet':
                future = self.models[channel].make_future_dataframe(periods=periods)
                forecast = self.models[channel].predict(future)
                pred = forecast['yhat'].values[-periods:]
            elif self.model_type == 'arima':
                pred = self.models[channel].predict(n_periods=periods)
            predictions[channel] = pred.mean()
        return predictions

2.3 线性规划求解

有了转化率预测后，我们可以构建线性规划问题并求解：

python复制from scipy.optimize import linprog

class BudgetAllocator:
    def __init__(self, channels, total_budget, min_budget=None):
        self.channels = channels
        self.total_budget = total_budget
        self.min_budget = min_budget or {ch:0 for ch in channels}
        
    def allocate(self, conversion_rates):
        # 目标函数系数（求最小化，所以取负）
        c = [-conversion_rates[ch] for ch in self.channels]
        
        # 预算约束：所有渠道预算之和 ≤ 总预算
        A_ub = [[1] * len(self.channels)]
        b_ub = [self.total_budget]
        
        # 各渠道最低预算约束
        bounds = [(self.min_budget[ch], None) for ch in self.channels]
        
        # 求解
        res = linprog(c, A_ub=A_ub, b_ub=b_ub, bounds=bounds, method='highs')
        
        if res.success:
            allocation = {ch: res.x[i] for i, ch in enumerate(self.channels)}
            # 由于浮点计算，确保总和正好等于总预算
            total = sum(allocation.values())
            allocation = {k: v/total * self.total_budget for k, v in allocation.items()}
            return allocation
        else:
            raise ValueError("Optimization failed: " + res.message)

2.4 动态调整机制

市场环境不断变化，静态的预算分配很快就会失效。我们需要建立动态调整机制：

1. 监控频率：

   • 高频渠道（如程序化广告）：每小时监控
   • 低频渠道（如电视广告）：每日/每周监控

2. 调整策略：

   • 小幅度调整：根据表现微调预算比例
   • 重新分配：当某些渠道持续表现不佳时，将预算转移到其他渠道
   • 探索性预算：保留小部分预算用于测试新渠道

3. 安全机制：

   • 单日最大调整幅度限制
   • 渠道最低预算保护
   • 异常检测避免错误调整

实际应用中，我们会设置"影子模式"——同时运行新旧两种分配策略但只按旧策略执行，对比两者效果后再决定是否切换。

3. 用户细分模型：聚类算法实践指南

有效的用户细分是精准营销的基础。通过将用户划分为具有相似特征和行为的群体，我们可以制定更有针对性的广告策略，提高营销效率。

3.1 特征工程：构建用户画像

高质量的特征是聚类成功的前提。常见的用户特征包括：

1. 行为特征：

   • 最近一次互动时间（Recency）
   • 互动频率（Frequency）
   • 互动深度（Depth）
   • 偏好类别/产品

2. 消费特征：

   • 历史消费金额
   • 平均订单价值
   • 折扣敏感度
   • 支付方式偏好

3. 人口统计特征（如有）：

   • 年龄段
   • 性别
   • 地理位置

4. 设备特征：

   • 使用设备类型
   • 操作系统
   • 网络环境

python复制def build_user_features(raw_data):
    """从原始数据构建用户特征"""
    features = {}
    
    # RFM特征
    features['recency'] = (datetime.now() - raw_data['last_interaction']).days
    features['frequency'] = raw_data['interaction_count_30d']
    features['monetary'] = raw_data['total_spend_90d']
    
    # 行为特征
    features['avg_session_duration'] = raw_data['total_duration'] / raw_data['session_count']
    features['preferred_category'] = raw_data['most_visited_category']
    
    # 设备特征
    features['is_mobile'] = 1 if raw_data['device_type'] == 'mobile' else 0
    
    # 标准化处理
    scaler = StandardScaler()
    scaled_features = scaler.fit_transform(pd.DataFrame([features]))
    
    return scaled_features

3.2 降维处理：PCA与t-SNE

高维数据直接聚类效果往往不佳，我们需要先进行降维：

1. PCA（主成分分析）：

   • 线性降维方法
   • 保留最大方差的方向
   • 计算高效，适合作为预处理步骤

2. t-SNE：

   • 非线性降维
   • 擅长保留局部结构
   • 可视化效果好但计算成本高

python复制from sklearn.decomposition import PCA
from sklearn.manifold import TSNE

class DimensionalityReducer:
    def __init__(self, method='pca', n_components=5):
        self.method = method
        self.n_components = n_components
        if method == 'pca':
            self.model = PCA(n_components=n_components)
        elif method == 'tsne':
            self.model = TSNE(n_components=n_components)
            
    def fit_transform(self, X):
        # 先使用PCA初步降维到50维，再应用t-SNE
        if self.method == 'tsne' and X.shape[1] > 50:
            X = PCA(n_components=50).fit_transform(X)
        return self.model.fit_transform(X)

3.3 K-means聚类实战

K-means是最常用的聚类算法，但其实施有几个关键点需要注意：

1. 确定最佳K值：

   • 肘部法则（观察SSE下降拐点）
   • 轮廓系数
   • 业务解释性

2. 初始化改进：

   • K-means++初始化
   • 多次随机初始化取最优

3. 空簇处理：

   • 重新分配最远点
   • 移除空簇

python复制from sklearn.cluster import KMeans
from sklearn.metrics import silhouette_score

class UserSegmenter:
    def __init__(self, max_clusters=10):
        self.max_clusters = max_clusters
        self.best_k = None
        self.best_model = None
        
    def find_optimal_k(self, X):
        silhouette_scores = []
        for k in range(2, self.max_clusters+1):
            kmeans = KMeans(n_clusters=k, init='k-means++', n_init=10)
            labels = kmeans.fit_predict(X)
            score = silhouette_score(X, labels)
            silhouette_scores.append(score)
        
        self.best_k = np.argmax(silhouette_scores) + 2  # 因为k从2开始
        self.best_model = KMeans(n_clusters=self.best_k, init='k-means++', n_init=10)
        return self.best_k
    
    def fit(self, X):
        if self.best_k is None:
            self.find_optimal_k(X)
        self.best_model.fit(X)
        return self.best_model.labels_
    
    def describe_clusters(self, X, labels):
        """分析各簇特征"""
        df = pd.DataFrame(X)
        df['cluster'] = labels
        
        cluster_profiles = {}
        for c in range(self.best_k):
            profile = {
                'size': (labels == c).sum(),
                'means': df[df['cluster'] == c].mean().to_dict(),
                'stds': df[df['cluster'] == c].std().to_dict()
            }
            cluster_profiles[c] = profile
        
        return cluster_profiles

3.4 聚类结果分析与应用

获得聚类结果后，我们需要：

1. 分析簇特征：

   • 计算各簇的中心点和分布
   • 识别显著差异特征
   • 可视化关键特征分布

2. 业务解读：

   • 为每个簇创建人物画像
   • 例如："高价值忠诚用户"、"价格敏感新用户"等

3. 营销策略制定：

   • 针对不同簇设计个性化广告
   • 制定差异化的沟通策略
   • 优化渠道投放组合

4. 持续监控与迭代：

   • 定期重新聚类观察变化
   • 跟踪各簇的转化表现
   • 调整簇数量和特征组成

实际应用中，我们会使用平行坐标系图等可视化工具帮助业务人员理解各簇特征差异。

4. 广告竞价策略：强化学习实战

实时竞价(RTB)是数字广告的核心环节，如何在毫秒级响应时间内做出最优出价决策极具挑战性。上下文Bandit算法通过平衡探索与利用，为这个问题提供了有效解决方案。

4.1 上下文Bandit算法原理

LinUCB算法是上下文Bandit的典型代表，其核心思想是：

1. 对每个动作a（出价策略），维护一个参数向量θ_a
2. 对于有特征x的请求，预测回报：r = x^T θ_a
3. 选择上界最大的动作：a = argmax(x^T θ_a + α√(x^T A_a^{-1} x))
   • 第一项：预期回报（利用）
   • 第二项：不确定性（探索）

其中：

• A_a = ∑x_i x_i^T + I （正则化设计矩阵）
• b_a = ∑r_i x_i （观测回报向量）
• θ_a = A_a^{-1} b_a （闭式解）

4.2 特征工程：构建竞价上下文

有效的上下文特征应包括：

1. 用户特征：

   • 历史行为
   • 人口统计信息
   • 设备信息

2. 广告位特征：

   • 网站/APP类别
   • 位置尺寸
   • 可见性评分

3. 环境特征：

   • 时间段
   • 网络类型
   • 地理位置

4. 历史表现特征：

   • 该广告位历史CTR
   • 同类用户转化率

python复制def extract_bidding_context(ad_request, user_profile):
    """从广告请求和用户画像中提取特征"""
    context = {}
    
    # 用户特征
    context['user_activity'] = user_profile['activity_level']
    context['user_value'] = user_profile['predicted_ltv']
    
    # 广告位特征
    context['ad_position'] = ad_request['position']
    context['ad_size'] = ad_request['width'] * ad_request['height']
    
    # 环境特征
    context['hour_of_day'] = ad_request['hour']
    context['connection_type'] = 1 if ad_request['connection'] == 'wifi' else 0
    
    # 历史表现
    context['avg_ctr'] = ad_request['site']['avg_ctr']
    
    # 转换为向量
    feature_names = ['user_activity', 'user_value', 'ad_position', 
                    'ad_size', 'hour_of_day', 'connection_type', 'avg_ctr']
    feature_vector = [context[name] for name in feature_names]
    
    # 标准化
    feature_vector = (feature_vector - mean) / std
    
    return np.array(feature_vector)

4.3 LinUCB实现与优化

基础版LinUCB实现：

python复制class LinUCB:
    def __init__(self, n_actions, context_dim, alpha=0.1):
        self.alpha = alpha
        self.n_actions = n_actions
        self.context_dim = context_dim
        
        # 为每个动作维护这些变量
        self.A = [np.eye(context_dim) for _ in range(n_actions)]
        self.b = [np.zeros(context_dim) for _ in range(n_actions)]
        self.theta = [np.zeros(context_dim) for _ in range(n_actions)]
    
    def select_action(self, context):
        scores = []
        for a in range(self.n_actions):
            A_inv = np.linalg.inv(self.A[a])
            self.theta[a] = A_inv @ self.b[a]
            
            # 计算UCB分数
            score = self.theta[a].T @ context + \
                   self.alpha * np.sqrt(context.T @ A_inv @ context)
            scores.append(score)
        
        return np.argmax(scores)
    
    def update(self, context, action, reward):
        # 更新统计量
        self.A[action] += np.outer(context, context)
        self.b[action] += reward * context

针对高并发场景的优化：

1. 并行计算：

   • 各动作的分数计算相互独立，可并行化
   • 使用多线程或GPU加速

2. 稀疏更新：

   • 对非零特征维度进行选择性更新
   • 减少矩阵运算量

3. 模型分片：

   • 按广告类别或用户群体分片
   • 每个分片独立维护模型

4. 近似计算：

   • 使用迭代法近似求解线性系统
   • 避免精确求逆的高计算成本

4.4 奖励设计与模型评估

合理的奖励设计对模型效果至关重要：

1. 奖励函数选择：

   • 直接转化：1（转化），0（无转化）
   • 渐进式奖励：点击(0.3)，加入购物车(0.6)，购买(1.0)
   • 利润导向：实际产生的毛利

2. 延迟奖励处理：

   • 转化可能有延迟（几天甚至几周）
   • 使用奖励回放缓冲区
   • 实现部分信用分配

3. 评估指标：

   • 短期：CTR、CPC
   • 中期：转化率、ROAS
   • 长期：用户LTV变化

4. 离线评估：

   • 回放历史数据
   • 计算反事实表现
   • 重要性采样校正

生产环境中，我们会运行A/B测试：将小部分流量随机分配，其余按模型分配，比较两组表现。