数据分析大师

# 深层认知偏差库 (Cognitive Biases in Data Analysis)

> 数据分析最危险的陷阱不是计算错误，而是认知偏差——让你得出"正确的错误答案"。

---

## 一、为什么要学习认知偏差

### 大师与普通分析师的区别

```
普通分析师：数据说什么，我就报告什么
大师分析师：数据可能在骗我，让我先检验一下
```

### 认知偏差的危害

| 偏差类型 | 危害 |
|---------|------|
| 统计陷阱 | 数据本身就是扭曲的 |
| 采样偏差 | 样本不代表总体 |
| 因果谬误 | 把相关当因果 |
| 心理偏误 | 分析师自己带入的偏见 |

---

## 二、统计陷阱

### 1. 辛普森悖论 (Simpson's Paradox)

**定义**：分组数据和整体数据呈现相反趋势。

**经典案例**：
```
某医院两种疗法对比：

整体数据：
疗法A：治愈率 80%
疗法B：治愈率 90%
→ 结论：疗法B更好？

分组后（按病情轻重）：

轻症患者：
疗法A：治愈率 95%（100人中95人）
疗法B：治愈率 90%（900人中810人）

重症患者：
疗法A：治愈率 50%（900人中450人）
疗法B：治愈率 30%（100人中30人）

真相：
- 疗法A在轻症和重症中都更好
- 但因为疗法A被更多用于重症，整体数据反而更差
- 正确结论：疗法A更好
```

**检验方法**：
- 对关键结论进行分组验证
- 问：是否存在隐藏的混淆变量？
- 问：不同分组的样本量分布是否均匀？

**应用场景**：
- 产品/渠道效果对比
- A/B测试结果分析
- 绩效评估

---

### 2. 幸存者偏差 (Survivorship Bias)

**定义**：只看到"存活"下来的样本，忽略了"消失"的样本。

**经典案例**：
```
二战飞机加固案例：
- 军方统计返航飞机的弹孔分布
- 发现机翼和机身弹孔多，驾驶舱弹孔少
- 军方想法：加固机翼和机身
- 正确答案：加固驾驶舱
- 原因：驾驶舱中弹的飞机没能返航
```

**数据分析中的表现**：
```
❌ "成功创业者都辍学，所以辍学有助于创业"
   → 忽略了大量辍学后失败的人

❌ "我们的用户满意度很高（来自反馈表）"
   → 不满意的用户可能已经流失，不会填反馈

❌ "老用户ARPU比新用户高，所以用户会越来越值钱"
   → 低ARPU用户可能已经流失了
```

**检验方法**：
- 问：有没有我看不到的数据？
- 问：什么样的对象不会进入我的样本？
- 主动寻找"流失/失败/消失"的数据

---

### 3. 基准率谬误 (Base Rate Neglect)

**定义**：忽略事件本身的基础概率，过度关注条件概率。

**经典案例**：
```
某疾病检测准确率99%
患病率 0.1%（千分之一）

如果你检测阳性，真正患病的概率是多少？

直觉答案：99%
实际答案：约9%

计算：
1000人中：1人真患病，999人健康
检测结果：1人真阳性 + 9.99人假阳性 ≈ 11人阳性
真患病概率：1/11 ≈ 9%
```

**数据分析中的表现**：
```
❌ "这个用户有高风险特征，一定会流失"
   → 但如果基础流失率只有1%，即使是高风险用户也可能不流失

❌ "营销活动转化率达到10%！"
   → 如果没有活动转化率也是8%，活动效果被高估了
```

**检验方法**：
- 始终先了解基础概率（基准率）
- 用贝叶斯思维：先验 × 证据强度 = 后验
- 对比有/无干预的基础数据

---

### 4. 回归均值 (Regression to the Mean)

**定义**：极端表现之后，往往会向平均水平回归。

**经典案例**：
```
飞行员训练案例：
- 教练发现：表扬好表现后，学员下次表现变差
- 教练发现：批评差表现后，学员下次表现变好
- 教练结论：批评比表扬有效

实际原因：
- 好表现本身就是偶然偏高，下次自然回落
- 差表现本身就是偶然偏低，下次自然回升
- 与表扬/批评无关，是统计规律
```

**数据分析中的表现**：
```
❌ "上月销冠这个月业绩下滑，一定是松懈了"
   → 可能只是从偶然高点回归正常

❌ "我们的干预措施让异常值恢复正常了"
   → 异常值本来就会回归，不一定是干预的功劳

❌ "新策略让最差的店铺提升了"
   → 最差的店铺本来就有更大的回升空间
```

**检验方法**：
- 对极端值保持警惕
- 用多期数据而非单期数据
- 设置对照组来排除自然回归

---

### 5. 多重比较问题 (Multiple Comparisons)

**定义**：做足够多的比较，总会偶然发现"显著"结果。

**案例说明**：
```
如果显著性水平设为5%（p<0.05）：
- 做1次比较，假阳性概率 5%
- 做20次比较，至少1次假阳性概率 64%
- 做100次比较，几乎必然出现假阳性
```

**数据分析中的表现**：
```
❌ "我们分析了50个维度，发现年龄25-28岁+使用iPhone+
    居住在二线城市的用户转化率显著更高！"
   → 做了太多比较，很可能是假阳性

❌ "不断切数据，直到找到一个好看的切法"
   → 数据挖掘偏差
```

**检验方法**：
- 区分"探索性分析"和"验证性分析"
- 探索发现的结论需要新数据验证
- 使用Bonferroni校正等方法调整显著性水平
- 记录所有做过的比较，不只是显著的那些

---

## 三、采样与选择偏差

### 6. 选择偏差 (Selection Bias)

**定义**：样本选择方式导致样本不代表总体。

**常见类型**：

| 类型 | 说明 | 示例 |
|-----|------|------|
| 自选择偏差 | 样本自己选择是否参与 | 满意度调查只有愿意填的人填 |
| 无应答偏差 | 部分人群系统性不响应 | 电话调查遗漏年轻人 |
| 便利抽样 | 选最容易获取的样本 | 只分析活跃用户 |
| 时间选择偏差 | 选择性地选取时间段 | 只看旺季数据得出乐观结论 |

**检验方法**：
- 问：谁会进入样本？谁不会？
- 检查样本与总体的分布差异
- 考虑加权或分层抽样

---

### 7. 信息偏差 (Information Bias)

**定义**：数据收集方式本身引入的偏差。

**常见类型**：

| 类型 | 说明 |
|-----|------|
| 测量偏差 | 测量工具不准确 |
| 回忆偏差 | 被调查者回忆不准确 |
| 报告偏差 | 被调查者选择性报告 |
| 观察者偏差 | 数据收集者的主观影响 |

**示例**：
```
❌ "用户自报的收入数据"
   → 高收入倾向于低报，低收入倾向于高报

❌ "用户说他们会购买这个新功能"
   → 说和做是两回事

❌ "根据用户投诉数据，A功能问题最多"
   → 可能只是A功能用户更愿意投诉
```

---

### 8. 健康工人效应 (Healthy Worker Effect)

**定义**：在职人群比一般人群更健康，导致职业风险被低估。

**泛化应用**：
```
活跃用户比流失用户更可能有正向特征
   → 但这不代表"让用户活跃起来"就能获得这些特征

付费用户比免费用户更有价值
   → 但不意味着让免费用户付费就能变得同样有价值

愿意参加调研的用户更满意
   → 调研结果高估了整体满意度
```

---

## 四、因果谬误

### 9. 相关≠因果

**核心警示**：
```
A和B同时变化，可能的原因：
1. A导致B
2. B导致A
3. C同时导致A和B（混淆变量）
4. 纯粹巧合
```

**经典案例**：
```
❌ "冰淇淋销量↑ 与 溺水事故↑ 相关，所以冰淇淋导致溺水"
   → 混淆变量：夏天（天热→冰淇淋销量↑，天热→游泳多→溺水↑）

❌ "高活跃用户付费率高，所以提高活跃度能提升付费"
   → 可能方向反了：愿意付费的用户本来就更活跃
   → 或者：高收入用户同时更活跃也更愿意付费
```

**建立因果关系的方法**：
1. 随机对照实验（A/B测试）
2. 自然实验/断点回归
3. 工具变量法
4. 双重差分法

---

### 10. 反向因果 (Reverse Causality)

**定义**：把因果方向搞反了。

**示例**：
```
❌ "优秀员工都参加了培训，所以培训让员工变优秀"
   → 可能是优秀员工更愿意参加培训

❌ "高消费用户都是VIP，所以VIP身份促进消费"
   → VIP是因为消费高才给的身份

❌ "多次使用某功能的用户留存更好，所以推广该功能能提高留存"
   → 可能是本来就想留下的用户才会多用功能
```

---

### 11. 遗漏变量偏差 (Omitted Variable Bias)

**定义**：遗漏了重要的影响因素，导致错误归因。

**示例**：
```
分析：广告投放↑ → 销售↑ → 结论：广告有效

遗漏变量：季节
实际情况：Q4旺季 → 同时增加广告投放 + 销售自然上涨
如果不控制季节因素，会高估广告效果

正确做法：同比对比，或使用对照组
```

---

## 五、心理偏误

### 12. 确认偏误 (Confirmation Bias)

**定义**：倾向于寻找支持自己观点的证据，忽略反面证据。

**表现**：
```
❌ 先有结论，再找数据支持
❌ 只展示支持自己观点的图表
❌ 对不符合预期的数据找理由解释掉
❌ 分析到支持预设结论就停止
```

**对抗方法**：
- 主动寻找反面证据
- 问："什么数据能推翻我的结论？"
- 让持不同意见的人审查分析
- 先做分析，再形成结论

---

### 13. 锚定效应 (Anchoring Effect)

**定义**：第一个看到的数字会影响后续判断。

**表现**：
```
❌ "去年增长50%，今年只增长20%，很差"
   → 被去年的数字锚定了，20%可能已经很好

❌ 先看到一个错误的预测，后续分析都围绕它调整
```

**对抗方法**：
- 独立思考，不先看别人的分析
- 多用不同的参照系
- 质疑"参照物"本身是否合理

---

### 14. 可得性偏误 (Availability Heuristic)

**定义**：容易想起的例子被认为更常见。

**表现**：
```
❌ "最近看到好几个用户投诉这个问题，一定很严重"
   → 可能只是个案更容易被记住

❌ "我身边的人都喜欢这个功能，用户一定都喜欢"
   → 你身边的人不代表用户整体
```

**对抗方法**：
- 用数据说话，不靠印象
- 区分"印象中的频率"和"实际频率"
- 主动寻找不同类型的案例

---

### 15. 后见之明偏误 (Hindsight Bias)

**定义**：事后觉得"早就知道会这样"。

**表现**：
```
❌ "这个项目失败是显而易见的"（但当时没人预测到）
❌ "数据明显显示会下跌"（但用同样的数据你当时没看出来）
```

**危害**：
- 低估决策时的不确定性
- 无法从失败中学习真正的教训
- 对未来预测过度自信

---

## 六、检验清单

### 每次分析完成前的偏差检验

```markdown
## 偏差检验清单

### 数据层面
- [ ] 样本是否代表总体？谁被遗漏了？
- [ ] 数据收集方式是否引入偏差？
- [ ] 是否存在幸存者偏差？
- [ ] 时间段选择是否有偏？

### 分析层面
- [ ] 是否做了过多比较导致假阳性？
- [ ] 分组后结论是否仍然成立？（辛普森悖论）
- [ ] 极端值是否会回归均值？
- [ ] 基准率是多少？

### 因果层面
- [ ] 相关性是否被当成因果？
- [ ] 因果方向是否正确？
- [ ] 是否有遗漏的混淆变量？
- [ ] 能否用实验验证因果？

### 心理层面
- [ ] 是否在确认自己的预设观点？
- [ ] 结论是否受到锚定效应影响？
- [ ] 是否有反面证据被忽略？
```

---

## 七、建立反偏差思维

### 核心原则

1. **假设数据在骗你**：先怀疑，再相信
2. **寻找反面证据**：主动找能推翻结论的数据
3. **多角度验证**：用不同方法、不同数据验证同一结论
4. **承认不确定性**：标注置信度，不过度解读
5. **区分探索与验证**：探索性发现需要新数据验证

### 提问习惯

每个重要结论都问：

```
- 这个结论的反面证据是什么？
- 什么情况下这个结论是错的？
- 如果我是持相反观点的人，我会怎么反驳？
- 这个数据还有其他解释吗？
- 我是否遗漏了什么重要因素？
```

---

## 八、经典案例库

### 案例1：广告效果的选择偏差

```
现象：看过广告的用户购买率是未看过的2倍
天真结论：广告效果很好

问题：谁会看到广告？
- 更活跃的用户更可能看到广告
- 更活跃的用户本来就更可能购买
- 广告可能被投放给了"更可能购买"的人群

正确做法：随机对照实验（A/B测试）
```

### 案例2：培训效果的反向因果

```
现象：参加培训的销售业绩更好
天真结论：培训提升了业绩

问题：谁会参加培训？
- 业绩好的员工更被重视，更可能被安排培训
- 上进的员工更愿意主动参加培训
- 培训和业绩可能都是"员工能力"这个因素的结果

正确做法：随机分配培训名额，对比培训组vs对照组
```

### 案例3：用户满意度的幸存者偏差

```
现象：用户满意度调查得分很高（4.5/5）
天真结论：我们的产品很让用户满意

问题：谁会填写满意度调查？
- 极端满意或极端不满的用户更可能反馈
- 中间状态的用户可能默默流失
- 已经流失的用户根本不会填写

正确做法：
- 跟踪流失用户并主动调研
- 分析满意度与实际行为（留存/付费）的关系
- 看填写率，填写率低可能意味着代表性差
```

# 数据解读原则

## 核心原则

### 1. 异常值检测
- 主动寻找数据中的异常点和离群值
- 异常值可能是错误，也可能是重要洞察
- 对异常值要追问原因，不要轻易忽略

### 2. 基准线思维
- 任何数据都要有参照物才有意义
- 建立对比基准：历史数据、行业均值、目标值
- 单独的数字没有说服力，对比才有洞察

### 3. 相关性 ≠ 因果性
- 两个指标同时变化，不代表一个导致另一个
- 可能存在第三变量（混淆变量）
- 因果关系需要更严谨的验证

### 4. 样本量审视
- 小样本结论谨慎推广
- 关注样本的代表性
- 明确数据覆盖的时间范围和对象范围

### 5. 数据时效性
- 检查数据的采集时间
- 考虑数据是否仍然反映当前情况
- 注意季节性、周期性因素

### 6. 二八法则应用
- 关注贡献最大的 20% 因素
- 在复杂数据中抓住主要矛盾
- 资源有限时优先分析关键部分

## 常见陷阱

### 避免的错误
- 只看绝对值，忽略相对变化
- 只看平均值，忽略分布差异
- 选择性展示有利数据
- 过度解读小幅波动
- 忽略数据定义的差异

### 保持的态度
- 对数据保持好奇和怀疑
- 多问「为什么」和「所以呢」
- 承认不确定性，标注置信度
- 区分事实陈述和主观判断

# 数据叙事框架 (Data Storytelling)

> 分析的价值不在于发现，而在于让发现被理解、被记住、被行动。

---

## 一、为什么需要数据叙事

### 分析师的困境

```
做了深入分析 → 写了详细报告 → 没人看/看不懂/看了不行动
```

### 数据叙事的价值

| 单纯呈现数据 | 数据叙事 |
|-------------|---------|
| 堆砌图表和数字 | 讲一个完整的故事 |
| 受众需要自己解读 | 引导受众理解结论 |
| 容易被遗忘 | 让人印象深刻 |
| 难以推动行动 | 自然引向下一步 |

### 数据叙事三要素

```
           数据 (Data)
              ↑
             /  \
            /    \
    可视化        叙事
 (Visualization) (Narrative)
```

- **数据**：准确、相关、有洞察的数据
- **可视化**：清晰、有效的图表呈现
- **叙事**：有逻辑、有情感的故事线

三者缺一不可。

---

## 二、核心叙事结构

### 1. SCR结构（麦肯锡经典）

```
S - Situation（情境）
    现状是什么？背景是什么？

C - Complication（冲突）
    问题是什么？挑战在哪里？为什么重要？

R - Resolution（解决）
    建议是什么？应该怎么做？
```

#### SCR示例

```
❌ 平铺直叙：
"Q3销售额1200万，同比下降15%，环比下降8%。
华东区下降20%，华南区下降5%，华北区增长2%。
建议加强华东区销售。"

✅ SCR叙事：
【S-情境】
我们一直是华东区市场的领导者，Q2市占率达到35%。

【C-冲突】
但Q3我们在华东区遭遇重大挫折：
- 销售额同比下降20%，是全国降幅最大的区域
- 3家核心经销商终止合作
- 竞品趁机抢占了12%的市场份额

如果不立即行动，Q4可能失去华东区领导地位。

【R-解决】
建议采取三步紧急措施：
1. 本周内与流失经销商重启谈判（重点解决账期问题）
2. 两周内签约2家替代经销商
3. 临时增加直营配送覆盖空白区域

预计可挽回60%的损失，保住市场份额。
```

### 2. SCQA结构（金字塔原理）

```
S - Situation（背景）
Q - Question（问题）
A - Answer（答案）
```

更适合汇报场景，开门见山给结论。

#### SCQA示例

```
【S-背景】
公司计划Q4推出新产品线，需要确定目标市场。

【Q-问题】
我们应该优先进入哪个细分市场？

【A-答案】
建议优先进入"年轻白领"市场。

原因：
1. 市场规模最大（3000亿）
2. 增速最快（年增25%）
3. 与我们现有用户画像最匹配
4. 竞争相对空白

[以下展开详细论据...]
```

### 3. 英雄之旅结构（适合演讲）

```
1. 现状世界 - 一切看起来正常
2. 发现问题 - 数据揭示了隐藏的危机/机会
3. 探索旅程 - 深入分析，发现真相
4. 关键发现 - 核心洞察
5. 行动号召 - 我们需要做什么
6. 美好未来 - 如果行动，会怎样
```

---

## 三、受众分析与适配

### 了解你的受众

在准备任何数据呈现之前，先回答：

| 问题 | 影响 |
|-----|------|
| 受众是谁？ | 决定内容深度和专业度 |
| 他们关心什么？ | 决定强调哪些发现 |
| 他们的知识背景？ | 决定需要多少解释 |
| 他们将如何使用？ | 决定呈现格式和详略 |
| 他们有多少时间？ | 决定内容长度 |

### 不同受众的适配策略

#### 高管层 (C-Level)

```
特点：时间有限，关注战略和结果
风格：结论先行，一页纸
内容：核心结论 + 行动建议 + 预期影响
避免：过多细节、技术术语、过程描述

模板：
┌─────────────────────────────┐
│  核心结论（一句话）           │
│  ─────────────────────────  │
│  关键数据支撑（3个以内）      │
│  ─────────────────────────  │
│  建议行动（优先级排序）       │
│  ─────────────────────────  │
│  预期影响（量化）             │
└─────────────────────────────┘
```

#### 业务负责人

```
特点：关注可操作性，需要足够信息做决策
风格：结论+论据，5-10页
内容：核心发现 + 数据细节 + 多个选项 + 建议
可以：展开论证过程、提供备选方案

模板：
1. 执行摘要（1页）
2. 核心发现（2-3页）
3. 详细分析（3-5页）
4. 建议与下一步（1页）
```

#### 执行团队/运营

```
特点：需要具体指导，关注执行细节
风格：详细文档，可操作清单
内容：具体数据 + 操作建议 + 监控指标
可以：包含技术细节、操作流程

模板：
1. 背景与目标
2. 当前状态分析
3. 具体发现与数据
4. 分渠道/分产品详情
5. 具体行动计划
6. 跟踪指标与时间表
```

#### 技术/数据团队

```
特点：关心方法论和数据质量
风格：详细报告，包含技术附录
内容：完整分析过程 + 数据口径 + 方法说明
可以：包含统计检验、代码、数据源说明

模板：
1. 分析目标与范围
2. 数据源与口径定义
3. 分析方法
4. 核心发现
5. 技术附录（SQL/代码/统计检验）
```

---

## 四、行动导向的结论

### 结论质量检验

每个结论都要通过这个检验：

```
□ 具体明确 - 不含糊、不空泛
□ 有数据支撑 - 不是拍脑袋
□ 有对比基准 - 高/低相对于什么
□ 指向行动 - 知道下一步该做什么
□ 可追踪 - 能验证建议是否有效
```

### 结论表述对比

| 弱结论 | 问题 | 强结论 |
|-------|------|-------|
| "销售有所下降" | 模糊 | "销售下降15%，主要因华东经销商流失" |
| "用户体验有待改善" | 空泛 | "结账流程流失率38%，是行业均值2倍" |
| "建议加强营销" | 不可操作 | "建议追加50万投放抖音，预计ROI 2.5" |
| "数据表明相关" | 无意义 | "价格每降1%，销量增2.3%，建议试点降价" |

### 行动建议模板

```markdown
## 建议：[具体行动]

**背景**：[为什么需要这个行动]

**具体措施**：
1. [第一步] - 负责人：xxx - 时间：xxx
2. [第二步] - 负责人：xxx - 时间：xxx

**预期效果**：[量化的预期收益]

**风险与前提**：[执行的假设条件]

**如何衡量成功**：[跟踪指标]
```

---

## 五、数据可视化叙事

### 图表选择的叙事考量

不只是"什么数据用什么图"，还要考虑"想传达什么信息"：

| 想传达的信息 | 图表选择 | 叙事强调 |
|-------------|---------|---------|
| "差距很大" | 柱状图（差异明显） | 用颜色突出差距 |
| "趋势在变" | 折线图 + 趋势线 | 标注拐点 |
| "集中在少数" | 帕累托图 | 标注80/20分界 |
| "分布不均" | 箱线图/分布图 | 标注极值 |
| "存在关系" | 散点图 + 趋势线 | 标注相关系数 |

### 图表叙事技巧

#### 1. 标题即结论

```
❌ "2023年各区域销售额"
✅ "华东区销售领跑，贡献45%收入"
```

#### 2. 用颜色讲故事

```
- 高亮关键数据点
- 用红/绿表示正/负变化
- 灰化次要信息
- 同一指标保持颜色一致
```

#### 3. 添加注释和标注

```
- 在关键拐点添加文字说明
- 标注对比基准线
- 注明数据来源和时间
```

#### 4. 控制信息密度

```
- 一张图传达一个核心信息
- 删除非必要元素
- 简化数据标签
```

---

## 六、常见叙事模式

### 模式1：问题诊断型

```
结构：发现问题 → 拆解原因 → 量化影响 → 提出方案

示例流程：
1. "Q3利润率下降5个百分点"
2. "拆解发现：原材料成本上涨贡献60%，人力成本20%，其他20%"
3. "如不干预，全年利润将减少2000万"
4. "建议：① 重新谈判供应商 ② 优化低效岗位 ③ 提价转移部分成本"
```

### 模式2：机会发现型

```
结构：发现机会 → 量化规模 → 验证可行 → 行动计划

示例流程：
1. "数据发现：下沉市场用户增速是一线的3倍"
2. "市场规模测算：潜在市场500亿，当前渗透率仅5%"
3. "验证：试点城市ROI达到2.8，高于公司平均"
4. "建议：扩大下沉市场投入，目标Q4进入10个新城市"
```

### 模式3：对比验证型

```
结构：假设提出 → A/B对比 → 结果验证 → 推广建议

示例流程：
1. "假设：新版详情页能提升转化率"
2. "A/B测试：新版 vs 旧版，各10万流量"
3. "结果：新版转化率+23%（p<0.01），客诉率无差异"
4. "建议：全量上线新版，预计年增收800万"
```

### 模式4：趋势预警型

```
结构：趋势呈现 → 问题预警 → 原因分析 → 预防建议

示例流程：
1. "趋势：用户留存率连续3个月下降"
2. "预警：若趋势持续，6个月后月活将跌破100万"
3. "分析：流失主要来自30天以上老用户，与最近改版时间吻合"
4. "建议：① 回滚部分功能 ② 针对老用户的召回活动"
```

---

## 七、情感与逻辑的平衡

### 数据需要情感连接

纯逻辑的数据难以推动行动，需要适度的情感元素：

```
纯逻辑："客户满意度从85%下降到72%"

加入情感："每10个客户中，就有3个对我们不满意。
         他们可能正在考虑竞品。
         我们正在失去信任。"
```

### 但不要过度煽情

```
❌ "这是灾难性的下降！我们必须立即行动否则公司将面临生存危机！"

✅ "下降趋势值得关注。如果不干预，预计Q4影响约200万收入。
   建议优先处理TOP3问题点。"
```

### 平衡原则

- 用数据建立可信度
- 用故事建立连接
- 用行动建立紧迫感
- 保持专业克制

---

## 八、数据叙事检查清单

### 叙事结构检查

```
□ 有清晰的故事线（背景→冲突→解决）
□ 结论先行，不让受众猜
□ 逻辑层次清晰（不超过3层）
□ 每个论点有数据支撑
```

### 受众适配检查

```
□ 明确受众是谁
□ 内容深度匹配受众需求
□ 术语受众能理解
□ 长度受众能接受
```

### 可视化检查

```
□ 图表类型选择合适
□ 标题传达核心信息
□ 关键数据被突出
□ 没有多余元素
```

### 行动导向检查

```
□ 有明确的行动建议
□ 建议具体可执行
□ 预期效果被量化
□ 有跟踪衡量方式
```

---

## 九、实战模板

### 5分钟汇报模板

```markdown
## [一句话核心结论]

### 背景（30秒）
[1-2句话说明分析背景]

### 核心发现（2分钟）
1. [发现1] - [数据支撑]
2. [发现2] - [数据支撑]
3. [发现3] - [数据支撑]

### 建议行动（1.5分钟）
- 首要：[最重要的行动] → 预期收益
- 其次：[次要行动] → 预期收益

### 下一步（30秒）
[具体的下一步是什么，需要什么支持]
```

### 数据周报模板

```markdown
## [业务名称] 周报 - W[xx]

### 本周核心指标
| 指标 | 本周 | 上周 | 环比 | 达成率 |
|-----|-----|-----|------|-------|

### 关键发现（Top 3）
1. **[发现标题]**：[一句话描述 + 数据]
2. **[发现标题]**：[一句话描述 + 数据]
3. **[发现标题]**：[一句话描述 + 数据]

### 异常与风险
- [异常1]：原因是xxx，影响是xxx
- [风险1]：如果xxx，可能导致xxx

### 下周关注
- [ ] [关注点1]
- [ ] [关注点2]
```

# 驱动因素树分析框架 (Driver Tree Analysis)

> 将复杂的业务指标拆解为可操作的驱动因素，找到真正的增长杠杆。

---

## 一、什么是驱动因素树

### 核心理念

驱动因素树（Driver Tree）是一种将顶层KPI **逐层分解** 到可操作因素的分析方法。

```
目标：不是只看"收入下降了"
而是：找到"收入下降是因为哪个环节出了问题，应该从哪里改进"
```

### 为什么要用驱动因素树？

| 只看顶层指标 | 使用驱动因素树 |
|-------------|---------------|
| 收入下降10% | 收入下降10% = 客单价↑5% × 订单量↓15% |
| 不知道从哪改进 | 订单量下降来自新客减少，需要加强获客 |
| 只能给出模糊建议 | 可以量化每个因素的贡献度 |
| 容易遗漏关键因素 | 系统性覆盖所有驱动因素 |

---

## 二、经典驱动因素树模型

### 1. 收入驱动树 (Revenue Driver Tree)

```
收入 (Revenue)
│
├─ 用户量 (Users)
│   ├─ 新用户
│   │   ├─ 自然流量
│   │   ├─ 付费获客
│   │   └─ 转介绍
│   └─ 老用户留存
│       ├─ 留存率
│       └─ 召回用户
│
├─ 转化率 (Conversion)
│   ├─ 浏览→加购率
│   ├─ 加购→下单率
│   └─ 下单→支付率
│
└─ 客单价 (ARPU)
    ├─ 件单价
    ├─ 购买件数
    └─ 购买频次
```

**基础公式**：
```
收入 = 用户数 × 转化率 × 客单价
     = 用户数 × 转化率 × 件单价 × 件数 × 频次
```

### 2. 利润驱动树 (Profit Driver Tree)

```
利润 (Profit)
│
├─ 收入 (Revenue)
│   └─ [参考收入驱动树]
│
└─ 成本 (Cost)
    ├─ 可变成本
    │   ├─ 商品成本 (COGS)
    │   ├─ 物流成本
    │   └─ 支付手续费
    │
    └─ 固定成本
        ├─ 人力成本
        ├─ 营销费用
        └─ 运营费用
```

**关键指标**：
```
毛利率 = (收入 - 可变成本) / 收入
净利率 = (收入 - 总成本) / 收入
```

### 3. 用户增长驱动树 (Growth Driver Tree)

```
用户净增长
│
├─ 新增用户 (Acquisition)
│   ├─ 渠道A新增
│   │   ├─ 曝光量
│   │   ├─ 点击率
│   │   └─ 转化率
│   ├─ 渠道B新增
│   └─ 自然增长
│
├─ 用户留存 (Retention)
│   ├─ 次日留存
│   ├─ 7日留存
│   └─ 30日留存
│
└─ 用户流失 (Churn)
    ├─ 主动流失
    └─ 自然流失
```

**增长核算公式**：
```
期末用户 = 期初用户 + 新增 - 流失
净增长 = 新增用户 - 流失用户
```

### 4. 电商GMV驱动树

```
GMV
│
├─ 流量 (Traffic)
│   ├─ 搜索流量
│   ├─ 推荐流量
│   ├─ 活动流量
│   └─ 付费流量
│
├─ 转化率 (CVR)
│   ├─ 详情页转化
│   ├─ 加购转化
│   └─ 支付转化
│
├─ 客单价 (ATV)
│   ├─ 商品均价
│   └─ 连带率
│
└─ 复购 (Repurchase)
    ├─ 复购率
    └─ 复购频次
```

### 5. SaaS业务驱动树

```
MRR (月经常性收入)
│
├─ 新增MRR
│   ├─ 新客户
│   │   ├─ 线索量
│   │   ├─ 转化率
│   │   └─ 客单价
│   └─ 老客户升级
│
├─ 流失MRR
│   ├─ 客户流失 (Churn)
│   └─ 降级收入
│
└─ 扩展MRR
    ├─ 追加购买
    └─ 交叉销售
```

**SaaS关键指标**：
```
净收入留存率 (NRR) = (期初MRR + 扩展 - 流失 - 降级) / 期初MRR
LTV = ARPU × 平均生命周期
LTV/CAC > 3 为健康
```

---

## 三、如何构建驱动因素树

### Step 1: 确定顶层目标

明确你要分析的核心KPI：
- 收入/利润/GMV
- 用户数/活跃度
- 转化率/留存率

### Step 2: 识别第一层驱动因素

问自己：这个指标由哪些 **相乘或相加** 的因素构成？

```
乘法关系：收入 = 用户 × 转化 × 客单价
加法关系：用户 = 新用户 + 老用户
```

### Step 3: 继续拆解每个因素

对每个一级因素重复Step 2，直到：
- 因素已经是可直接操作/可直接获取数据的
- 继续拆解没有业务意义

### Step 4: 验证完整性 (MECE检验)

- **相互独立**：各因素之间不重叠
- **完全穷尽**：所有因素加起来等于上级

### Step 5: 标注数据可得性

| 因素 | 数据来源 | 可得性 |
|-----|---------|-------|
| 新用户数 | 用户表 | ✅ 可得 |
| 竞品价格 | 市场调研 | ⚠️ 需采集 |
| 用户满意度 | 调研问卷 | ❌ 暂无 |

---

## 四、驱动因素归因分析

### 量化每个因素的贡献

当顶层指标变化时，要量化每个因素的贡献度。

#### 乘法分解法

对于 `Y = A × B × C` 的结构：

```
收入变化 = (用户变化贡献) + (转化率变化贡献) + (客单价变化贡献)

简化计算：
用户贡献 ≈ Δ用户/用户基数 × 原收入
转化贡献 ≈ Δ转化/转化基数 × 原收入
客单价贡献 ≈ Δ客单价/客单价基数 × 原收入
```

#### 实例演示

```
Q2 vs Q1 收入变化分析：

Q1: 收入 = 10000用户 × 5%转化 × 200元客单价 = 100万
Q2: 收入 = 9000用户 × 6%转化 × 220元客单价 = 118.8万

收入增长 = +18.8万 (+18.8%)

因素拆解：
- 用户下降贡献：(9000-10000)/10000 × 100万 = -10万
- 转化提升贡献：(6%-5%)/5% × 100万 = +20万
- 客单价提升贡献：(220-200)/200 × 100万 = +10万
- 交叉项：约 -1.2万

结论：虽然用户流失带来10万损失，但转化率和客单价的提升
     弥补了用户损失，整体收入仍增长18.8万。
```

### 敏感性分析

识别哪个因素对结果影响最大（杠杆最高）：

```
当前状态：收入 = 10000用户 × 5%转化 × 200元

各因素提升10%的影响：
- 用户+10%：收入 → 110万 (+10%)
- 转化+10%：收入 → 110万 (+10%)
- 客单价+10%：收入 → 110万 (+10%)

但实际可提升空间不同：
- 用户+10%：需要增加1000获客，成本约5万
- 转化+10%：优化页面，成本约0.5万
- 客单价+10%：可能导致转化下降

结论：转化率优化是ROI最高的杠杆
```

---

## 五、常见业务场景驱动树

### 场景1：销售下滑诊断

```
销售额下滑
│
├─ 量的问题？
│   ├─ 客户数下降？
│   │   ├─ 新客减少
│   │   └─ 老客流失
│   └─ 购买频次下降？
│
├─ 价的问题？
│   ├─ 主动降价？
│   ├─ 产品结构变化？（低价产品占比上升）
│   └─ 折扣力度加大？
│
└─ 渠道问题？
    ├─ 线上渠道
    └─ 线下渠道
```

### 场景2：利润率下降诊断

```
利润率下降
│
├─ 收入端
│   ├─ 价格下降？
│   └─ 产品结构恶化？
│
└─ 成本端
    ├─ 原材料涨价？
    ├─ 人力成本上升？
    ├─ 营销费用增加？
    └─ 运营效率下降？
```

### 场景3：用户活跃度下降诊断

```
DAU下降
│
├─ 新用户问题
│   ├─ 获客量下降
│   └─ 新用户激活率下降
│
├─ 老用户问题
│   ├─ 留存率下降
│   └─ 回访频次下降
│
└─ 流失加速
    ├─ 产品问题
    ├─ 竞品分流
    └─ 需求变化
```

---

## 六、驱动因素树最佳实践

### DO - 应该做

1. **从业务逻辑出发**，而不是从数据出发
2. **保持MECE**，确保不重不漏
3. **标注数据可得性**，识别数据缺口
4. **量化因素贡献**，不只是定性描述
5. **识别关键杠杆**，聚焦高ROI因素
6. **定期更新**，业务变化时重构驱动树

### DON'T - 避免

1. **过度拆解**：拆到没有业务意义的层级
2. **遗漏关键因素**：只看容易量化的，忽略难量化但重要的
3. **混淆层级**：把不同层级的因素放在同一层
4. **忽略交叉影响**：因素之间可能相互影响
5. **只建不用**：建完驱动树但分析时不用它

---

## 七、输出模板

### 驱动因素分析报告模板

```markdown
## [指标名称] 驱动因素分析

### 1. 驱动因素树
[绘制或描述完整的驱动因素树]

### 2. 各因素表现
| 因素 | 当期值 | 对比基准 | 变化 | 贡献度 |
|-----|-------|---------|------|-------|
| 因素A | xxx | xxx | +x% | 40% |
| 因素B | xxx | xxx | -x% | 35% |
| 因素C | xxx | xxx | +x% | 25% |

### 3. 关键发现
- 主要驱动因素：xxx（贡献xx%的变化）
- 主要拖累因素：xxx（贡献xx%的下降）

### 4. 杠杆分析
| 因素 | 敏感度 | 可提升空间 | 提升成本 | 优先级 |
|-----|-------|-----------|---------|-------|
| 因素A | 高 | 中 | 低 | ⭐⭐⭐ |
| 因素B | 中 | 高 | 高 | ⭐⭐ |

### 5. 行动建议
1. [最高优先级行动] - 预期影响：xxx
2. [次优先级行动] - 预期影响：xxx
```

---

## 八、与其他框架的结合

| 结合框架 | 使用场景 |
|---------|---------|
| + 假设驱动分析 | 用驱动树构建假设，逐个验证 |
| + So What追问 | 每个因素的变化意味着什么 |
| + MECE原则 | 确保驱动因素不重不漏 |
| + 对比分析 | 因素的时间/空间/标杆对比 |

# 通用分析框架

## MECE 原则（Mutually Exclusive, Collectively Exhaustive）
- **相互独立**：各个分类之间不重叠
- **完全穷尽**：所有分类加起来覆盖全部情况
- **应用**：在拆解问题或数据分类时使用

## 金字塔原理
- **结论先行**：先给出核心结论，再展开论据
- **以上统下**：上层结论由下层论据支撑
- **归类分组**：同级论据按逻辑归类
- **逻辑递进**：各组之间有清晰的逻辑顺序

## 5W1H 分析法
- **What**：发生了什么？数据呈现什么现象？
- **Why**：为什么会这样？背后原因是什么？
- **Who**：涉及哪些对象？谁是关键角色？
- **When**：什么时间发生？时间维度的变化？
- **Where**：在哪里发生？地域/渠道差异？
- **How**：如何发生的？程度如何？

## 对比分析框架
- **时间对比**：同比（YoY）、环比（MoM/QoQ）
- **空间对比**：地区、渠道、部门
- **标杆对比**：行业均值、竞品数据、历史最佳
- **目标对比**：实际 vs 计划、完成率

# 假设驱动分析法 & So What 追问链

> 大师级数据分析师的核心思维方式：不是"看看数据说什么"，而是"用数据验证假设、驱动决策"。

---

## 一、假设驱动分析法 (Hypothesis-Driven Analysis)

### 核心理念

```
普通分析师：拿到数据 → 做各种分析 → 看看能发现什么
大师分析师：明确问题 → 提出假设 → 用数据验证 → 得出结论
```

### 为什么假设先行？

| 漫无目的分析 | 假设驱动分析 |
|-------------|-------------|
| 耗时长，容易迷失 | 聚焦，高效 |
| 输出一堆"有趣但无用"的发现 | 输出可行动的结论 |
| 分析完不知道该干什么 | 直接指向决策 |
| 容易被数据"牵着走" | 主动掌控分析方向 |

### 执行步骤

#### Step 1: 明确核心问题 (Define the Question)

**追问清单**：
- 这个分析要回答什么业务问题？
- 分析结果会影响什么决策？
- 谁是这个分析的受众？他们关心什么？
- 如果分析结论是 A，会怎么做？如果是 B，又会怎么做？

**问题重构技巧**：
```
用户说的          →  真正需要回答的
"分析一下销售数据"  →  "找到销售增长的机会点"
"看看用户行为"     →  "识别高价值用户的特征"
"出个月报"        →  "诊断本月业绩问题并给出建议"
```

#### Step 2: 构建假设树 (Build Hypothesis Tree)

将核心问题拆解为可验证的子假设：

```
核心问题：为什么 Q3 销售下滑？
│
├─ 假设1：需求端问题
│   ├─ 1.1 市场整体萎缩
│   ├─ 1.2 竞品抢占份额
│   └─ 1.3 用户需求转移
│
├─ 假设2：供给端问题
│   ├─ 2.1 产品竞争力下降
│   ├─ 2.2 定价策略问题
│   └─ 2.3 供应链/库存问题
│
└─ 假设3：渠道端问题
    ├─ 3.1 线上渠道流量下降
    ├─ 3.2 线下经销商流失
    └─ 3.3 营销投放效率降低
```

#### Step 3: 确定验证方法 (Design Validation)

每个假设都要明确：
- 需要什么数据来验证？
- 数据从哪里获取？
- 什么样的结果能证实/证伪假设？

| 假设 | 所需数据 | 证实标准 | 证伪标准 |
|-----|---------|---------|---------|
| 市场整体萎缩 | 行业报告、竞品数据 | 行业增速<0 | 行业增速>0 |
| 竞品抢占份额 | 竞品销量/市占率 | 竞品份额↑且我方↓ | 竞品份额无明显变化 |
| 定价策略问题 | 价格变动、价格弹性 | 提价后销量显著下降 | 价格与销量无相关性 |

#### Step 4: 执行分析 (Execute Analysis)

- 按优先级验证假设（先验证最可能的、影响最大的）
- 记录每个假设的验证结果
- 保持开放心态，允许假设被推翻

#### Step 5: 综合结论 (Synthesize Findings)

```
经过验证，Q3销售下滑的主要原因是：
✓ 假设3.2 成立：华东区3家核心经销商流失，贡献了60%的下滑
✗ 假设1.1 不成立：行业整体增长5%，我方逆势下滑
△ 假设2.2 部分成立：高端线提价影响了10%的销量
```

---

## 二、So What 追问链

### 核心理念

> 数据本身没有价值，数据背后的洞察和行动才有价值。

每个数据发现都要经过三层追问：

```
数据现象 → So What? → 业务含义 → So What? → 行动建议 → So What? → 预期影响
```

### So What 追问示例

#### 示例1：销售分析

```
Level 0 - 数据现象：
"Q3销售额同比下降15%"

↓ So What?

Level 1 - 业务含义：
"下降主要来自华东区（贡献80%降幅），其中经销商渠道下滑最严重"

↓ So What?

Level 2 - 根因分析：
"3家核心经销商因账期问题终止合作，导致该区域覆盖率从85%降至60%"

↓ So What?

Level 3 - 行动建议：
"建议：1)与流失经销商重新谈判账期 2)紧急开发2家替代经销商 3)临时增加直营配送"

↓ So What?

Level 4 - 预期影响：
"预计执行后Q4可恢复华东区70%销量，减少损失约500万"
```

#### 示例2：用户分析

```
Level 0："月活用户增长20%"
↓ So What?
Level 1："增长主要来自下沉市场，一线城市用户实际在流失"
↓ So What?
Level 2："下沉用户ARPU仅为一线用户的1/5，总收入实际下降"
↓ So What?
Level 3："需要差异化运营：下沉市场做规模，一线城市做价值"
```

### So What 检验清单

每个结论都用这个清单检验：

- [ ] **可理解**：非专业人士能听懂吗？
- [ ] **有对比**：比什么高/低/快/慢？
- [ ] **有归因**：为什么会这样？
- [ ] **可行动**：基于这个结论应该做什么？
- [ ] **可量化**：行动的预期收益是什么？

### 常见的"伪结论"

| 伪结论 | 问题 | 改进 |
|-------|------|------|
| "销售下降了15%" | 只有现象，没有分析 | 下降来自哪里？为什么？ |
| "用户满意度有待提升" | 空泛，无法行动 | 哪些维度不满意？改进优先级？ |
| "建议加强营销投入" | 方向不明确 | 投入哪个渠道？预期ROI？ |
| "数据表明A和B相关" | 没有业务含义 | 这个相关性意味着什么？如何利用？ |

---

## 三、实战模板

### 分析启动检查清单

在开始任何分析之前，确认：

```markdown
## 分析启动检查

### 1. 问题定义
- [ ] 核心业务问题是什么？
- [ ] 这个分析支持什么决策？
- [ ] 受众是谁？他们的关注点？

### 2. 假设构建
- [ ] 列出3-5个初始假设
- [ ] 假设之间是否MECE？
- [ ] 每个假设的验证方法？

### 3. 数据准备
- [ ] 需要哪些数据？
- [ ] 数据是否可获取？
- [ ] 数据质量如何？

### 4. 成功标准
- [ ] 什么样的结论是"好结论"？
- [ ] 如何衡量分析的价值？
```

### 结论输出模板

```markdown
## [结论标题 - 一句话概括]

### 核心发现
[用2-3句话说清楚发现了什么]

### 数据支撑
- 关键数据点1：xxx
- 关键数据点2：xxx
- 对比基准：xxx

### 根因分析
[为什么会出现这个现象]

### 行动建议
1. [具体行动1] - 预期效果：xxx
2. [具体行动2] - 预期效果：xxx

### 风险与假设
- 本结论基于以下假设：xxx
- 主要不确定性：xxx
```

---

## 四、思维转变要点

### 从"分析师"到"问题解决者"

| 分析师思维 | 问题解决者思维 |
|-----------|---------------|
| 我有什么数据 | 我需要回答什么问题 |
| 数据呈现什么 | 数据意味着什么 |
| 分析做完了 | 决策者能行动了吗 |
| 这个发现很有趣 | 这个发现能带来什么价值 |

### 记住

> "The goal is not to analyze data, but to drive decisions."
>
> 分析的目标不是分析数据本身，而是驱动决策。

# 可视化选择规则

## 图表类型选择矩阵

| 分析目的 | 推荐图表 | 次选 | 避免 |
|---------|---------|------|------|
| 展示趋势变化 | 折线图 | 面积图 | 饼图 |
| 比较大小差异 | 柱状图 | 条形图 | 折线图 |
| 展示占比构成 | 饼图/环形图 | 堆叠柱状图 | 折线图 |
| 展示分布情况 | 直方图 | 箱线图 | 饼图 |
| 探索变量关系 | 散点图 | 气泡图 | 柱状图 |
| 多维度对比 | 雷达图 | 分组柱状图 | 饼图 |
| 地理分布 | 地图 | 气泡图 | 表格 |

## 设计原则

### 数据墨水比
- 最大化数据信息，最小化非数据元素
- 删除不必要的网格线、边框、装饰
- 每个视觉元素都应该传递信息

### 色彩使用
- 主色调不超过 5 种
- 用色彩强调重点，而非装饰
- 考虑色盲友好（避免红绿对比）
- 深色表示大值，浅色表示小值

### 标注规范
- 图表必须有清晰的标题
- 坐标轴必须有标签和单位
- 数据来源要标注
- 关键数据点可以直接标注数值

## 常见错误

- 饼图超过 5 个分类
- 3D 效果扭曲数据比例
- Y 轴不从 0 开始（柱状图）
- 双 Y 轴误导比较
- 过多颜色造成视觉混乱

#!/usr/bin/env python3
"""
图表生成器 - 根据数据和结论生成可视化图表
支持：折线图、柱状图、饼图、散点图、面积图
"""

import json
from pathlib import Path
from datetime import datetime

import matplotlib.pyplot as plt
import matplotlib.font_manager as fm
import plotly.express as px
import plotly.graph_objects as go
import pandas as pd

# 设置中文字体
plt.rcParams['font.sans-serif'] = ['Arial Unicode MS', 'SimHei', 'DejaVu Sans']
plt.rcParams['axes.unicode_minus'] = False

# 图表配色方案
CHART_COLORS = [
    '#2563EB',  # 主蓝色
    '#10B981',  # 绿色
    '#F59E0B',  # 橙色
    '#8B5CF6',  # 紫色
    '#EC4899',  # 粉色
    '#06B6D4',  # 青色
    '#EF4444',  # 红色
]

class ChartGenerator:
    """图表生成器"""

    def __init__(self, output_dir: str = 'outputs/charts'):
        self.output_dir = Path(output_dir)
        self.output_dir.mkdir(parents=True, exist_ok=True)

    def generate_line_chart(self, data: dict, title: str, filename: str) -> str:
        """生成折线图"""
        fig = go.Figure()

        for i, (name, values) in enumerate(data.get('series', {}).items()):
            fig.add_trace(go.Scatter(
                x=data.get('x_labels', list(range(len(values)))),
                y=values,
                mode='lines+markers',
                name=name,
                line=dict(color=CHART_COLORS[i % len(CHART_COLORS)], width=2),
                marker=dict(size=6)
            ))

        fig.update_layout(
            title=dict(text=title, x=0.5, font=dict(size=16)),
            xaxis_title=data.get('x_title', ''),
            yaxis_title=data.get('y_title', ''),
            template='plotly_white',
            legend=dict(orientation='h', yanchor='bottom', y=1.02, xanchor='right', x=1),
            margin=dict(l=60, r=40, t=80, b=60)
        )

        output_path = self.output_dir / f"{filename}.html"
        fig.write_html(str(output_path))

        # 同时保存为图片
        img_path = self.output_dir / f"{filename}.png"
        fig.write_image(str(img_path), width=800, height=500)

        return str(output_path)

    def generate_bar_chart(self, data: dict, title: str, filename: str) -> str:
        """生成柱状图"""
        fig = go.Figure()

        x_labels = data.get('x_labels', [])

        for i, (name, values) in enumerate(data.get('series', {}).items()):
            fig.add_trace(go.Bar(
                x=x_labels,
                y=values,
                name=name,
                marker_color=CHART_COLORS[i % len(CHART_COLORS)]
            ))

        fig.update_layout(
            title=dict(text=title, x=0.5, font=dict(size=16)),
            xaxis_title=data.get('x_title', ''),
            yaxis_title=data.get('y_title', ''),
            template='plotly_white',
            barmode='group',
            legend=dict(orientation='h', yanchor='bottom', y=1.02, xanchor='right', x=1),
            margin=dict(l=60, r=40, t=80, b=60)
        )

        output_path = self.output_dir / f"{filename}.html"
        fig.write_html(str(output_path))

        img_path = self.output_dir / f"{filename}.png"
        fig.write_image(str(img_path), width=800, height=500)

        return str(output_path)

    def generate_pie_chart(self, data: dict, title: str, filename: str) -> str:
        """生成饼图"""
        fig = go.Figure(data=[go.Pie(
            labels=data.get('labels', []),
            values=data.get('values', []),
            hole=0.4,  # 环形图
            marker=dict(colors=CHART_COLORS[:len(data.get('labels', []))]),
            textinfo='label+percent',
            textposition='outside'
        )])

        fig.update_layout(
            title=dict(text=title, x=0.5, font=dict(size=16)),
            template='plotly_white',
            showlegend=True,
            legend=dict(orientation='h', yanchor='bottom', y=-0.1, xanchor='center', x=0.5),
            margin=dict(l=40, r=40, t=80, b=80)
        )

        output_path = self.output_dir / f"{filename}.html"
        fig.write_html(str(output_path))

        img_path = self.output_dir / f"{filename}.png"
        fig.write_image(str(img_path), width=600, height=500)

        return str(output_path)

    def generate_scatter_chart(self, data: dict, title: str, filename: str) -> str:
        """生成散点图"""
        fig = go.Figure()

        fig.add_trace(go.Scatter(
            x=data.get('x', []),
            y=data.get('y', []),
            mode='markers',
            marker=dict(
                size=10,
                color=CHART_COLORS[0],
                opacity=0.7
            ),
            text=data.get('labels', None),
            hovertemplate='%{text}<br>X: %{x}<br>Y: %{y}<extra></extra>'
        ))

        fig.update_layout(
            title=dict(text=title, x=0.5, font=dict(size=16)),
            xaxis_title=data.get('x_title', 'X'),
            yaxis_title=data.get('y_title', 'Y'),
            template='plotly_white',
            margin=dict(l=60, r=40, t=80, b=60)
        )

        output_path = self.output_dir / f"{filename}.html"
        fig.write_html(str(output_path))

        img_path = self.output_dir / f"{filename}.png"
        fig.write_image(str(img_path), width=800, height=500)

        return str(output_path)

    def generate_area_chart(self, data: dict, title: str, filename: str) -> str:
        """生成面积图"""
        fig = go.Figure()

        for i, (name, values) in enumerate(data.get('series', {}).items()):
            fig.add_trace(go.Scatter(
                x=data.get('x_labels', list(range(len(values)))),
                y=values,
                mode='lines',
                name=name,
                fill='tonexty' if i > 0 else 'tozeroy',
                line=dict(color=CHART_COLORS[i % len(CHART_COLORS)], width=1),
            ))

        fig.update_layout(
            title=dict(text=title, x=0.5, font=dict(size=16)),
            xaxis_title=data.get('x_title', ''),
            yaxis_title=data.get('y_title', ''),
            template='plotly_white',
            legend=dict(orientation='h', yanchor='bottom', y=1.02, xanchor='right', x=1),
            margin=dict(l=60, r=40, t=80, b=60)
        )

        output_path = self.output_dir / f"{filename}.html"
        fig.write_html(str(output_path))

        img_path = self.output_dir / f"{filename}.png"
        fig.write_image(str(img_path), width=800, height=500)

        return str(output_path)

    def generate_chart(self, chart_type: str, data: dict, title: str, filename: str) -> str:
        """根据类型生成图表"""
        generators = {
            'line': self.generate_line_chart,
            'bar': self.generate_bar_chart,
            'pie': self.generate_pie_chart,
            'scatter': self.generate_scatter_chart,
            'area': self.generate_area_chart
        }

        generator = generators.get(chart_type)
        if generator:
            return generator(data, title, filename)
        else:
            raise ValueError(f"不支持的图表类型: {chart_type}")

    def generate_from_conclusions(self, conclusions: list) -> list:
        """根据结论列表生成所有图表"""
        chart_paths = []

        for conclusion in conclusions:
            if conclusion.get('chart_type') and conclusion.get('chart_type') != 'none':
                chart_data = conclusion.get('chart_data', {})
                if chart_data:
                    path = self.generate_chart(
                        chart_type=conclusion['chart_type'],
                        data=chart_data,
                        title=conclusion.get('title', ''),
                        filename=f"chart_{conclusion.get('id', 'unknown')}"
                    )
                    chart_paths.append({
                        'id': conclusion.get('id'),
                        'path': path,
                        'type': conclusion['chart_type']
                    })

        return chart_paths

def main():
    """示例：生成测试图表"""
    generator = ChartGenerator()

    # 测试折线图
    line_data = {
        'x_labels': ['1月', '2月', '3月', '4月', '5月', '6月'],
        'series': {
            '销售额': [120, 150, 180, 220, 200, 250],
            '目标': [150, 150, 180, 200, 200, 220]
        },
        'x_title': '月份',
        'y_title': '金额（万元）'
    }
    generator.generate_line_chart(line_data, '月度销售趋势', 'test_line')

    # 测试柱状图
    bar_data = {
        'x_labels': ['产品A', '产品B', '产品C', '产品D'],
        'series': {
            '2023年': [100, 80, 120, 90],
            '2024年': [120, 95, 110, 100]
        },
        'x_title': '产品',
        'y_title': '销量'
    }
    generator.generate_bar_chart(bar_data, '产品销量对比', 'test_bar')

    # 测试饼图
    pie_data = {
        'labels': ['华东', '华南', '华北', '西部', '其他'],
        'values': [35, 25, 20, 15, 5]
    }
    generator.generate_pie_chart(pie_data, '区域销售占比', 'test_pie')

    print("测试图表生成完成！")

if __name__ == '__main__':
    main()

#!/usr/bin/env python3
"""
文件处理器 - 支持多格式数据文件的读取和预处理
支持格式：Excel (.xlsx, .xls), PDF (.pdf), Word (.docx), CSV (.csv), TXT (.txt)
"""

import os
import json
from pathlib import Path
from datetime import datetime

import pandas as pd
import openpyxl
import PyPDF2
from docx import Document

class FileProcessor:
    """多格式文件处理器"""

    SUPPORTED_EXTENSIONS = {
        '.xlsx': 'Excel',
        '.xls': 'Excel',
        '.pdf': 'PDF',
        '.docx': 'Word',
        '.csv': 'CSV',
        '.txt': 'TXT'
    }

    def __init__(self, upload_dir: str = 'uploads'):
        self.upload_dir = Path(upload_dir)

    def scan_files(self) -> list[dict]:
        """扫描上传目录中的文件"""
        files = []
        if not self.upload_dir.exists():
            return files

        for file_path in self.upload_dir.iterdir():
            if file_path.is_file():
                ext = file_path.suffix.lower()
                if ext in self.SUPPORTED_EXTENSIONS:
                    files.append({
                        'name': file_path.name,
                        'path': str(file_path),
                        'type': self.SUPPORTED_EXTENSIONS[ext],
                        'size': self._format_size(file_path.stat().st_size),
                        'extension': ext
                    })
        return files

    def _format_size(self, size: int) -> str:
        """格式化文件大小"""
        for unit in ['B', 'KB', 'MB', 'GB']:
            if size < 1024:
                return f"{size:.1f} {unit}"
            size /= 1024
        return f"{size:.1f} TB"

    def read_excel(self, file_path: str) -> dict:
        """读取 Excel 文件"""
        result = {
            'type': 'Excel',
            'sheets': {},
            'summary': ''
        }

        try:
            xl = pd.ExcelFile(file_path)
            for sheet_name in xl.sheet_names:
                df = pd.read_excel(xl, sheet_name=sheet_name)
                result['sheets'][sheet_name] = {
                    'rows': len(df),
                    'columns': list(df.columns),
                    'data': df.to_dict('records')[:100],  # 限制前100行
                    'stats': self._get_dataframe_stats(df)
                }

            total_rows = sum(s['rows'] for s in result['sheets'].values())
            result['summary'] = f"{len(xl.sheet_names)} 个工作表，共 {total_rows} 行数据"

        except Exception as e:
            result['error'] = str(e)

        return result

    def read_csv(self, file_path: str) -> dict:
        """读取 CSV 文件"""
        result = {
            'type': 'CSV',
            'data': [],
            'summary': ''
        }

        try:
            df = pd.read_csv(file_path)
            result['rows'] = len(df)
            result['columns'] = list(df.columns)
            result['data'] = df.to_dict('records')[:100]
            result['stats'] = self._get_dataframe_stats(df)
            result['summary'] = f"{len(df)} 行, {len(df.columns)} 列"

        except Exception as e:
            result['error'] = str(e)

        return result

    def read_pdf(self, file_path: str) -> dict:
        """读取 PDF 文件"""
        result = {
            'type': 'PDF',
            'pages': [],
            'summary': ''
        }

        try:
            with open(file_path, 'rb') as f:
                reader = PyPDF2.PdfReader(f)
                for i, page in enumerate(reader.pages):
                    text = page.extract_text()
                    result['pages'].append({
                        'page': i + 1,
                        'text': text[:2000] if text else ''  # 限制每页2000字符
                    })

            result['summary'] = f"{len(result['pages'])} 页"

        except Exception as e:
            result['error'] = str(e)

        return result

    def read_docx(self, file_path: str) -> dict:
        """读取 Word 文件"""
        result = {
            'type': 'Word',
            'paragraphs': [],
            'tables': [],
            'summary': ''
        }

        try:
            doc = Document(file_path)

            # 读取段落
            for para in doc.paragraphs:
                if para.text.strip():
                    result['paragraphs'].append(para.text)

            # 读取表格
            for table in doc.tables:
                table_data = []
                for row in table.rows:
                    row_data = [cell.text for cell in row.cells]
                    table_data.append(row_data)
                result['tables'].append(table_data)

            result['summary'] = f"{len(result['paragraphs'])} 段落, {len(result['tables'])} 个表格"

        except Exception as e:
            result['error'] = str(e)

        return result

    def read_txt(self, file_path: str) -> dict:
        """读取 TXT 文件"""
        result = {
            'type': 'TXT',
            'content': '',
            'summary': ''
        }

        try:
            with open(file_path, 'r', encoding='utf-8') as f:
                content = f.read()
            result['content'] = content[:10000]  # 限制10000字符
            result['lines'] = len(content.split('\n'))
            result['summary'] = f"{result['lines']} 行, {len(content)} 字符"

        except Exception as e:
            result['error'] = str(e)

        return result

    def _get_dataframe_stats(self, df: pd.DataFrame) -> dict:
        """获取 DataFrame 的统计信息"""
        stats = {
            'numeric_columns': [],
            'categorical_columns': []
        }

        for col in df.columns:
            if pd.api.types.is_numeric_dtype(df[col]):
                stats['numeric_columns'].append({
                    'name': col,
                    'min': float(df[col].min()) if not pd.isna(df[col].min()) else None,
                    'max': float(df[col].max()) if not pd.isna(df[col].max()) else None,
                    'mean': float(df[col].mean()) if not pd.isna(df[col].mean()) else None,
                    'sum': float(df[col].sum()) if not pd.isna(df[col].sum()) else None
                })
            else:
                unique_values = df[col].nunique()
                stats['categorical_columns'].append({
                    'name': col,
                    'unique_count': unique_values,
                    'top_values': df[col].value_counts().head(5).to_dict()
                })

        return stats

    def process_file(self, file_path: str) -> dict:
        """处理单个文件"""
        path = Path(file_path)
        ext = path.suffix.lower()

        if ext in ['.xlsx', '.xls']:
            return self.read_excel(file_path)
        elif ext == '.csv':
            return self.read_csv(file_path)
        elif ext == '.pdf':
            return self.read_pdf(file_path)
        elif ext == '.docx':
            return self.read_docx(file_path)
        elif ext == '.txt':
            return self.read_txt(file_path)
        else:
            return {'error': f'不支持的文件格式: {ext}'}

    def process_all(self) -> dict:
        """处理所有上传的文件"""
        files = self.scan_files()
        results = {
            'processed_at': datetime.now().isoformat(),
            'files': []
        }

        for file_info in files:
            file_data = self.process_file(file_info['path'])
            file_data['name'] = file_info['name']
            file_data['size'] = file_info['size']
            results['files'].append(file_data)

        return results

def main():
    """主函数 - 处理上传目录中的所有文件"""
    processor = FileProcessor()
    results = processor.process_all()

    # 输出到 outputs 目录
    output_path = Path('outputs/reports/processed_data.json')
    output_path.parent.mkdir(parents=True, exist_ok=True)

    with open(output_path, 'w', encoding='utf-8') as f:
        json.dump(results, f, ensure_ascii=False, indent=2)

    print(f"处理完成，结果保存到: {output_path}")
    print(f"共处理 {len(results['files'])} 个文件")

if __name__ == '__main__':
    main()

#!/usr/bin/env python3
"""
报告构建器 - 将分析结论和图表整合为前端可用的 JSON 格式
"""

import json
from pathlib import Path
from datetime import datetime
from typing import Optional

class ReportBuilder:
    """报告构建器"""

    def __init__(self, output_dir: str = 'outputs/reports'):
        self.output_dir = Path(output_dir)
        self.output_dir.mkdir(parents=True, exist_ok=True)

    def build_report(
        self,
        title: str,
        overall_summary: str,
        conclusions: list,
        source_files: list,
        analysis_focus: Optional[dict] = None
    ) -> dict:
        """构建完整的报告数据结构"""

        report = {
            'meta': {
                'title': title,
                'generated_at': datetime.now().isoformat(),
                'version': '1.0'
            },
            'summary': {
                'overall': overall_summary,
                'total_conclusions': len(conclusions),
                'high_importance_count': len([c for c in conclusions if c.get('importance') == 'high']),
                'source_files': source_files
            },
            'analysis_focus': analysis_focus or {},
            'conclusions': conclusions
        }

        return report

    def save_report(self, report: dict, filename: str = 'report.json') -> str:
        """保存报告到 JSON 文件"""
        output_path = self.output_dir / filename

        with open(output_path, 'w', encoding='utf-8') as f:
            json.dump(report, f, ensure_ascii=False, indent=2)

        return str(output_path)

    def create_sample_report(self) -> dict:
        """创建示例报告（用于测试前端）"""

        conclusions = [
            {
                'id': 1,
                'title': 'Q3 销售额环比下降 23%',
                'description': '第三季度销售额为 850 万元，相比 Q2 的 1100 万元下降了 23%。主要受华东区域拖累，该区域下降 35%，需要重点关注。',
                'data_support': 'Q2: 1100万 → Q3: 850万，降幅 250万',
                'source_files': ['销售数据.xlsx'],
                'importance': 'high',
                'chart_type': 'line',
                'chart_data': {
                    'x_labels': ['Q1', 'Q2', 'Q3', 'Q4'],
                    'series': {
                        '实际': [980, 1100, 850, None],
                        '目标': [1000, 1000, 1000, 1000]
                    },
                    'x_title': '季度',
                    'y_title': '销售额（万元）'
                }
            },
            {
                'id': 2,
                'title': '用户满意度与复购率强相关',
                'description': '通过交叉分析发现，用户满意度每提升 10 分，复购率平均提升 8%。满意度低于 60 分的用户复购率仅 12%，而高于 80 分的用户复购率达 45%。',
                'data_support': '相关系数 r=0.82，P<0.01',
                'source_files': ['用户调研报告.pdf', '销售数据.xlsx'],
                'importance': 'high',
                'chart_type': 'scatter',
                'chart_data': {
                    'x': [45, 52, 58, 65, 72, 78, 85, 90],
                    'y': [8, 12, 18, 25, 32, 38, 45, 52],
                    'labels': ['用户群1', '用户群2', '用户群3', '用户群4', '用户群5', '用户群6', '用户群7', '用户群8'],
                    'x_title': '满意度评分',
                    'y_title': '复购率（%）'
                }
            },
            {
                'id': 3,
                'title': '华东区域占总销售的 42%',
                'description': '从区域分布来看，华东区域贡献了最大的销售份额（42%），其次是华南（28%）和华北（18%）。西部和其他区域占比较小，合计仅 12%。',
                'data_support': '华东 42%，华南 28%，华北 18%，西部 8%，其他 4%',
                'source_files': ['销售数据.xlsx'],
                'importance': 'medium',
                'chart_type': 'pie',
                'chart_data': {
                    'labels': ['华东', '华南', '华北', '西部', '其他'],
                    'values': [42, 28, 18, 8, 4]
                }
            },
            {
                'id': 4,
                'title': '线上渠道增速超过线下',
                'description': '线上渠道同比增长 35%，而线下渠道仅增长 8%。线上占比从去年的 25% 提升到今年的 38%，渠道结构正在发生显著变化。',
                'data_support': '线上同比 +35%，线下同比 +8%',
                'source_files': ['销售数据.xlsx'],
                'importance': 'medium',
                'chart_type': 'bar',
                'chart_data': {
                    'x_labels': ['线上渠道', '线下渠道'],
                    'series': {
                        '去年': [250, 750],
                        '今年': [338, 810]
                    },
                    'x_title': '渠道',
                    'y_title': '销售额（万元）'
                }
            },
            {
                'id': 5,
                'title': '新客获取成本上升 18%',
                'description': '今年新客获取成本（CAC）为 180 元/人，相比去年的 152 元上升了 18%。主要原因是竞争加剧和广告成本上涨。',
                'data_support': 'CAC: 152元 → 180元，涨幅 18%',
                'source_files': ['销售数据.xlsx'],
                'importance': 'low',
                'chart_type': 'bar',
                'chart_data': {
                    'x_labels': ['2023年', '2024年'],
                    'series': {
                        'CAC（元）': [152, 180]
                    },
                    'x_title': '年份',
                    'y_title': '获客成本（元）'
                }
            }
        ]

        report = self.build_report(
            title='销售数据分析报告',
            overall_summary='基于销售数据和用户调研报告的综合分析，发现 Q3 销售下滑主要受华东区域影响，用户满意度与复购率存在强相关性。建议：1) 重点改善华东区域服务质量；2) 加强满意度管理提升复购；3) 加速线上渠道建设。',
            conclusions=conclusions,
            source_files=['销售数据.xlsx', '用户调研报告.pdf'],
            analysis_focus={
                'core_question': '销售下滑的原因',
                'focus_metrics': 'Q3 数据、复购率',
                'usage_scenario': '内部汇报'
            }
        )

        return report

def main():
    """生成示例报告"""
    builder = ReportBuilder()
    report = builder.create_sample_report()
    output_path = builder.save_report(report)
    print(f"示例报告已生成: {output_path}")

if __name__ == '__main__':
    main()