Perform multi-touch attribution analysis using Markov chains, Shapley values, and custom attribution models. Use when you need to analyze marketing channel effectiveness, calculate conversion attribution, optimize marketing budgets, or understand customer journey paths. Supports channel transition analysis, ROI calculation, and marketing optimization insights with Chinese language support.
一个用于营销渠道归因分析和效果评估的综合性Claude Code技能，支持多种归因模型和可视化分析。
pip install -r requirements.txt
from core_attribution import AttributionAnalyzer
from attribution_visualizer import AttributionVisualizer
# 1. 数据分析和归因
analyzer = AttributionAnalyzer()
results = analyzer.run_complete_analysis('marketing_data.csv')
# 2. 可视化分析
visualizer = AttributionVisualizer()
visualizer.create_attribution_dashboard(results)
from markov_chains import MarkovChainAttributor
# 1. 构建马尔可夫链模型
markov_attributor = MarkovChainAttributor()
markov_results = markov_attributor.run_complete_markov_analysis(customer_paths)
# 2. 生成马尔可夫可视化
visualizer.create_markov_visualization(markov_results)
from shapley_values import ShapleyValueAttributor
# 1. 计算Shapley值
shapley_attributor = ShapleyValueAttributor()
shapley_results = shapley_attributor.run_complete_shapley_analysis(customer_paths)
# 2. 生成Shapley可视化
visualizer.create_shapley_visualization(shapley_results)
user_id,timestamp,channel,conversion_status,conversion_value,cost
USER001,2024-01-15T10:30:00Z,paid_search,0,0,50
USER001,2024-01-16T14:20:00Z,social_media,0,0,30
USER001,2024-01-18T09:15:00Z,email,1,1000,10
attribution_results.csv: 完整归因分析结果channel_performance.csv: 渠道性能指标和ROI分析customer_paths.csv: 重构的客户旅程路径transition_matrix.csv: 马尔可夫链转移概率矩阵attribution_dashboard.png: 综合归因分析仪表板markov_analysis.png: 马可夫链分析可视化shapley_analysis.png: Shapley值分析可视化channel_network_graph.png: 渠道转换网络图attribution_report.md: 详细归因分析报告optimization_recommendations.md: 渠道优化建议报告class AttributionAnalyzer:
def load_and_validate_data() # 数据加载与验证
def build_customer_paths() # 构建客户路径
def first_touch_attribution() # 首次接触归因
def last_touch_attribution() # 最后接触归因
def linear_attribution() # 线性归因
def time_decay_attribution() # 时间衰减归因
def position_based_attribution() # 位置归因
def compare_attribution_models() # 模型比较分析
class MarkovChainAttributor:
def build_transition_matrix() # 构建转移矩阵
def calculate_removal_effects() # 计算移除效应
def calculate_attribution_weights() # 计算归因权重
def analyze_channel_transitions() # 渠道转换分析
def build_channel_graph() # 构建渠道转换图
def simulate_attribution_scenarios() # 场景模拟分析
class ShapleyValueAttributor:
def calculate_shapley_values() # 计算Shapley值
def calculate_channel_synergy() # 计算渠道协同效应
def analyze_marginal_contributions() # 分析边际贡献
def optimize_channel_mix() # 优化渠道组合
def run_complete_shapley_analysis() # 完整Shapley分析
class AttributionVisualizer:
def create_attribution_dashboard() # 综合分析仪表板
def create_markov_visualization() # 马尔可夫链可视化
def create_shapley_visualization() # Shapley值可视化
def _plot_attribution_weights_comparison() # 归因权重对比
def _plot_channel_performance_metrics() # 渠道性能指标
def _plot_journey_path_analysis() # 客户旅程分析
# 扩展基础归因分析器
class CustomAttributor(AttributionAnalyzer):
def custom_weighted_attribution(self, paths_df, weight_config):
"""自定义权重归因模型"""
# 实现自定义权重逻辑
pass
def business_rule_attribution(self, paths_df, business_rules):
"""业务规则归因模型"""
# 基于业务规则的归因逻辑
pass
# 运行所有可用模型
results = analyzer.compare_attribution_models(paths_df)
# 选择最佳模型
best_model = self._select_best_model(results)
# 基于归因结果优化预算
optimization_results = analyzer.generate_attribution_summary(df)
# 获取优化建议
for recommendation in optimization_results['recommended_actions']:
print(f"{recommendation['type']}: {recommendation['channel']}")
| 模型类型 | 优势 | 劣势 | 适用场景 | |----------|------|------|----------| | 首次接触 | 简单直接 | 忽略后续触点 | 新产品推广 | | 最后接触 | 考虑最终决策 | 忽略前期影响 | 销售转化 | | 线性 | 公平分配 | 不考虑差异 | 平均效果 | | 时间衰减 | 考虑时间因素 | 需要调参 | 时间敏感 | | 马可夫链 | 动态概率 | 计算复杂 | 复杂路径 | | Shapley值 | 理论最优 | 计算量大 | 精确归因 |
用户标识不一致
# 标准化用户ID
df['user_id'] = df['user_id'].astype(str).str.lower()
时间戳格式问题
# 转换时间戳格式
df['timestamp'] = pd.to_datetime(df['timestamp'], errors='coerce')
数据量过大
# 分批处理大数据
chunk_size = 10000
for chunk in pd.read_csv('large_data.csv', chunksize=chunk_size):
# 处理每个数据块
pass
模型计算时间过长
# 对于Shapley值计算，限制渠道数量
channels = df['channel'].unique()[:10] # 只分析前10个渠道
# 使用更高效的数据类型
dtypes = {
'user_id': 'category',
'channel': 'category',
'cost': 'float32'
}
df = pd.read_csv('data.csv', dtype=dtypes)
# 多进程处理Shapley值计算
from multiprocessing import Pool
with Pool(processes=4) as pool:
results = pool.map(calculate_shapley_for_channel, channels)
由归因分析与建模系统支持 | Powered by Attribution Analysis Engine
npx skills add liangdabiao/attribution-analysis-modeling下载完整 Skill 目录，包含 SKILL.md 及所有相关文件
Category:business