营销分析

引言 在数字化营销时代，企业在各个渠道投入大量预算，但如何准确衡量每个渠道的贡献？如何科学地评估营销ROI？本文将系统讲解营销归因和ROI分析的方法论。 一、营销归因模型 1.1 归因模型类型 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 营销归因模型: 单触点归因: 最后点击 (Last Click): 描述: 转化前最后一次触点获得100%功劳 优势: 简单易理解 劣势: 忽略其他触点贡献 适用: 短决策周期、单一渠道 首次点击 (First Click): 描述: 第一次触点获得100%功劳 优势: 重视获客渠道 劣势: 忽略后续培育 适用: 品牌认知、线索获取 末次点击 (Last Non-Direct Click): 描述: 排除直接访问的最后点击获100%功劳 优势: 更准确反映渠道价值 劣势: 仍然简单化 适用: 多渠道营销 多触点归因: 线性归因 (Linear): 描述: 所有触点平均分配功劳 公式: 100% / N (每个触点) 优势: 考虑所有触点 劣势: 忽略微妙差异 适用: 长决策周期 时间衰减 (Time Decay): 描述: 越接近转化的触点权重越高 公式: 指数衰减函数 优势: 反映时间影响 劣势: 可能忽略早期触点 适用: 较长决策周期 位置归因 (Position Based): 描述: 首尾触点40%+40%，中间20% 公式: 首40% + 中间平分20% + 尾40% 优势: 重视获客和转化 劣势: 固定比例可能不适用 适用: 品牌与转化并重 U型归因 (U-Shaped): 描述: 同位置归因（首尾各40%） 适用: 重视首次互动和最终转化 算法归因: 数据驱动 (Data-Driven): 描述: 基于历史数据计算权重 方法: 机器学习算法 优势: 最准确 劣势: 需要大量数据 适用: 大型企业、成熟数据体系 Shapley值: 描述: 博弈论方法计算边际贡献 方法: Shapley Value算法 优势: 理论上最公平 劣势: 计算复杂 适用: 精确分析需求 1.2 归因模型实现 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 # 营销归因模型实现 import pandas as pd import numpy as np from typing import List, Dict class MarketingAttribution: """营销归因分析""" def __init__(self, touchpoint_data): """ touchpoint_data: 触点数据 [{ 'user_id': 'user_001', 'journey': ['organic', 'social', 'email', 'direct'], 'conversion': 1, 'value': 99 }] """ self.data = touchpoint_data def last_click_attribution(self): """最后点击归因""" attribution_results = [] for journey_data in self.data: journey = journey_data['journey'] if len(journey) > 0 and journey_data['conversion']: last_touch = journey[-1] attribution_results.append({ 'user_id': journey_data['user_id'], 'attributed_channel': last_touch, 'value': journey_data['value'], 'model': 'last_click' }) df = pd.DataFrame(attribution_results) return df.groupby('attributed_channel')['value'].sum().to_dict() def first_click_attribution(self): """首次点击归因""" attribution_results = [] for journey_data in self.data: journey = journey_data['journey'] if len(journey) > 0 and journey_data['conversion']: first_touch = journey[0] attribution_results.append({ 'user_id': journey_data['user_id'], 'attributed_channel': first_touch, 'value': journey_data['value'], 'model': 'first_click' }) df = pd.DataFrame(attribution_results) return df.groupby('attributed_channel')['value'].sum().to_dict() def linear_attribution(self): """线性归因""" attribution_results = [] for journey_data in self.data: journey = journey_data['journey'] if len(journey) > 0 and journey_data['conversion']: value_per_touch = journey_data['value'] / len(journey) for touch in journey: attribution_results.append({ 'user_id': journey_data['user_id'], 'attributed_channel': touch, 'value': value_per_touch, 'model': 'linear' }) df = pd.DataFrame(attribution_results) return df.groupby('attributed_channel')['value'].sum().to_dict() def time_decay_attribution(self, decay_rate=0.5): """ 时间衰减归因 decay_rate: 衰减率 """ attribution_results = [] for journey_data in self.data: journey = journey_data['journey'] if len(journey) > 0 and journey_data['conversion']: # 计算权重（最近触点权重最高） positions = list(range(len(journey))) weights = [decay_rate ** (len(journey) - 1 - pos) for pos in positions] total_weight = sum(weights) # 归一化权重 normalized_weights = [w / total_weight for w in weights] # 分配价值 for i, touch in enumerate(journey): attributed_value = journey_data['value'] * normalized_weights[i] attribution_results.append({ 'user_id': journey_data['user_id'], 'attributed_channel': touch, 'value': attributed_value, 'model': 'time_decay' }) df = pd.DataFrame(attribution_results) return df.groupby('attributed_channel')['value'].sum().to_dict() def position_based_attribution(self, first_last_weight=0.4): """ 位置归因 first_last_weight: 首尾触点权重（默认40%） """ attribution_results = [] for journey_data in self.data: journey = journey_data['journey'] if len(journey) > 0 and journey_data['conversion']: journey_len = len(journey) if journey_len == 1: # 只有一个触点，获得100% weights = [1.0] elif journey_len == 2: # 两个触点，各50% weights = [0.5, 0.5] else: # 首尾各40%，中间平分20% middle_weight = (1 - 2 * first_last_weight) / (journey_len - 2) weights = [first_last_weight] + \ [middle_weight] * (journey_len - 2) + \ [first_last_weight] # 分配价值 for i, touch in enumerate(journey): attributed_value = journey_data['value'] * weights[i] attribution_results.append({ 'user_id': journey_data['user_id'], 'attributed_channel': touch, 'value': attributed_value, 'model': 'position_based' }) df = pd.DataFrame(attribution_results) return df.groupby('attributed_channel')['value'].sum().to_dict() def compare_models(self): """对比不同归因模型""" results = { 'last_click': self.last_click_attribution(), 'first_click': self.first_click_attribution(), 'linear': self.linear_attribution(), 'time_decay': self.time_decay_attribution(), 'position_based': self.position_based_attribution() } # 转换为DataFrame方便比较 comparison_df = pd.DataFrame(results).fillna(0) return comparison_df 二、ROI计算方法 2.1 基础ROI计算 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 # ROI计算工具 class ROICalculator: """ROI计算器""" @staticmethod def calculate_basic_roi(revenue, cost): """ 计算基础ROI revenue: 收入 cost: 成本 """ if cost == 0: return float('inf') roi = ((revenue - cost) / cost) * 100 return round(roi, 2) @staticmethod def calculate_roas(revenue, ad_spend): """ 计算ROAS (Return on Ad Spend) revenue: 收入 ad_spend: 广告花费 """ if ad_spend == 0: return float('inf') roas = revenue / ad_spend return round(roas, 2) @staticmethod def calculate_break_even_roas(profit_margin): """ 计算盈亏平衡ROAS profit_margin: 利润率 (0-1) """ if profit_margin == 0: return float('inf') break_even_roas = 1 / profit_margin return round(break_even_roas, 2) @staticmethod def calculate_ltv_cac_ratio(ltv, cac): """ 计算LTV/CAC比率 ltv: 用户生命周期价值 cac: 获客成本 """ if cac == 0: return float('inf') ratio = ltv / cac return round(ratio, 2) @staticmethod def calculate_payback_period(cac, monthly_arpu): """ 计算回本周期 cac: 获客成本 monthly_arpu: 月均每用户收入 """ if monthly_arpu == 0: return float('inf') payback_months = cac / monthly_arpu return round(payback_months, 1) def calculate_channel_roi(self, channel_data): """ 计算各渠道ROI channel_data: { 'channel_name': { 'spend': 广告花费, 'impressions': 展示次数, 'clicks': 点击次数, 'conversions': 转化次数, 'revenue': 收入 } } """ results = [] for channel, data in channel_data.items(): spend = data['spend'] revenue = data['revenue'] clicks = data['clicks'] conversions = data['conversions'] # 基础指标 roi = self.calculate_basic_roi(revenue, spend) roas = self.calculate_roas(revenue, spend) # 衍生指标 cpc = spend / clicks if clicks > 0 else 0 cpa = spend / conversions if conversions > 0 else 0 ctr = (clicks / data['impressions'] * 100) if data['impressions'] > 0 else 0 cvr = (conversions / clicks * 100) if clicks > 0 else 0 results.append({ 'channel': channel, 'spend': round(spend, 2), 'revenue': round(revenue, 2), 'profit': round(revenue - spend, 2), 'roi': f"{roi}%", 'roas': f"{roas}x", 'cpc': round(cpc, 2), 'cpa': round(cpa, 2), 'ctr': round(ctr, 2), 'cvr': round(cvr, 2) }) return pd.DataFrame(results) def calculate_marketing_mix_roi(self, marketing_data): """ 计算营销组合ROI marketing_data: { 'channels': 渠道数据, 'total_revenue': 总收入, 'baseline_revenue': 基线收入（无营销投入） } """ total_spend = sum(ch['spend'] for ch in marketing_data['channels'].values()) incremental_revenue = marketing_data['total_revenue'] - marketing_data['baseline_revenue'] marketing_roi = self.calculate_basic_roi(incremental_revenue, total_spend) marketing_roas = self.calculate_roas(incremental_revenue, total_spend) return { 'total_spend': round(total_spend, 2), 'total_revenue': round(marketing_data['total_revenue'], 2), 'baseline_revenue': round(marketing_data['baseline_revenue'], 2), 'incremental_revenue': round(incremental_revenue, 2), 'marketing_roi': f"{marketing_roi}%", 'marketing_roas': f"{marketing_roas}x" } 2.2 客户生命周期价值 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 # LTV计算模型 class LifetimeValueCalculator: """客户生命周期价值计算器""" def __init__(self, customer_data): """ customer_data: 客户数据 必需列: customer_id, purchase_date, revenue """ self.data = customer_data def calculate_simple_ltv(self, avg_purchase_value, purchase_frequency, lifespan): """ 简单LTV计算 avg_purchase_value: 平均购买金额 purchase_frequency: 购买频率（次/年） lifespan: 客户生命周期（年） """ ltv = avg_purchase_value * purchase_frequency * lifespan return round(ltv, 2) def calculate_advanced_ltv(self): """ 高级LTV计算（基于历史数据） """ # 计算每个客户的指标 customer_metrics = self.data.groupby('customer_id').agg({ 'revenue': ['sum', 'mean', 'count'], 'purchase_date': ['min', 'max'] }).reset_index() customer_metrics.columns = [ 'customer_id', 'total_revenue', 'avg_purchase_value', 'purchase_count', 'first_purchase', 'last_purchase' ] # 计算客户生命周期（天数） customer_metrics['lifespan_days'] = ( customer_metrics['last_purchase'] - customer_metrics['first_purchase'] ).dt.days + 1 # 计算年度购买频率 customer_metrics['purchase_frequency'] = ( customer_metrics['purchase_count'] / (customer_metrics['lifespan_days'] / 365) ).fillna(0) # 计算LTV customer_metrics['ltv'] = ( customer_metrics['avg_purchase_value'] * customer_metrics['purchase_frequency'] * (customer_metrics['lifespan_days'] / 365) ) # 整体指标 metrics = { 'avg_ltv': customer_metrics['ltv'].mean(), 'median_ltv': customer_metrics['ltv'].median(), 'avg_purchase_value': customer_metrics['avg_purchase_value'].mean(), 'avg_purchase_frequency': customer_metrics['purchase_frequency'].mean(), 'avg_lifespan_days': customer_metrics['lifespan_days'].mean() } return { 'customer_metrics': customer_metrics, 'summary': {k: round(v, 2) for k, v in metrics.items()} } def calculate_cohort_ltv(self): """同期群LTV分析""" # 获取首次购买月份 self.data['first_purchase_month'] = self.data.groupby('customer_id')['purchase_date'].transform('min').dt.to_period('M') # 按同期群和月份分组 cohort_revenue = self.data.groupby( ['first_purchase_month', 'purchase_date'] )['revenue'].sum().reset_index() # 计算累计收入 cohort_revenue['period_number'] = ( cohort_revenue['purchase_date'].dt.to_period('M') - cohort_revenue['first_purchase_month'] ).apply(lambda x: x.n) cohort_pivot = cohort_revenue.pivot( index='first_purchase_month', columns='period_number', values='revenue' ).cumsum(axis=1) return cohort_pivot def predict_ltv(self, features, target='ltv'): """ 预测LTV（使用机器学习） features: 客户特征数据 target: 目标变量 """ from sklearn.model_selection import train_test_split from sklearn.ensemble import RandomForestRegressor from sklearn.metrics import r2_score, mean_squared_error # 分割数据 X_train, X_test, y_train, y_test = train_test_split( features, target, test_size=0.2, random_state=42 ) # 训练模型 model = RandomForestRegressor( n_estimators=100, max_depth=10, random_state=42 ) model.fit(X_train, y_train) # 预测 y_pred = model.predict(X_test) # 评估 r2 = r2_score(y_test, y_pred) rmse = np.sqrt(mean_squared_error(y_test, y_pred)) return { 'model': model, 'r2_score': round(r2, 4), 'rmse': round(rmse, 2), 'feature_importance': dict(zip( features.columns, model.feature_importances_ )) } 三、预算优化策略 3.1 基于ROI的预算分配 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 # 预算优化算法 class BudgetOptimizer: """预算优化器""" def __init__(self, channel_performance): """ channel_performance: 渠道表现数据 [{ 'channel': '渠道名', 'spend': 投入, 'revenue': 收入, 'impressions': 展示, 'capacity': 最大容量 }] """ self.data = channel_performance def optimize_by_roi(self, total_budget): """ 基于ROI优化预算分配 策略: 按ROI比例分配预算 """ # 计算每个渠道的ROI for item in self.data: item['roi'] = ((item['revenue'] - item['spend']) / item['spend'] * 100) item['roas'] = item['revenue'] / item['spend'] # 过滤正ROI的渠道 positive_roi_channels = [ch for ch in self.data if ch['roi'] > 0] if not positive_roi_channels: return {'error': '没有正ROI的渠道'} # 计算总ROI权重 total_roi_weight = sum(ch['roi'] for ch in positive_roi_channels) # 分配预算 allocation = [] for channel in positive_roi_channels: allocated_budget = (channel['roi'] / total_roi_weight) * total_budget # 检查容量限制 if 'capacity' in channel and allocated_budget > channel['capacity']: allocated_budget = channel['capacity'] # 预估收入 estimated_revenue = allocated_budget * (1 + channel['roi'] / 100) allocation.append({ 'channel': channel['channel'], 'current_spend': channel['spend'], 'current_revenue': channel['revenue'], 'current_roi': round(channel['roi'], 2), 'allocated_budget': round(allocated_budget, 2), 'estimated_revenue': round(estimated_revenue, 2) }) total_estimated_revenue = sum(item['estimated_revenue'] for item in allocation) optimized_roi = ((total_estimated_revenue - total_budget) / total_budget) * 100 return { 'allocation': allocation, 'total_budget': total_budget, 'total_estimated_revenue': round(total_estimated_revenue, 2), 'optimized_roi': round(optimized_roi, 2), 'recommendation': self._get_roi_recommendation(optimized_roi) } def optimize_by_marginal_roi(self, total_budget, increment=1000): """ 基于边际ROI优化 策略: 逐步分配预算到边际ROI最高的渠道 """ remaining_budget = total_budget allocation = {channel['channel']: 0 for channel in self.data} while remaining_budget > 0: best_channel = None best_marginal_roi = -float('inf') # 寻找边际ROI最高的渠道 for channel in self.data: current_spend = allocation[channel['channel']] current_revenue = channel['revenue'] * (current_spend / channel['spend']) # 计算增加投入后的ROI new_spend = current_spend + increment if new_spend > channel.get('capacity', float('inf')): continue new_revenue = channel['revenue'] * (new_spend / channel['spend']) marginal_revenue = new_revenue - current_revenue marginal_roi = (marginal_revenue / increment) * 100 if marginal_roi > best_marginal_roi: best_marginal_roi = marginal_roi best_channel = channel['channel'] # 分配预算 if best_channel and best_marginal_roi > 0: allocation[best_channel] += increment remaining_budget -= increment else: # 没有正边际ROI的渠道，停止分配 break # 转换为结果格式 allocation_list = [] total_estimated_revenue = 0 for channel in self.data: channel_name = channel['channel'] allocated = allocation[channel_name] if allocated > 0: estimated_revenue = channel['revenue'] * (allocated / channel['spend']) total_estimated_revenue += estimated_revenue allocation_list.append({ 'channel': channel_name, 'allocated_budget': round(allocated, 2), 'estimated_revenue': round(estimated_revenue, 2), 'channel_roi': round(channel['roi'], 2) }) optimized_roi = ((total_estimated_revenue - total_budget) / total_budget) * 100 return { 'allocation': allocation_list, 'total_budget': total_budget, 'total_estimated_revenue': round(total_estimated_revenue, 2), 'optimized_roi': round(optimized_roi, 2) } def _get_roi_recommendation(self, roi): """获取ROI优化建议""" if roi > 300: return '优秀：扩大投入规模' elif roi > 200: return '良好：继续优化' elif roi > 100: return '一般：需要改进' else: return '差：重新评估策略' 3.2 多目标预算优化 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 # 多目标预算优化 from scipy.optimize import minimize import numpy as np class MultiObjectiveBudgetOptimizer: """多目标预算优化器""" def __init__(self, channels_data): """ channels_data: 渠道数据 [{ 'name': '渠道名', 'min_spend': 最小投入, 'max_spend': 最大容量, 'conversion_rate': 转化率, 'avg_order_value': 平均客单价 }] """ self.channels = channels_data def optimize(self, total_budget, objectives): """ 多目标优化 objectives: { 'maximize_revenue': 权重, 'maximize_conversions': 权重, 'minimize_cost': 权重 } """ n_channels = len(self.channels) # 决策变量：每个渠道的预算 x0 = np.array([total_budget / n_channels] * n_channels) # 约束条件 constraints = [ # 预算总和约束 {'type': 'eq', 'fun': lambda x: np.sum(x) - total_budget} ] # 每个渠道的预算范围 bounds = [] for channel in self.channels: bounds.append((channel['min_spend'], channel['max_spend'])) # 目标函数 def objective_function(x): total_revenue = 0 total_conversions = 0 for i, channel in enumerate(self.channels): spend = x[i] # 估算收入（基于历史数据的简化模型） estimated_conversions = spend * channel['conversion_rate'] estimated_revenue = estimated_conversions * channel['avg_order_value'] total_conversions += estimated_conversions total_revenue += estimated_revenue # 多目标加权和（转换为最小化问题） weighted_score = ( -objectives.get('maximize_revenue', 1) * total_revenue + -objectives.get('maximize_conversions', 1) * total_conversions + objectives.get('minimize_cost', 0) * total_budget ) return weighted_score # 优化 result = minimize( objective_function, x0, method='SLSQP', bounds=bounds, constraints=constraints ) # 解析结果 if result.success: allocation = [] total_revenue = 0 total_conversions = 0 for i, channel in enumerate(self.channels): spend = result.x[i] conversions = spend * channel['conversion_rate'] revenue = conversions * channel['avg_order_value'] total_conversions += conversions total_revenue += revenue allocation.append({ 'channel': channel['name'], 'allocated_budget': round(spend, 2), 'estimated_conversions': round(conversions), 'estimated_revenue': round(revenue, 2) }) roi = ((total_revenue - total_budget) / total_budget) * 100 cpa = total_budget / total_conversions if total_conversions > 0 else 0 return { 'success': True, 'allocation': allocation, 'total_budget': total_budget, 'estimated_revenue': round(total_revenue, 2), 'estimated_conversions': round(total_conversions), 'estimated_roi': round(roi, 2), 'estimated_cpa': round(cpa, 2) } else: return { 'success': False, 'error': result.message } 四、营销效果预测 4.1 时间序列预测 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 # 营销效果预测 import pandas as pd from sklearn.linear_model import LinearRegression from sklearn.metrics import mean_absolute_error, mean_squared_error class MarketingEffectPredictor: """营销效果预测器""" def __init__(self, historical_data): """ historical_data: 历史营销数据 { 'date': 日期, 'channel': 渠道, 'spend': 投入, 'impressions': 展示, 'clicks': 点击, 'conversions': 转化, 'revenue': 收入 } """ self.data = historical_data def predict_performance(self, channel, future_spend, periods=4): """ 预测未来表现 channel: 渠道名称 future_spend: 预期投入 periods: 预测周期数 """ # 过滤该渠道的历史数据 channel_data = self.data[self.data['channel'] == channel].copy() # 准备特征 channel_data['period'] = range(len(channel_data)) # 简单线性回归预测 X = channel_data[['period', 'spend']].values y_revenue = channel_data['revenue'].values y_conversions = channel_data['conversions'].values # 训练收入预测模型 revenue_model = LinearRegression() revenue_model.fit(X, y_revenue) # 训练转化预测模型 conversions_model = LinearRegression() conversions_model.fit(X, y_conversions) # 预测未来 future_periods = len(channel_data) + np.arange(periods) future_X = np.column_stack([future_periods, [future_spend] * periods]) predicted_revenue = revenue_model.predict(future_X) predicted_conversions = conversions_model.predict(future_X) # 计算预测指标 predictions = [] for i in range(periods): predictions.append({ 'period': i + 1, 'spend': future_spend, 'predicted_revenue': round(predicted_revenue[i], 2), 'predicted_conversions': round(predicted_conversions[i]), 'predicted_roi': round(((predicted_revenue[i] - future_spend) / future_spend) * 100, 2), 'predicted_cpa': round(future_spend / predicted_conversions[i], 2) if predicted_conversions[i] > 0 else 0 }) # 模型评估 y_revenue_pred = revenue_model.predict(X) y_conversions_pred = conversions_model.predict(X) model_metrics = { 'revenue_mae': mean_absolute_error(y_revenue, y_revenue_pred), 'revenue_rmse': np.sqrt(mean_squared_error(y_revenue, y_revenue_pred)), 'conversions_mae': mean_absolute_error(y_conversions, y_conversions_pred), 'conversions_rmse': np.sqrt(mean_squared_error(y_conversions, y_conversions_pred)) } return { 'channel': channel, 'predictions': predictions, 'model_metrics': {k: round(v, 2) for k, v in model_metrics.items()}, 'confidence': self._assess_confidence(model_metrics) } def _assess_confidence(self, metrics): """评估预测置信度""" if metrics['revenue_mae'] < 1000 and metrics['conversions_mae'] < 10: return 'High' elif metrics['revenue_mae'] < 5000 and metrics['conversions_mae'] < 50: return 'Medium' else: return 'Low' 五、营销仪表盘设计 5.1 核心指标监控 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 营销分析仪表盘指标: 第一层 - 执行指标: 实时监控: - 今日花费 - 今日展示次数 - 今日点击次数 - 今日转化次数 - 实时CTR/CVR 操作预警: - 预算使用率 - 异常流量 - 转化率下降 - CPA上升 第二层 - 效果指标: 渠道表现: - 各渠道ROI - 各渠道ROAS - 各渠道CPA - 各渠道转化率 趋势分析: - 周环比变化 - 月环比变化 - 同比变化 - 移动平均 第三层 - 业务指标: 获客指标: - 新增用户数 - CAC - LTV - LTV/CAC比率 收入指标: - 营销带来收入 - 营销ROI - 边际ROI - 盈亏平衡点 第四层 - 预测指标: 预测数据: - 预测本月收入 - 预测LTV - 预测转化率 - 预测置信度 优化建议: - 预算调整建议 - 渠道优化建议 - 创意优化建议 - 出价优化建议 总结 营销归因和ROI分析是数据驱动营销的核心： ...