EDA Reporter
CommunityFeed in a CSV or Excel file to get automatic data profiling, correlation analysis and outlier detection, with an interactive HTML report and a Markdown summary.
yangluxin v1.0.0
Install command
$ cow skill install eda-reporter Run in a terminal or send to CowAgent to install
ProductivityData
输入 CSV/Excel 文件，自动完成数据画像、相关性分析、异常值检测，生成带交互图表的 HTML 报告和 Markdown 摘要。
---
name: eda-reporter
description: 输入 CSV/Excel 文件，自动完成数据画像、相关性分析、异常值检测，生成带交互图表的 HTML 报告和 Markdown 摘要。
---
# EDA Reporter
> 自动化探索性数据分析（EDA）技能。输入 CSV 或 Excel 文件，自动完成数据质量检测、分布分析、相关性计算、异常值识别，并生成带交互图表的 HTML 报告和 Markdown 摘要。
## 能力说明
- **数据加载**：支持 CSV（自动检测编码）和 Excel（.xlsx / .xls），自动推断列类型（数值、分类、日期时间）
- **数据画像**：缺失值比例、唯一值计数、分布统计（均值/标准差/偏度/峰度）、分类频率
- **相关性分析**：数值列间 Pearson/Spearman 相关矩阵，识别高度相关特征对
- **异常值检测**：IQR 和 Z-Score 双算法，按列输出异常值数量和样本
- **可视化报告**：基于 ECharts 的交互式 HTML 报告（直方图、热力图、箱线图），以及 Markdown 纯文本摘要
## 使用方式
### 基本用法
```
请对 /path/to/data.csv 进行探索性数据分析，生成报告
```
```
分析这份 Excel 数据：/data/sales_2024.xlsx，重点关注数值列的异常值
```
### 完整参数
```bash
python scripts/render.py <input_file> [options]
参数：
input_file 输入文件路径（.csv 或 .xlsx/.xls）
--output <dir> 报告输出目录（默认：与输入文件同目录）
--format <html|md|both> 输出格式（默认：both）
--config <yaml> 自定义阈值配置文件（默认：resources/thresholds.yaml）
--title <str> 报告标题（默认：自动使用文件名）
--sample <int> 大文件采样行数（默认：不采样）
--corr-method <str> 相关系数方法 pearson|spearman（默认：pearson）
```
### 调用示例
```bash
# 生成完整 HTML + Markdown 报告
python scripts/render.py data/customers.csv --output reports/
# 仅生成 Markdown，适合在对话中展示
python scripts/render.py data/sales.xlsx --format md
# 大文件采样分析
python scripts/render.py data/logs_10m.csv --sample 50000 --format html
# 使用自定义异常检测阈值
python scripts/render.py data/metrics.csv --config my_thresholds.yaml
```
## 输出说明
### HTML 报告（report.html）
- 数据概览卡片（行数、列数、总缺失率、内存占用）
- 各列详细画像（含直方图或频率柱图）
- 相关性热力图（数值列）
- 异常值箱线图（含异常点高亮）
- 可交互，支持图表缩放和数据导出
### Markdown 摘要（report.md）
- 结构化文字描述，适合在对话窗口直接阅读
- 包含数据质量警告（缺失率 > 20%、高相关特征对、大量异常值）
- 附完整分析建议
## 依赖要求
### Python 包
```
pandas>=1.5.0
openpyxl>=3.0.0
numpy>=1.23.0
scipy>=1.9.0
scikit-learn>=1.1.0
jinja2>=3.1.0
pyyaml>=6.0
chardet>=5.0.0
```
安装：
```bash
pip install pandas openpyxl numpy scipy scikit-learn jinja2 pyyaml chardet
```
### 系统要求
- Python 3.8+
- 无需外部 API Key，完全本地运行
## 注意事项
- 纯数值列（如 ID 类）会被自动识别为分类列（唯一值比例 > 95%）
- 相关性分析要求至少 2 列数值列，否则跳过
- 超过 100 万行的文件建议使用 `--sample` 参数
- 日期时间列会自动解析，不参与数值统计，但会展示时间跨度信息
{
"tooltip": {
"trigger": "item",
"axisPointer": { "type": "shadow" },
"formatter": "function(p){ if(p.seriesIndex===0){ var v=p.data; return p.name+'<br/>最大值: '+v[5]+'<br/>Q3: '+v[4]+'<br/>中位数: '+v[3]+'<br/>Q1: '+v[2]+'<br/>最小值: '+v[1]; } return '异常值: '+p.data[1]; }"
},
"grid": {
"left": "5%",
"right": "5%",
"top": "10%",
"bottom": "15%",
"containLabel": true
},
"xAxis": {
"type": "category",
"boundaryGap": true,
"splitArea": { "show": false },
"splitLine": { "show": false },
"axisLabel": { "rotate": 30, "fontSize": 11 }
},
"yAxis": {
"type": "value",
"splitArea": { "show": true },
"splitLine": { "lineStyle": { "type": "dashed", "color": "#e0e0e0" } }
},
"series": [
{
"name": "箱线图",
"type": "boxplot",
"itemStyle": {
"color": "#5b8ff9",
"borderColor": "#3d6fd4",
"borderWidth": 1.5
},
"emphasis": {
"itemStyle": { "borderColor": "#1a3a9f", "shadowBlur": 6 }
}
},
{
"name": "异常值",
"type": "scatter",
"itemStyle": {
"color": "#e84749",
"opacity": 0.7
},
"symbolSize": 6
}
]
}
{
"tooltip": {
"position": "top",
"formatter": "function(p){ return p.data[0] + ' × ' + p.data[1] + '<br/>相关系数: ' + p.data[2].toFixed(3); }"
},
"grid": {
"left": "15%",
"right": "10%",
"top": "10%",
"bottom": "15%",
"containLabel": true
},
"xAxis": {
"type": "category",
"splitArea": { "show": true },
"axisLabel": { "rotate": 45, "fontSize": 11 }
},
"yAxis": {
"type": "category",
"splitArea": { "show": true },
"axisLabel": { "fontSize": 11 }
},
"visualMap": {
"min": -1,
"max": 1,
"calculable": true,
"orient": "horizontal",
"left": "center",
"bottom": "2%",
"inRange": {
"color": [
"#d73027",
"#f46d43",
"#fdae61",
"#fee08b",
"#ffffbf",
"#d9ef8b",
"#a6d96a",
"#66bd63",
"#1a9850"
]
},
"textStyle": { "fontSize": 11 }
},
"series": [
{
"type": "heatmap",
"label": {
"show": true,
"fontSize": 10,
"formatter": "function(p){ return p.data[2].toFixed(2); }"
},
"emphasis": {
"itemStyle": { "shadowBlur": 10, "shadowColor": "rgba(0,0,0,0.4)" }
}
}
]
}
{
"tooltip": {
"trigger": "axis",
"axisPointer": { "type": "shadow" },
"formatter": "{b}<br/>频次: {c}"
},
"grid": {
"left": "5%",
"right": "5%",
"top": "15%",
"bottom": "12%",
"containLabel": true
},
"xAxis": {
"type": "category",
"axisTick": { "alignWithLabel": true },
"axisLabel": { "rotate": 30, "fontSize": 11 }
},
"yAxis": {
"type": "value",
"name": "频次",
"nameTextStyle": { "fontSize": 11 },
"splitLine": { "lineStyle": { "type": "dashed", "color": "#e0e0e0" } }
},
"series": [
{
"type": "bar",
"barMaxWidth": 40,
"itemStyle": {
"color": {
"type": "linear",
"x": 0,
"y": 0,
"x2": 0,
"y2": 1,
"colorStops": [
{ "offset": 0, "color": "#5b8ff9" },
{ "offset": 1, "color": "#3d6fd4" }
]
},
"borderRadius": [3, 3, 0, 0]
},
"emphasis": {
"itemStyle": { "color": "#2450c0" }
}
}
]
}
<!doctype html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8" />
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
<title>{{ title }} — EDA Report</title>
<script src="https://cdn.jsdelivr.net/npm/echarts@5/dist/echarts.min.js"></script>
<style>
*,
*::before,
*::after {
box-sizing: border-box;
margin: 0;
padding: 0;
}
body {
font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", sans-serif;
background: #f4f6fa;
color: #1a1a2e;
font-size: 14px;
line-height: 1.6;
}
/* ── Layout ── */
.page {
max-width: 1280px;
margin: 0 auto;
padding: 24px 20px 60px;
}
header {
background: linear-gradient(135deg, #2c3e7a 0%, #1a4fa8 100%);
color: #fff;
border-radius: 12px;
padding: 28px 32px 24px;
margin-bottom: 24px;
}
header h1 {
font-size: 24px;
font-weight: 700;
letter-spacing: -0.3px;
}
header .meta {
margin-top: 8px;
opacity: 0.75;
font-size: 12px;
}
header .meta span {
margin-right: 20px;
}
.stat-grid {
display: grid;
grid-template-columns: repeat(auto-fill, minmax(160px, 1fr));
gap: 14px;
margin-bottom: 28px;
}
.stat-card {
background: #fff;
border-radius: 10px;
padding: 16px 18px;
box-shadow: 0 1px 4px rgba(0, 0, 0, 0.08);
}
.stat-card .label {
font-size: 11px;
color: #888;
text-transform: uppercase;
letter-spacing: 0.5px;
margin-bottom: 4px;
}
.stat-card .value {
font-size: 22px;
font-weight: 700;
color: #1a4fa8;
}
.stat-card .sub {
font-size: 11px;
color: #aaa;
margin-top: 2px;
}
section {
background: #fff;
border-radius: 12px;
padding: 24px 26px;
margin-bottom: 22px;
box-shadow: 0 1px 5px rgba(0, 0, 0, 0.07);
}
section h2 {
font-size: 16px;
font-weight: 700;
color: #1a2a5e;
margin-bottom: 16px;
padding-bottom: 10px;
border-bottom: 2px solid #eef1f8;
}
section h3 {
font-size: 13px;
font-weight: 600;
color: #334;
margin: 14px 0 6px;
}
/* ── Tables ── */
table {
width: 100%;
border-collapse: collapse;
font-size: 13px;
}
th {
background: #f0f4ff;
font-weight: 600;
color: #2c3e7a;
padding: 8px 12px;
text-align: left;
}
td {
padding: 7px 12px;
border-bottom: 1px solid #f0f2f7;
vertical-align: top;
}
tr:last-child td {
border-bottom: none;
}
tr:hover td {
background: #fafbff;
}
/* ── Badges ── */
.badge {
display: inline-block;
font-size: 10px;
font-weight: 600;
padding: 2px 8px;
border-radius: 20px;
text-transform: uppercase;
letter-spacing: 0.4px;
}
.badge-numeric {
background: #e8f0fe;
color: #1a4fa8;
}
.badge-categorical {
background: #fce8ff;
color: #7b1fa2;
}
.badge-datetime {
background: #e8f5e9;
color: #2e7d32;
}
.badge-text {
background: #fff3e0;
color: #e65100;
}
.badge-warn {
background: #fff3cd;
color: #856404;
}
.badge-critical {
background: #ffe0e0;
color: #c62828;
}
.badge-ok {
background: #e8f5e9;
color: #2e7d32;
}
/* ── Charts ── */
.chart-grid {
display: grid;
grid-template-columns: repeat(auto-fill, minmax(360px, 1fr));
gap: 18px;
margin-top: 14px;
}
.chart-card {
border: 1px solid #eef1f8;
border-radius: 10px;
padding: 14px 12px 10px;
background: #fafbff;
}
.chart-card .chart-title {
font-size: 12px;
font-weight: 600;
color: #555;
margin-bottom: 6px;
}
.chart-container {
width: 100%;
}
.chart-full {
border: 1px solid #eef1f8;
border-radius: 10px;
padding: 14px 12px 10px;
background: #fafbff;
margin-top: 14px;
}
/* ── Pair list ── */
.pair-list {
list-style: none;
}
.pair-list li {
display: flex;
align-items: center;
gap: 10px;
padding: 6px 0;
border-bottom: 1px solid #f4f4f8;
font-size: 13px;
}
.pair-list li:last-child {
border-bottom: none;
}
.pair-r {
font-weight: 700;
color: #1a4fa8;
min-width: 60px;
text-align: right;
}
.pair-r.negative {
color: #c62828;
}
.col-tag {
background: #e8f0fe;
color: #1a4fa8;
border-radius: 4px;
padding: 1px 7px;
font-size: 11px;
font-family: monospace;
}
/* ── Anomaly summary ── */
.anomaly-summary {
display: flex;
gap: 24px;
margin-bottom: 16px;
flex-wrap: wrap;
}
.anomaly-stat {
text-align: center;
}
.anomaly-stat .n {
font-size: 28px;
font-weight: 800;
color: #e84749;
}
.anomaly-stat .l {
font-size: 11px;
color: #888;
}
/* ── Warnings ── */
.warning-box {
background: #fff8e1;
border-left: 4px solid #f9a825;
border-radius: 0 8px 8px 0;
padding: 10px 14px;
margin-bottom: 10px;
font-size: 13px;
color: #5d4037;
}
.warning-box.critical {
background: #ffebee;
border-color: #e53935;
color: #b71c1c;
}
/* ── Missing bar ── */
.miss-bar-wrap {
display: flex;
align-items: center;
gap: 8px;
}
.miss-bar {
height: 6px;
border-radius: 3px;
background: #eee;
flex: 1;
max-width: 80px;
}
.miss-bar-fill {
height: 100%;
border-radius: 3px;
background: #5b8ff9;
}
.miss-bar-fill.warn {
background: #f9a825;
}
.miss-bar-fill.critical {
background: #e53935;
}
/* ── Responsive ── */
@media (max-width: 700px) {
.stat-grid {
grid-template-columns: repeat(2, 1fr);
}
.chart-grid {
grid-template-columns: 1fr;
}
}
</style>
</head>
<body>
<div class="page">
<!-- Header -->
<header>
<h1>{{ title }}</h1>
<div class="meta">
<span>📁 {{ metadata.file_name }}</span>
<span>💾 {{ metadata.file_size_mb }} MB</span>
<span>🕒 {{ generated_at }}</span>
{% if metadata.sampled %}<span
>🔀 采样 {{ "{:,}".format(metadata.sampled_rows) }} / {{
"{:,}".format(metadata.total_rows) }} 行</span
>{% endif %}
</div>
</header>
<!-- Overview stats -->
{% set ov = profile.overview %}
<div class="stat-grid">
<div class="stat-card">
<div class="label">行数</div>
<div class="value">{{ "{:,}".format(ov.rows) }}</div>
{% if ov.duplicate_rows > 0 %}
<div class="sub">{{ ov.duplicate_rows }} 重复行</div>
{% endif %}
</div>
<div class="stat-card">
<div class="label">列数</div>
<div class="value">{{ ov.cols }}</div>
<div class="sub">
{% for t, n in ov.column_type_counts.items() %}{{ t }}:{{ n }} {%
endfor %}
</div>
</div>
<div class="stat-card">
<div class="label">总缺失率</div>
<div class="value">{{ "%.1f"|format(ov.missing_rate * 100) }}%</div>
<div class="sub">{{ ov.total_missing }} 格</div>
</div>
<div class="stat-card">
<div class="label">内存占用</div>
<div class="value">{{ ov.memory_mb }}</div>
<div class="sub">MB</div>
</div>
<div class="stat-card">
<div class="label">异常值列</div>
<div class="value">{{ anomaly.summary.affected_columns }}</div>
<div class="sub">
共 {{ "{:,}".format(anomaly.summary.total_anomalies) }} 个
</div>
</div>
<div class="stat-card">
<div class="label">强相关对</div>
<div class="value">
{{ corr.strong_pairs | length if not corr.skipped else "—" }}
</div>
<div class="sub">
{% if not corr.skipped %}{{ corr.method }}{% else %}列数不足{% endif
%}
</div>
</div>
</div>
<!-- Column profiles -->
<section>
<h2>各列画像</h2>
<table>
<thead>
<tr>
<th>列名</th>
<th>类型</th>
<th>有效值</th>
<th>缺失率</th>
<th>唯一值</th>
<th>统计摘要</th>
<th>异常值</th>
</tr>
</thead>
<tbody>
{% for col, prof in profile.columns.items() %} {% set col_type =
prof.type %} {% set miss_r = prof.missing_rate %} {% set anom =
anomaly.columns.get(col, {}) %}
<tr>
<td><code style="font-size: 12px">{{ col }}</code></td>
<td>
<span class="badge badge-{{ col_type }}">{{ col_type }}</span>
</td>
<td>{{ "{:,}".format(prof.count) }}</td>
<td>
<div class="miss-bar-wrap">
<div class="miss-bar">
<div
class="miss-bar-fill {% if miss_r >= 0.5 %}critical{% elif miss_r >= 0.2 %}warn{% endif %}"
style="width:{{ [miss_r * 100, 100] | min }}%"
></div>
</div>
<span>{{ "%.1f"|format(miss_r * 100) }}%</span>
{% if miss_r >= 0.5 %}<span class="badge badge-critical"
>严重</span
>
{% elif miss_r >= 0.2 %}<span class="badge badge-warn"
>偏高</span
>{% endif %}
</div>
</td>
<td>
{{ "{:,}".format(prof.unique) if prof.unique is defined else "—"
}}
</td>
<td style="font-size: 12px; color: #555">
{% if col_type == "numeric" %} 均值 {{ prof.mean }}，σ {{
prof.std }}<br />
[{{ prof.min }} ~ {{ prof.max }}] {% if prof.skewness and
prof.skewness | abs > 1.0 %}
<span class="badge badge-warn" style="margin-top: 2px"
>偏态 {{ "%.2f"|format(prof.skewness) }}</span
>
{% endif %} {% elif col_type == "categorical" %} {% if
prof.top_values %} {% for v in prof.top_values[:3] %}{{ v.value
}}({{ v.count }}){% if not loop.last %}、{% endif %}{% endfor %}
{% if prof.has_more %} …{% endif %} {% endif %} {% elif col_type
== "datetime" %} {{ prof.min }}<br />~ {{ prof.max }} {% elif
col_type == "text" %} 均长 {{ prof.avg_length }}，最长 {{
prof.max_length }} {% endif %}
</td>
<td style="font-size: 12px">
{% if anom and not anom.get('skipped') %} {% if
anom.anomaly_count > 0 %}
<span style="color: #e84749; font-weight: 600"
>{{ anom.anomaly_count }}</span
>
<span style="color: #999"
>({{ "%.1f"|format(anom.anomaly_rate * 100) }}%)</span
>
{% else %}
<span style="color: #2e7d32">0</span>
{% endif %} {% else %}—{% endif %}
</td>
</tr>
{% endfor %}
</tbody>
</table>
</section>
<!-- Distribution charts -->
<section>
<h2>分布图</h2>
<div class="chart-grid" id="chart-grid-dist"></div>
</section>
<!-- Correlation heatmap -->
{% if not corr.skipped %}
<section>
<h2>相关性分析</h2>
{% if corr.multicollinear_pairs %} {% for p in corr.multicollinear_pairs
%}
<div class="warning-box critical">
多重共线性警告：<code>{{ p.col_a }}</code> 与
<code>{{ p.col_b }}</code> 相关系数 <strong>{{ p.r }}</strong>
</div>
{% endfor %} {% endif %} {% if corr.strong_pairs %}
<h3>
强相关特征对（|r| ≥ {{ config.get('correlation',
{}).get('strong_threshold', 0.8) }}）
</h3>
<ul class="pair-list" style="margin-bottom: 16px">
{% for p in corr.strong_pairs[:12] %}
<li>
<span class="col-tag">{{ p.col_a }}</span>
<span style="color: #bbb">×</span>
<span class="col-tag">{{ p.col_b }}</span>
<span class="pair-r {% if p.r < 0 %}negative{% endif %}"
>{{ p.r }}</span
>
</li>
{% endfor %}
</ul>
{% endif %}
<div class="chart-full">
<div id="chart-heatmap" class="chart-container"></div>
</div>
</section>
{% endif %}
<!-- Anomaly boxplot -->
{% if anomaly.summary.affected_columns > 0 %}
<section>
<h2>异常值分布</h2>
<div class="anomaly-summary">
<div class="anomaly-stat">
<div class="n">{{ anomaly.summary.total_anomalies }}</div>
<div class="l">异常值总数</div>
</div>
<div class="anomaly-stat">
<div class="n">{{ anomaly.summary.affected_columns }}</div>
<div class="l">受影响列</div>
</div>
</div>
{% for w in anomaly.warnings %}
<div class="warning-box">{{ w }}</div>
{% endfor %}
<div class="chart-full">
<div id="chart-boxplot" class="chart-container"></div>
</div>
</section>
{% endif %}
</div>
<!-- .page -->
<script>
// ── 图表数据（由 Jinja2 注入） ─────────────────────────────
const columnCharts = {{ charts.column_charts | tojson }};
const heatmapConfig = {{ charts.heatmap | tojson }};
const boxplotConfig = {{ charts.boxplot | tojson }};
// ── 渲染每列分布图 ────────────────────────────────────────
const distGrid = document.getElementById('chart-grid-dist');
Object.entries(columnCharts).forEach(([col, cfg]) => {
if (!cfg) return;
const card = document.createElement('div');
card.className = 'chart-card';
const titleDiv = document.createElement('div');
titleDiv.className = 'chart-title';
titleDiv.textContent = col;
card.appendChild(titleDiv);
const div = document.createElement('div');
div.className = 'chart-container';
div.style.height = (cfg._height || 300) + 'px';
card.appendChild(div);
distGrid.appendChild(card);
const chart = echarts.init(div);
// 去掉内部 _私有字段
const option = Object.fromEntries(Object.entries(cfg).filter(([k]) => !k.startsWith('_')));
chart.setOption(option);
});
// ── 渲染热力图 ───────────────────────────────────────────
if (heatmapConfig) {
const el = document.getElementById('chart-heatmap');
if (el) {
el.style.height = (heatmapConfig._height || 500) + 'px';
const chart = echarts.init(el);
const option = Object.fromEntries(Object.entries(heatmapConfig).filter(([k]) => !k.startsWith('_')));
// tooltip formatter 是字符串，需还原为函数
if (option.tooltip && typeof option.tooltip.formatter === 'string') {
try { option.tooltip.formatter = eval('(' + option.tooltip.formatter + ')'); } catch(e) {}
}
chart.setOption(option);
}
}
// ── 渲染箱线图 ───────────────────────────────────────────
if (boxplotConfig) {
const el = document.getElementById('chart-boxplot');
if (el) {
el.style.height = (boxplotConfig._height || 380) + 'px';
const chart = echarts.init(el);
const option = Object.fromEntries(Object.entries(boxplotConfig).filter(([k]) => !k.startsWith('_')));
if (option.tooltip && typeof option.tooltip.formatter === 'string') {
try { option.tooltip.formatter = eval('(' + option.tooltip.formatter + ')'); } catch(e) {}
}
chart.setOption(option);
}
}
// ── 响应窗口 resize ──────────────────────────────────────
window.addEventListener('resize', () => {
echarts.getInstanceByDom && document.querySelectorAll('.chart-container').forEach(el => {
const inst = echarts.getInstanceByDom(el);
if (inst) inst.resize();
});
});
</script>
</body>
</html>
# EDA Reporter - 分析阈值与行为配置
# ── 数据质量 ──────────────────────────────────────────────
data_quality:
# 缺失率超过此值时，在报告中标记为"高缺失"警告
missing_rate_warn: 0.20 # 20%
missing_rate_critical: 0.50 # 50%，标记为"严重缺失"
# 唯一值比例超过此值时，数值列视为 ID 类分类列（不做数值统计）
id_column_unique_ratio: 0.95
# 分类列基数超过此值时，不展示完整频率分布（只展示 Top N）
categorical_high_cardinality: 50
categorical_top_n: 20
# 数值列全为同一值（零方差）时跳过统计
skip_zero_variance: true
# ── 异常值检测 ────────────────────────────────────────────
anomaly:
# IQR 方法：outlier 定义为 < Q1 - k*IQR 或 > Q3 + k*IQR
iqr_multiplier: 1.5 # 标准值；改为 3.0 则更宽松
# Z-Score 方法：|z| > threshold 视为异常
zscore_threshold: 3.0
# 两种方法均标记为异常才计入最终异常列表（false = 任一方法即计入）
require_both_methods: false
# 某列异常率超过此值时，在报告顶部展示警告
anomaly_rate_warn: 0.05 # 5%
# ── 相关性分析 ────────────────────────────────────────────
correlation:
# 相关系数绝对值超过此值时，标记为"强相关"
strong_threshold: 0.80
# 超过此值时，在报告中发出"可能存在多重共线性"警告
multicollinearity_threshold: 0.95
# 相关性分析最少需要几列数值列
min_numeric_columns: 2
# 热力图最多展示前 N 列（避免矩阵过大）
heatmap_max_columns: 30
# ── 分布分析 ──────────────────────────────────────────────
distribution:
# 直方图 bin 数量（auto = 使用 Sturges 公式自动计算）
histogram_bins: auto
# 偏度绝对值超过此值时，标记为"高偏态"
skewness_warn: 1.0
# 峰度绝对值超过此值时，标记为"厚尾分布"
kurtosis_warn: 3.0
# ── 报告生成 ──────────────────────────────────────────────
report:
# 报告语言：zh（中文）或 en（英文）
language: zh
# HTML 报告中图表的默认高度（px）
chart_height: 350
# Markdown 摘要中每列描述最大字符数
md_description_max_chars: 200
# 是否在报告末尾附上原始统计数据表格（JSON）
include_raw_stats: false
# 大文件自动采样阈值（行数），超过则提示用户使用 --sample
auto_sample_warn_rows: 500000
"""
anomaly.py — 异常值检测模块
对数值列使用 IQR 和 Z-Score 双算法检测异常值。
"""
from __future__ import annotations
from typing import Any
import numpy as np
import pandas as pd
def _iqr_bounds(series: pd.Series, multiplier: float = 1.5) -> tuple[float, float]:
q1 = float(series.quantile(0.25))
q3 = float(series.quantile(0.75))
iqr = q3 - q1
return q1 - multiplier * iqr, q3 + multiplier * iqr
def _zscore_mask(series: pd.Series, threshold: float = 3.0) -> pd.Series:
mean = series.mean()
std = series.std()
if std == 0:
return pd.Series([False] * len(series), index=series.index)
return ((series - mean) / std).abs() > threshold
def detect_anomalies(
df: pd.DataFrame,
column_types: dict[str, str],
iqr_multiplier: float = 1.5,
zscore_threshold: float = 3.0,
require_both: bool = False,
anomaly_rate_warn: float = 0.05,
) -> dict[str, Any]:
"""
对所有数值列检测异常值。
返回：
{
"columns": {
col: {
"total": int,
"anomaly_count": int,
"anomaly_rate": float,
"iqr_only": int,
"zscore_only": int,
"both": int,
"bounds": {"iqr_lower": float, "iqr_upper": float},
"boxplot_data": [min, Q1, median, Q3, max],
"outlier_values": [float, ...], # 最多 20 个样本
"warning": bool,
}
},
"warnings": [str, ...], # 高异常率列的警告信息
"summary": {
"total_anomalies": int,
"affected_columns": int,
}
}
"""
numeric_cols = [c for c, t in column_types.items() if t == "numeric"]
result_cols: dict[str, Any] = {}
warnings: list[str] = []
total_anomalies = 0
affected_columns = 0
for col in numeric_cols:
series = df[col].dropna()
n = len(series)
if n < 4:
result_cols[col] = {
"total": n,
"anomaly_count": 0,
"anomaly_rate": 0.0,
"skipped": True,
"reason": "有效值不足 4 个",
}
continue
# IQR 检测
lower, upper = _iqr_bounds(series, iqr_multiplier)
iqr_mask = (series < lower) | (series > upper)
# Z-Score 检测
zscore_mask = _zscore_mask(series, zscore_threshold)
# 合并
if require_both:
anomaly_mask = iqr_mask & zscore_mask
else:
anomaly_mask = iqr_mask | zscore_mask
iqr_only = int((iqr_mask & ~zscore_mask).sum())
zscore_only = int((~iqr_mask & zscore_mask).sum())
both = int((iqr_mask & zscore_mask).sum())
anomaly_count = int(anomaly_mask.sum())
anomaly_rate = round(anomaly_count / n, 4)
# 箱线图数据（ECharts boxplot 格式）
q1 = float(series.quantile(0.25))
q3 = float(series.quantile(0.75))
boxplot_data = [
round(float(series.min()), 6),
round(q1, 6),
round(float(series.median()), 6),
round(q3, 6),
round(float(series.max()), 6),
]
# 异常值样本（最多 20 个，按绝对偏差降序）
outlier_series = series[anomaly_mask]
mean_val = series.mean()
outlier_sorted = outlier_series.reindex(
(outlier_series - mean_val).abs().sort_values(ascending=False).index
).head(20)
outlier_values = [round(float(v), 6) for v in outlier_sorted]
has_warning = anomaly_rate >= anomaly_rate_warn
if has_warning:
warnings.append(
f"列 '{col}' 异常率 {anomaly_rate:.1%}（{anomaly_count}/{n}），"
f"建议检查数据来源"
)
result_cols[col] = {
"total": n,
"anomaly_count": anomaly_count,
"anomaly_rate": anomaly_rate,
"iqr_only": iqr_only,
"zscore_only": zscore_only,
"both": both,
"bounds": {
"iqr_lower": round(lower, 6),
"iqr_upper": round(upper, 6),
"zscore_lower": round(float(series.mean() - zscore_threshold * series.std()), 6),
"zscore_upper": round(float(series.mean() + zscore_threshold * series.std()), 6),
},
"boxplot_data": boxplot_data,
"outlier_values": outlier_values,
"warning": has_warning,
}
if anomaly_count > 0:
total_anomalies += anomaly_count
affected_columns += 1
return {
"columns": result_cols,
"warnings": warnings,
"summary": {
"total_anomalies": total_anomalies,
"affected_columns": affected_columns,
"numeric_columns_checked": len(numeric_cols),
},
}
"""
correlate.py — 相关性分析模块
计算数值列间相关矩阵，识别强相关特征对。
"""
from __future__ import annotations
from typing import Any
import numpy as np
import pandas as pd
from scipy import stats
def compute_correlation(
df: pd.DataFrame,
column_types: dict[str, str],
method: str = "pearson",
strong_threshold: float = 0.80,
multicollinearity_threshold: float = 0.95,
max_columns: int = 30,
) -> dict[str, Any]:
"""
计算数值列间相关矩阵。
返回：
{
"columns": [...],
"matrix": [[...], ...], # 行列均为 columns
"strong_pairs": [...], # |r| >= strong_threshold 的列对
"multicollinear_pairs": [...],# |r| >= multicollinearity_threshold
"method": "pearson"|"spearman",
"skipped": bool, # 数值列不足时为 True
}
"""
numeric_cols = [c for c, t in column_types.items() if t == "numeric"]
if len(numeric_cols) < 2:
return {
"skipped": True,
"reason": f"数值列不足（仅有 {len(numeric_cols)} 列），至少需要 2 列",
"columns": [],
"matrix": [],
"strong_pairs": [],
"multicollinear_pairs": [],
"method": method,
}
# 列数过多时截取前 max_columns 列（按方差降序）
if len(numeric_cols) > max_columns:
variances = df[numeric_cols].var().sort_values(ascending=False)
numeric_cols = variances.index[:max_columns].tolist()
sub = df[numeric_cols].copy()
if method == "spearman":
corr_matrix = sub.rank().corr(method="pearson")
else:
corr_matrix = sub.corr(method="pearson")
# 将 NaN 替换为 0（列方差为零时会出现）
corr_matrix = corr_matrix.fillna(0)
# 提取强相关对（上三角，排除对角线）
strong_pairs: list[dict[str, Any]] = []
multicollinear_pairs: list[dict[str, Any]] = []
n = len(numeric_cols)
for i in range(n):
for j in range(i + 1, n):
r = float(corr_matrix.iloc[i, j])
abs_r = abs(r)
if abs_r >= strong_threshold:
pair = {
"col_a": numeric_cols[i],
"col_b": numeric_cols[j],
"r": round(r, 4),
"abs_r": round(abs_r, 4),
}
strong_pairs.append(pair)
if abs_r >= multicollinearity_threshold:
multicollinear_pairs.append(pair)
# 按相关系数绝对值降序
strong_pairs.sort(key=lambda x: x["abs_r"], reverse=True)
multicollinear_pairs.sort(key=lambda x: x["abs_r"], reverse=True)
# 矩阵数据（用于热力图）
# ECharts heatmap 格式：[[col_i, col_j, value], ...]
heatmap_data: list[list] = []
for i, ci in enumerate(numeric_cols):
for j, cj in enumerate(numeric_cols):
heatmap_data.append([ci, cj, round(float(corr_matrix.loc[ci, cj]), 4)])
return {
"skipped": False,
"columns": numeric_cols,
"matrix": corr_matrix.round(4).values.tolist(),
"heatmap_data": heatmap_data,
"strong_pairs": strong_pairs,
"multicollinear_pairs": multicollinear_pairs,
"method": method,
"truncated": len([c for c, t in column_types.items() if t == "numeric"]) > max_columns,
}
"""
loader.py — 数据加载模块
支持 CSV（自动检测编码）和 Excel，自动推断列类型。
"""
from __future__ import annotations
import json
import os
import sys
from pathlib import Path
from typing import Any
import chardet
import numpy as np
import pandas as pd
# 常见空值字符串
_NULL_VALUES = {"", "null", "NULL", "None", "none", "NA", "na", "N/A", "n/a",
"NaN", "nan", "#N/A", "#NA", "-", "--", "?"}
# 唯一值比例超过此值时视为 ID 列（分类）
_ID_UNIQUE_RATIO = 0.95
def _detect_encoding(path: str) -> str:
with open(path, "rb") as f:
raw = f.read(min(os.path.getsize(path), 100_000))
result = chardet.detect(raw)
encoding = result.get("encoding") or "utf-8"
# 常见别名统一
return {"GB2312": "gbk", "GBK": "gbk"}.get(encoding.upper(), encoding)
def _infer_column_type(series: pd.Series, id_unique_ratio: float = _ID_UNIQUE_RATIO) -> str:
"""推断列的语义类型：numeric / categorical / datetime / text。"""
if series.dtype == "bool":
return "categorical"
# 尝试解析日期时间
if series.dtype == object:
sample = series.dropna().head(200)
try:
parsed = pd.to_datetime(sample, infer_datetime_format=True, errors="coerce")
if parsed.notna().mean() > 0.8:
return "datetime"
except Exception:
pass
if pd.api.types.is_numeric_dtype(series):
# 唯一值极高 → ID 类分类列
n_unique = series.nunique()
n_total = series.count()
if n_total > 0 and (n_unique / n_total) >= id_unique_ratio and n_unique > 50:
return "categorical"
return "numeric"
# object / string
n_unique = series.nunique()
n_total = series.count()
if n_total > 0 and (n_unique / n_total) >= id_unique_ratio and n_unique > 50:
return "text"
return "categorical"
def load_file(
path: str,
sample: int | None = None,
id_unique_ratio: float = _ID_UNIQUE_RATIO,
) -> tuple[pd.DataFrame, dict[str, Any]]:
"""
加载 CSV 或 Excel 文件，返回 (DataFrame, metadata)。
metadata 包含：
- file_name, file_size_mb, total_rows, total_cols
- encoding（CSV 专用）
- column_types: {col: "numeric"|"categorical"|"datetime"|"text"}
- sampled: bool
"""
p = Path(path)
if not p.exists():
raise FileNotFoundError(f"文件不存在: {path}")
suffix = p.suffix.lower()
file_size_mb = round(p.stat().st_size / 1024 / 1024, 2)
encoding = None
if suffix == ".csv":
encoding = _detect_encoding(path)
df = pd.read_csv(
path,
encoding=encoding,
na_values=list(_NULL_VALUES),
keep_default_na=True,
low_memory=False,
)
elif suffix in (".xlsx", ".xls"):
df = pd.read_excel(path, na_values=list(_NULL_VALUES))
else:
raise ValueError(f"不支持的文件格式: {suffix}（仅支持 .csv / .xlsx / .xls）")
total_rows = len(df)
# 列名清洗：去除首尾空格
df.columns = [str(c).strip() for c in df.columns]
# 采样
sampled = False
if sample and len(df) > sample:
df = df.sample(n=sample, random_state=42).reset_index(drop=True)
sampled = True
# 尝试将 object 列转为数值
for col in df.select_dtypes(include="object").columns:
converted = pd.to_numeric(df[col], errors="coerce")
if converted.notna().sum() / max(df[col].count(), 1) > 0.9:
df[col] = converted
# 尝试解析日期时间列
for col in df.select_dtypes(include="object").columns:
try:
parsed = pd.to_datetime(df[col], infer_datetime_format=True, errors="coerce")
if parsed.notna().mean() > 0.8:
df[col] = parsed
except Exception:
pass
# 推断列类型
column_types: dict[str, str] = {}
for col in df.columns:
column_types[col] = _infer_column_type(df[col], id_unique_ratio)
metadata: dict[str, Any] = {
"file_name": p.name,
"file_path": str(p.resolve()),
"file_size_mb": file_size_mb,
"total_rows": total_rows,
"sampled_rows": len(df),
"total_cols": len(df.columns),
"encoding": encoding,
"sampled": sampled,
"column_types": column_types,
"columns": list(df.columns),
}
return df, metadata
# ── CLI 入口（供独立调试） ───────────────────────────────────
if __name__ == "__main__":
if len(sys.argv) < 2:
print("Usage: python loader.py <file_path> [sample_rows]")
sys.exit(1)
_path = sys.argv[1]
_sample = int(sys.argv[2]) if len(sys.argv) > 2 else None
_df, _meta = load_file(_path, sample=_sample)
print(json.dumps(_meta, ensure_ascii=False, indent=2))
print(f"\n前 3 行预览：\n{_df.head(3).to_string()}")
"""
profiler.py — 数据画像模块
计算每列的统计信息：缺失值、唯一值、分布统计、分类频率等。
"""
from __future__ import annotations
import math
from typing import Any
import numpy as np
import pandas as pd
from scipy import stats
def _safe_float(v: Any) -> float | None:
"""将 numpy 标量安全转为 Python float，NaN/Inf 返回 None。"""
try:
f = float(v)
return None if (math.isnan(f) or math.isinf(f)) else round(f, 6)
except (TypeError, ValueError):
return None
def profile_numeric(series: pd.Series, bins: int | str = "auto") -> dict[str, Any]:
"""对数值列生成统计画像。"""
clean = series.dropna()
n = len(clean)
if n == 0:
return {"count": 0, "missing": len(series), "missing_rate": 1.0}
# 基础统计
desc = clean.describe(percentiles=[0.01, 0.05, 0.25, 0.5, 0.75, 0.95, 0.99])
# 偏度 & 峰度
skewness = _safe_float(stats.skew(clean))
kurtosis = _safe_float(stats.kurtosis(clean)) # excess kurtosis
# 直方图数据
if bins == "auto":
n_bins = max(5, min(50, int(1 + 3.322 * math.log10(n)))) # Sturges
else:
n_bins = int(bins)
counts, edges = np.histogram(clean, bins=n_bins)
hist_labels = [f"{edges[i]:.4g}–{edges[i+1]:.4g}" for i in range(len(edges) - 1)]
return {
"type": "numeric",
"count": int(clean.count()),
"missing": int(series.isna().sum()),
"missing_rate": round(series.isna().mean(), 4),
"unique": int(clean.nunique()),
"mean": _safe_float(desc["mean"]),
"std": _safe_float(desc["std"]),
"min": _safe_float(desc["min"]),
"p1": _safe_float(desc["1%"]),
"p5": _safe_float(desc["5%"]),
"q1": _safe_float(desc["25%"]),
"median": _safe_float(desc["50%"]),
"q3": _safe_float(desc["75%"]),
"p95": _safe_float(desc["95%"]),
"p99": _safe_float(desc["99%"]),
"max": _safe_float(desc["max"]),
"skewness": skewness,
"kurtosis": kurtosis,
"zero_count": int((clean == 0).sum()),
"negative_count": int((clean < 0).sum()),
"histogram": {
"labels": hist_labels,
"counts": counts.tolist(),
},
}
def profile_categorical(series: pd.Series, top_n: int = 20) -> dict[str, Any]:
"""对分类列生成统计画像。"""
n_missing = int(series.isna().sum())
clean = series.dropna()
n = len(clean)
value_counts = clean.value_counts()
top = value_counts.head(top_n)
return {
"type": "categorical",
"count": n,
"missing": n_missing,
"missing_rate": round(series.isna().mean(), 4),
"unique": int(clean.nunique()),
"top_values": [
{"value": str(k), "count": int(v), "rate": round(v / n, 4)}
for k, v in top.items()
],
"has_more": len(value_counts) > top_n,
}
def profile_datetime(series: pd.Series) -> dict[str, Any]:
"""对日期时间列生成统计画像。"""
n_missing = int(series.isna().sum())
clean = series.dropna()
if len(clean) == 0:
return {"type": "datetime", "count": 0, "missing": n_missing}
dt = pd.to_datetime(clean, errors="coerce").dropna()
return {
"type": "datetime",
"count": len(dt),
"missing": n_missing,
"missing_rate": round(series.isna().mean(), 4),
"min": str(dt.min()),
"max": str(dt.max()),
"span_days": int((dt.max() - dt.min()).days),
"unique": int(dt.nunique()),
}
def profile_text(series: pd.Series) -> dict[str, Any]:
"""对高基数文本列生成简要画像。"""
n_missing = int(series.isna().sum())
clean = series.dropna().astype(str)
lengths = clean.str.len()
return {
"type": "text",
"count": len(clean),
"missing": n_missing,
"missing_rate": round(series.isna().mean(), 4),
"unique": int(clean.nunique()),
"avg_length": _safe_float(lengths.mean()),
"max_length": int(lengths.max()) if len(lengths) else 0,
"min_length": int(lengths.min()) if len(lengths) else 0,
}
def profile_dataframe(
df: pd.DataFrame,
column_types: dict[str, str],
bins: int | str = "auto",
top_n: int = 20,
) -> dict[str, Any]:
"""
对整个 DataFrame 生成完整画像。
返回：
{
"overview": {...},
"columns": { col_name: {...stats...} }
}
"""
total_cells = df.shape[0] * df.shape[1]
total_missing = int(df.isna().sum().sum())
memory_mb = round(df.memory_usage(deep=True).sum() / 1024 / 1024, 2)
# 各类型列数
type_counts: dict[str, int] = {}
for t in column_types.values():
type_counts[t] = type_counts.get(t, 0) + 1
overview = {
"rows": df.shape[0],
"cols": df.shape[1],
"total_cells": total_cells,
"total_missing": total_missing,
"missing_rate": round(total_missing / total_cells, 4) if total_cells else 0,
"memory_mb": memory_mb,
"column_type_counts": type_counts,
"duplicate_rows": int(df.duplicated().sum()),
}
columns: dict[str, Any] = {}
for col in df.columns:
col_type = column_types.get(col, "categorical")
if col_type == "numeric":
columns[col] = profile_numeric(df[col], bins=bins)
elif col_type == "datetime":
columns[col] = profile_datetime(df[col])
elif col_type == "text":
columns[col] = profile_text(df[col])
else:
columns[col] = profile_categorical(df[col], top_n=top_n)
return {"overview": overview, "columns": columns}
"""
render.py — 主入口：编排所有分析步骤，生成 HTML + Markdown 报告。
用法：
python render.py <input_file> [--output <dir>] [--format html|md|both]
[--config <yaml>] [--title <str>] [--sample <int>]
[--corr-method pearson|spearman]
"""
from __future__ import annotations
import argparse
import json
import os
import sys
from datetime import datetime
from pathlib import Path
from typing import Any
import yaml
from jinja2 import Environment, FileSystemLoader
# 将 scripts/ 目录加入 path，支持直接运行
sys.path.insert(0, str(Path(__file__).parent))
from anomaly import detect_anomalies
from correlate import compute_correlation
from loader import load_file
from profiler import profile_dataframe
from visualize import build_all_charts
# ── 配置加载 ──────────────────────────────────────────────
def _load_config(config_path: str | None) -> dict:
default_path = Path(__file__).parent.parent / "resources" / "thresholds.yaml"
path = Path(config_path) if config_path else default_path
if not path.exists():
return {}
with open(path, encoding="utf-8") as f:
return yaml.safe_load(f) or {}
def _get(cfg: dict, *keys: str, default=None):
"""安全读取嵌套配置。"""
v = cfg
for k in keys:
if not isinstance(v, dict):
return default
v = v.get(k, default)
return v
# ── Markdown 报告生成 ─────────────────────────────────────
def _fmt_rate(r: float) -> str:
return f"{r * 100:.1f}%"
def _quality_badge(rate: float, warn: float, critical: float) -> str:
if rate >= critical:
return "严重缺失"
if rate >= warn:
return "高缺失"
return "正常"
def build_markdown(
metadata: dict,
profile_result: dict,
corr_result: dict,
anomaly_result: dict,
config: dict,
title: str,
) -> str:
miss_warn = _get(config, "data_quality", "missing_rate_warn", default=0.2)
miss_crit = _get(config, "data_quality", "missing_rate_critical", default=0.5)
skew_warn = _get(config, "distribution", "skewness_warn", default=1.0)
kurt_warn = _get(config, "distribution", "kurtosis_warn", default=3.0)
anomaly_warn = _get(config, "anomaly", "anomaly_rate_warn", default=0.05)
strong_thr = _get(config, "correlation", "strong_threshold", default=0.8)
ov = profile_result["overview"]
lines: list[str] = []
lines.append(f"# {title}\n")
lines.append(f"> 生成时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')} ")
lines.append(f"> 文件：`{metadata['file_name']}` ({metadata['file_size_mb']} MB)\n")
# 概览
lines.append("## 数据概览\n")
lines.append(f"| 指标 | 值 |")
lines.append(f"|------|----|")
lines.append(f"| 行数 | {ov['rows']:,} |")
lines.append(f"| 列数 | {ov['cols']} |")
lines.append(f"| 重复行 | {ov['duplicate_rows']:,} ({_fmt_rate(ov['duplicate_rows'] / max(ov['rows'], 1))}) |")
lines.append(f"| 总缺失率 | {_fmt_rate(ov['missing_rate'])} |")
lines.append(f"| 内存占用 | {ov['memory_mb']} MB |")
if metadata.get("sampled"):
lines.append(f"| 采样 | 是（{metadata['sampled_rows']:,} 行，原始 {metadata['total_rows']:,} 行）|")
lines.append("")
type_counts = ov.get("column_type_counts", {})
type_str = "、".join(f"{t}列 {n} 个" for t, n in type_counts.items())
lines.append(f"列类型分布：{type_str}\n")
# 数据质量警告
quality_warnings: list[str] = []
for col, prof in profile_result["columns"].items():
r = prof.get("missing_rate", 0)
if r >= miss_warn:
badge = _quality_badge(r, miss_warn, miss_crit)
quality_warnings.append(f"- **{col}**：缺失率 {_fmt_rate(r)}（{badge}）")
if quality_warnings:
lines.append("### 数据质量警告\n")
lines.extend(quality_warnings)
lines.append("")
# 各列摘要
lines.append("## 各列分析\n")
for col, prof in profile_result["columns"].items():
col_type = prof.get("type", "unknown")
lines.append(f"### `{col}` ({col_type})\n")
miss_r = prof.get("missing_rate", 0)
lines.append(f"- 有效值：{prof.get('count', 0):,}，缺失：{prof.get('missing', 0):,}（{_fmt_rate(miss_r)}）")
if col_type == "numeric":
lines.append(f"- 均值：{prof.get('mean')}，标准差：{prof.get('std')}")
lines.append(f"- 范围：[{prof.get('min')} ~ {prof.get('max')}]，中位数：{prof.get('median')}")
skew = prof.get("skewness")
kurt = prof.get("kurtosis")
flags = []
if skew is not None and abs(skew) > skew_warn:
flags.append(f"高偏态（偏度={skew:.2f}）")
if kurt is not None and abs(kurt) > kurt_warn:
flags.append(f"厚尾（峰度={kurt:.2f}）")
if flags:
lines.append(f"- 分布特征：{'，'.join(flags)}")
# 异常值信息
anom = anomaly_result["columns"].get(col, {})
if not anom.get("skipped") and anom.get("anomaly_count", 0) > 0:
lines.append(
f"- 异常值：{anom['anomaly_count']} 个（{_fmt_rate(anom['anomaly_rate'])}），"
f"IQR 区间 [{anom['bounds']['iqr_lower']:.4g}, {anom['bounds']['iqr_upper']:.4g}]"
)
elif col_type == "categorical":
lines.append(f"- 唯一值：{prof.get('unique', 0)}")
top = prof.get("top_values", [])[:5]
if top:
top_str = "、".join(f"{v['value']}({v['count']})" for v in top)
lines.append(f"- 前 5 值：{top_str}")
elif col_type == "datetime":
lines.append(f"- 时间跨度：{prof.get('min')} ~ {prof.get('max')}（{prof.get('span_days', 0)} 天）")
lines.append("")
# 相关性
lines.append("## 相关性分析\n")
if corr_result.get("skipped"):
lines.append(f"> {corr_result.get('reason')}\n")
else:
method = corr_result.get("method", "pearson")
lines.append(f"使用 **{method}** 相关系数，分析 {len(corr_result['columns'])} 个数值列。\n")
strong = corr_result.get("strong_pairs", [])
if strong:
lines.append(f"### 强相关特征对（|r| ≥ {strong_thr}）\n")
for p in strong[:10]:
lines.append(f"- `{p['col_a']}` × `{p['col_b']}`：r = **{p['r']}**")
lines.append("")
mc = corr_result.get("multicollinear_pairs", [])
if mc:
lines.append("### 多重共线性警告\n")
for p in mc:
lines.append(f"- `{p['col_a']}` 与 `{p['col_b']}` 高度相关（r = {p['r']}），建议检查是否冗余特征")
lines.append("")
# 异常值汇总
lines.append("## 异常值汇总\n")
asum = anomaly_result.get("summary", {})
lines.append(f"- 检测列数：{asum.get('numeric_columns_checked', 0)}")
lines.append(f"- 受影响列：{asum.get('affected_columns', 0)}")
lines.append(f"- 异常值总计：{asum.get('total_anomalies', 0):,}\n")
for w in anomaly_result.get("warnings", []):
lines.append(f"> **警告**：{w}")
if anomaly_result.get("warnings"):
lines.append("")
# 分析建议
lines.append("## 分析建议\n")
suggestions: list[str] = []
if ov["missing_rate"] > miss_warn:
suggestions.append("整体缺失率较高，建议优先处理缺失值（填充或删除）。")
if ov["duplicate_rows"] > 0:
suggestions.append(f"存在 {ov['duplicate_rows']:,} 条重复行，建议去重后再建模。")
if corr_result.get("multicollinear_pairs"):
suggestions.append("存在高度相关特征，建议使用 PCA 降维或人工筛选冗余列。")
if anomaly_result["summary"]["affected_columns"] > 0:
suggestions.append("存在异常值，建议结合业务背景决定是否 Winsorize 或删除。")
if suggestions:
for s in suggestions:
lines.append(f"- {s}")
else:
lines.append("- 数据整体质量良好，可直接用于建模分析。")
lines.append("")
return "\n".join(lines)
# ── HTML 报告生成 ─────────────────────────────────────────
def build_html(
metadata: dict,
profile_result: dict,
corr_result: dict,
anomaly_result: dict,
charts: dict,
config: dict,
title: str,
) -> str:
template_path = Path(__file__).parent.parent / "resources" / "report_template.html"
env = Environment(
loader=FileSystemLoader(str(template_path.parent)),
autoescape=False,
)
def to_json(v: Any) -> str:
return json.dumps(v, ensure_ascii=False)
env.filters["tojson"] = to_json
template = env.get_template(template_path.name)
return template.render(
title=title,
generated_at=datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
metadata=metadata,
profile=profile_result,
corr=corr_result,
anomaly=anomaly_result,
charts=charts,
config=config,
)
# ── 主流程 ───────────────────────────────────────────────
def run(
input_file: str,
output_dir: str | None = None,
fmt: str = "both",
config_path: str | None = None,
title: str | None = None,
sample: int | None = None,
corr_method: str = "pearson",
) -> dict[str, str]:
"""
执行完整 EDA 流程，返回 {"html": path, "md": path} 中已生成的项。
"""
cfg = _load_config(config_path)
# 1. 加载数据
print(f"[1/5] 加载文件: {input_file}")
id_ratio = _get(cfg, "data_quality", "id_column_unique_ratio", default=0.95)
df, metadata = load_file(input_file, sample=sample, id_unique_ratio=id_ratio)
print(f" {metadata['sampled_rows']:,} 行 × {metadata['total_cols']} 列")
report_title = title or Path(input_file).stem
out_dir = Path(output_dir) if output_dir else Path(input_file).parent
out_dir.mkdir(parents=True, exist_ok=True)
# 2. 数据画像
print("[2/5] 数据画像...")
bins = _get(cfg, "distribution", "histogram_bins", default="auto")
top_n = _get(cfg, "data_quality", "categorical_top_n", default=20)
profile_result = profile_dataframe(df, metadata["column_types"], bins=bins, top_n=top_n)
# 3. 相关性分析
print("[3/5] 相关性分析...")
corr_result = compute_correlation(
df,
metadata["column_types"],
method=corr_method,
strong_threshold=_get(cfg, "correlation", "strong_threshold", default=0.80),
multicollinearity_threshold=_get(cfg, "correlation", "multicollinearity_threshold", default=0.95),
max_columns=_get(cfg, "correlation", "heatmap_max_columns", default=30),
)
# 4. 异常值检测
print("[4/5] 异常值检测...")
anomaly_result = detect_anomalies(
df,
metadata["column_types"],
iqr_multiplier=_get(cfg, "anomaly", "iqr_multiplier", default=1.5),
zscore_threshold=_get(cfg, "anomaly", "zscore_threshold", default=3.0),
require_both=_get(cfg, "anomaly", "require_both_methods", default=False),
anomaly_rate_warn=_get(cfg, "anomaly", "anomaly_rate_warn", default=0.05),
)
# 5. 生成报告
print("[5/5] 生成报告...")
charts = build_all_charts(
profile_result,
corr_result,
anomaly_result,
chart_height=_get(cfg, "report", "chart_height", default=350),
)
output_files: dict[str, str] = {}
base_name = Path(input_file).stem
if fmt in ("md", "both"):
md_content = build_markdown(metadata, profile_result, corr_result, anomaly_result, cfg, report_title)
md_path = out_dir / f"{base_name}_eda_report.md"
md_path.write_text(md_content, encoding="utf-8")
output_files["md"] = str(md_path)
print(f" Markdown → {md_path}")
if fmt in ("html", "both"):
html_content = build_html(metadata, profile_result, corr_result, anomaly_result, charts, cfg, report_title)
html_path = out_dir / f"{base_name}_eda_report.html"
html_path.write_text(html_content, encoding="utf-8")
output_files["html"] = str(html_path)
print(f" HTML → {html_path}")
print("\n✓ 分析完成")
return output_files
# ── CLI ──────────────────────────────────────────────────
def main() -> None:
parser = argparse.ArgumentParser(description="EDA Reporter — 自动探索性数据分析报告生成器")
parser.add_argument("input_file", help="输入文件路径（.csv / .xlsx / .xls）")
parser.add_argument("--output", "-o", default=None, help="报告输出目录（默认：与输入文件同目录）")
parser.add_argument("--format", "-f", choices=["html", "md", "both"], default="both", dest="fmt")
parser.add_argument("--config", "-c", default=None, help="thresholds.yaml 路径")
parser.add_argument("--title", "-t", default=None, help="报告标题")
parser.add_argument("--sample", "-s", type=int, default=None, help="大文件采样行数")
parser.add_argument("--corr-method", choices=["pearson", "spearman"], default="pearson")
args = parser.parse_args()
result = run(
input_file=args.input_file,
output_dir=args.output,
fmt=args.fmt,
config_path=args.config,
title=args.title,
sample=args.sample,
corr_method=args.corr_method,
)
for k, v in result.items():
print(f"{k.upper()}: {v}")
if __name__ == "__main__":
main()
"""
visualize.py — 图表数据生成模块
将 profile/correlate/anomaly 分析结果转换为 ECharts 配置对象。
读取 resources/chart_configs/ 中的基础配置并合并实际数据。
"""
from __future__ import annotations
import copy
import json
import os
from pathlib import Path
from typing import Any
# resources/chart_configs/ 目录（相对于本脚本的两级父目录）
_CHART_CONFIG_DIR = Path(__file__).parent.parent / "resources" / "chart_configs"
def _load_base_config(name: str) -> dict:
path = _CHART_CONFIG_DIR / f"{name}.json"
if path.exists():
with open(path, encoding="utf-8") as f:
return json.load(f)
return {}
def _deep_merge(base: dict, override: dict) -> dict:
"""递归合并两个字典，override 优先。"""
result = copy.deepcopy(base)
for k, v in override.items():
if k in result and isinstance(result[k], dict) and isinstance(v, dict):
result[k] = _deep_merge(result[k], v)
else:
result[k] = v
return result
# ── 直方图 ────────────────────────────────────────────────
def build_histogram(col: str, profile: dict, height: int = 350) -> dict[str, Any]:
"""为单个数值列生成直方图 ECharts 配置。"""
hist = profile.get("histogram", {})
labels = hist.get("labels", [])
counts = hist.get("counts", [])
base = _load_base_config("histogram")
override = {
"title": {"text": col, "textStyle": {"fontSize": 13, "fontWeight": "normal"}},
"xAxis": {"data": labels},
"series": [{"data": counts}],
}
config = _deep_merge(base, override)
config["_height"] = height
config["_col"] = col
config["_chart_type"] = "histogram"
return config
# ── 相关性热力图 ──────────────────────────────────────────
def build_heatmap(corr_result: dict, height: int = 500) -> dict[str, Any] | None:
"""生成相关性热力图 ECharts 配置。"""
if corr_result.get("skipped"):
return None
columns = corr_result["columns"]
heatmap_data = corr_result["heatmap_data"] # [[col_i, col_j, r], ...]
base = _load_base_config("heatmap")
override = {
"title": {
"text": "特征相关性矩阵",
"subtext": f"方法: {corr_result.get('method', 'pearson')}",
"textStyle": {"fontSize": 14},
},
"xAxis": {"data": columns},
"yAxis": {"data": columns},
"series": [{"data": heatmap_data}],
}
# 自适应高度（列数多时增大）
n = len(columns)
adaptive_height = max(height, n * 28 + 150)
config = _deep_merge(base, override)
config["_height"] = adaptive_height
config["_chart_type"] = "heatmap"
return config
# ── 箱线图 ────────────────────────────────────────────────
def build_boxplot(anomaly_result: dict, max_cols: int = 15, height: int = 380) -> dict[str, Any] | None:
"""为所有数值列生成箱线图（含异常点散点）ECharts 配置。"""
col_data = anomaly_result.get("columns", {})
valid = {c: v for c, v in col_data.items() if not v.get("skipped") and "boxplot_data" in v}
if not valid:
return None
# 列数过多时只取异常值最多的前 N 列
if len(valid) > max_cols:
valid = dict(
sorted(valid.items(), key=lambda x: x[1].get("anomaly_count", 0), reverse=True)[:max_cols]
)
col_names = list(valid.keys())
box_series_data = [v["boxplot_data"] for v in valid.values()]
# 散点数据：[[col_index, value], ...]
scatter_data: list[list] = []
for i, (col, v) in enumerate(valid.items()):
for ov in v.get("outlier_values", []):
scatter_data.append([i, ov])
base = _load_base_config("boxplot")
override = {
"title": {
"text": "异常值分布（箱线图）",
"subtext": f"共 {len(col_names)} 列，红点为异常值",
"textStyle": {"fontSize": 14},
},
"xAxis": {"data": col_names},
"series": [
{"data": box_series_data},
{"data": scatter_data},
],
}
config = _deep_merge(base, override)
config["_height"] = height
config["_chart_type"] = "boxplot"
return config
# ── 分类频率柱图 ──────────────────────────────────────────
def build_bar_categorical(col: str, profile: dict, top_n: int = 15, height: int = 320) -> dict[str, Any]:
"""为分类列生成频率柱图 ECharts 配置。"""
top_values = profile.get("top_values", [])[:top_n]
labels = [str(v["value"]) for v in top_values]
counts = [v["count"] for v in top_values]
config = {
"_chart_type": "bar_categorical",
"_col": col,
"_height": height,
"title": {"text": col, "textStyle": {"fontSize": 13, "fontWeight": "normal"}},
"tooltip": {"trigger": "axis", "axisPointer": {"type": "shadow"}},
"grid": {"left": "5%", "right": "5%", "top": "18%", "bottom": "15%", "containLabel": True},
"xAxis": {
"type": "category",
"data": labels,
"axisLabel": {"rotate": 30, "fontSize": 11, "overflow": "truncate", "width": 80},
},
"yAxis": {
"type": "value",
"name": "频次",
"splitLine": {"lineStyle": {"type": "dashed", "color": "#e0e0e0"}},
},
"series": [
{
"type": "bar",
"data": counts,
"barMaxWidth": 40,
"itemStyle": {
"color": "#5b8ff9",
"borderRadius": [3, 3, 0, 0],
},
}
],
}
return config
# ── 主函数：生成所有图表配置 ──────────────────────────────
def build_all_charts(
profile_result: dict,
corr_result: dict,
anomaly_result: dict,
chart_height: int = 350,
max_boxplot_cols: int = 15,
) -> dict[str, Any]:
"""
汇总生成所有图表配置。
返回：
{
"column_charts": { col: echarts_config }, # 每列一个图
"heatmap": echarts_config | None,
"boxplot": echarts_config | None,
}
"""
column_charts: dict[str, Any] = {}
columns_profile = profile_result.get("columns", {})
for col, prof in columns_profile.items():
col_type = prof.get("type")
if col_type == "numeric":
column_charts[col] = build_histogram(col, prof, height=chart_height)
elif col_type == "categorical":
if prof.get("top_values"):
column_charts[col] = build_bar_categorical(col, prof, height=chart_height)
heatmap = build_heatmap(corr_result, height=max(500, len(corr_result.get("columns", [])) * 28 + 150))
boxplot = build_boxplot(anomaly_result, max_cols=max_boxplot_cols, height=chart_height + 30)
return {
"column_charts": column_charts,
"heatmap": heatmap,
"boxplot": boxplot,
}